185 lines · 7.1 KB · Python
RawDownload
"""Plugin discovery and static serving for Hermes Web UI. Scans ~/.hermes/plugins/<name>/dashboard/ for manifest.json files,matching the official Hermes dashboard plugin format. Each plugin may have:  dashboard/    manifest.json   -- tab definition and entry point    dist/      index.js      -- plugin JS bundle (IIFE)      style.css     -- optional plugin stylesheet    plugin_api.py   -- optional backend API (not used in WebUI MVP)"""import jsonimport loggingimport osimport refrom pathlib import Path logger = logging.getLogger(__name__) # Valid dashboard-plugin name: a safe slug (it becomes a URL path component and# a settings key). Lowercase alnum + - / _, 1-64 chars, must start with a letter._VALID_PLUGIN_NAME = re.compile(r"^[a-z][a-z0-9_-]{0,63}$") # Valid tab.path: a clean same-origin absolute path. Must start with a single# '/' (NOT '//' — a leading '//' is a protocol-relative URL that would resolve# to a remote origin when assigned to iframe.src), then only safe path chars —# no quotes, whitespace, control chars, query ('?') or fragment ('#')._VALID_PLUGIN_TAB_PATH = re.compile(r"^/(?!/)[A-Za-z0-9._~/-]{0,255}$") # plugin_name -> manifest dict (as loaded from manifest.json)PLUGIN_MANIFESTS: dict[str, dict] = {} # plugin_name -> resolved static root dir_PLUGIN_STATIC_ROOTS: dict[str, Path] = {}  def _get_plugin_base() -> Path:    return Path(os.environ.get("HERMES_WEBUI_PLUGINS_DIR", str(Path.home() / ".hermes" / "plugins")))  def load_plugins() -> None:    """Scan plugin directories and load manifest.json for each dashboard plugin."""    plugin_base = _get_plugin_base()    if not plugin_base.is_dir():        logger.debug("No plugins directory at %s", plugin_base)        return     for entry in sorted(plugin_base.iterdir()):        if not entry.is_dir():            continue        manifest_path = entry / "dashboard" / "manifest.json"        if not manifest_path.is_file():            continue         try:            manifest = json.loads(manifest_path.read_text())        except Exception:            logger.exception("Failed to parse manifest for plugin %s", entry.name)            continue         name = manifest.get("name") or entry.name         # Validate the plugin name: it becomes a URL path component        # (/dashboard-plugins/<name>/...) and a settings key. Restrict to a safe        # slug so a manifest like name:"../foo" can't make the URL-space ambiguous.        if not _VALID_PLUGIN_NAME.match(str(name)):            logger.warning("Skipping plugin with invalid name %r (must match %s)", name, _VALID_PLUGIN_NAME.pattern)            continue         tab = manifest.get("tab", {})        tab_path = tab.get("path", f"/{name}")         # Validate tab.path: it's a same-origin route the plugin page is served        # at AND a value passed into client-side navigation. Require a clean        # absolute path — no quotes/control chars/query/fragment — so a hostile        # manifest can't shadow odd routes or inject via the path.        if not _VALID_PLUGIN_TAB_PATH.match(str(tab_path)):            logger.warning("Skipping plugin %s with invalid tab.path %r (must match %s)", name, tab_path, _VALID_PLUGIN_TAB_PATH.pattern)            continue         if name in PLUGIN_MANIFESTS:            logger.warning("Duplicate plugin name skipped: %s (already loaded)", name)            continue        if tab_path in (m.get("tab", {}).get("path") for m in PLUGIN_MANIFESTS.values()):            logger.warning("Plugin %s tab.path %r conflicts with another plugin; skipped", name, tab_path)            continue         PLUGIN_MANIFESTS[name] = manifest        logger.info("Loaded dashboard plugin: %s (label=%s)", name, manifest.get("label", ""))         # Pre-compute static root for fast serving (points to dashboard/)        dashboard_dir = entry / "dashboard"        if dashboard_dir.is_dir():            _PLUGIN_STATIC_ROOTS[name] = dashboard_dir.resolve()  def serve_plugin_static(plugin_name: str, rel_path: str) -> tuple[bytes, str] | None:    """    Serve a built static asset from a plugin's dashboard/dist/ (or static/) dir.     Returns (file_bytes, content_type) on success, None on not found.     Security: _PLUGIN_STATIC_ROOTS points at the plugin's whole dashboard/ dir    (the page route needs that), but the asset route must NOT expose plugin    source/config — e.g. dashboard/plugin_api.py, manifest.json, .env. So we    constrain served files to the built-asset subtrees (dist/ or static/), reject    dotfiles, and require a known static extension.    """    root = _PLUGIN_STATIC_ROOTS.get(plugin_name)    if not root:        return None     safe = (root / rel_path.lstrip("/")).resolve()    try:        safe.relative_to(root)    except ValueError:        return None  # path traversal attempt     # Only built-asset subtrees are servable (not the dashboard root itself,    # which holds plugin_api.py / manifest.json / config).    rel = safe.relative_to(root)    if not rel.parts or rel.parts[0] not in ("dist", "static"):        return None    # No dotfiles (.env, .git, etc.) anywhere in the path.    if any(part.startswith(".") for part in rel.parts):        return None     if not safe.is_file():        return None     # Allowlist of static asset extensions — refuse source/config (.py, .json,    # .toml, .env, .sh, ...) even if somehow placed under dist/.    ext = os.path.splitext(rel_path.lower())[1]    _STATIC_EXTS = {        ".js", ".css", ".html", ".png", ".jpg", ".jpeg", ".gif", ".svg",        ".ico", ".webp", ".woff", ".woff2", ".ttf", ".otf", ".map", ".txt",    }    if ext not in _STATIC_EXTS:        return None     data = safe.read_bytes()    content_type = {        ".js": "application/javascript; charset=utf-8",        ".css": "text/css; charset=utf-8",        ".html": "text/html; charset=utf-8",        ".json": "application/json; charset=utf-8",        ".png": "image/png",        ".svg": "image/svg+xml",        ".ico": "image/x-icon",    }.get(ext, "application/octet-stream")     return data, content_type  def get_plugin_metadata() -> list[dict]:    """    Return a list of plugin metadata suitable for the Settings → Plugins tab.    Each entry includes name, key, version, description, and tab info for linking.     Per-plugin enabled state is stored in settings.json under `dashboard_plugins`.    A plugin is enabled only if the user has explicitly toggled it on (default off).    """    from api.config import load_settings     plugin_settings = load_settings().get("dashboard_plugins", {})    plugins = []    for name, manifest in sorted(PLUGIN_MANIFESTS.items()):        tab = manifest.get("tab", {})        path = tab.get("path", f"/{name}")        plugins.append({            "name": manifest.get("label") or manifest.get("name") or name,            "key": name,            "version": manifest.get("version", "0.0.0"),            "description": manifest.get("description", ""),            "tab": {                "path": path,                "label": tab.get("label") or manifest.get("label") or name,            },            "enabled": bool(plugin_settings.get(name, False)),            "hooks": [],        })    return plugins