mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
Maciej Lisiewski
7062229090
pnpm 12.15.0 (native executable) installs every JS workspace from one pnpm-lock.yaml, imported from package-lock.json with the same resolved versions. website/, the WhatsApp bridge and the photon sidecar stay separate npm projects. Configuration moves from .npmrc and the root manifest into pnpm-workspace.yaml: the 14-day minimumReleaseAge and its excludes, overrides, allowBuilds (the old allowScripts set, with strictDepBuilds failing any unlisted build script), engineStrict, sideEffectsCache: false (in-place rebuilds must not write into the store) and verifyDepsBeforeRun: false (a `pnpm run` never installs). The pnpm version is pinned in pm/lock.json and in root devEngines. pmOnFail: ignore keeps that pin out of pnpm-lock.yaml: a leading pin document makes osv-scanner read zero packages. scripts/build/node-deps.mjs enforces the exact version instead, and every lock reader refuses a multi-document lock. A root preinstall refuses `npm install`; devEngines onFail is warn so npx (MCP servers, the tsc linter) keeps working inside the checkout. User-machine installs run --frozen-lockfile --trust-lockfile (release age is checked when locking, in CI, not per user). node-deps removes npm-era node_modules trees before the first pnpm install. Nix builds the store with fetchPnpmDeps and a hash file that CI checks.
1244 lines
62 KiB
Python
1244 lines
62 KiB
Python
"""Dashboard UI assets: SPA mount, theme normalisation/bootstrap CSS, dashboard-plugin discovery and the plugins-hub merge.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import json
|
|
import os
|
|
import threading
|
|
import time
|
|
from contextlib import contextmanager
|
|
import hermes_yaml as yaml
|
|
from fastapi import Depends, FastAPI, HTTPException, Request, WebSocket, WebSocketException
|
|
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, Response
|
|
from fastapi.staticfiles import StaticFiles
|
|
from starlette import status
|
|
from starlette._utils import get_route_path
|
|
from starlette.requests import HTTPConnection
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from hermes_cli.config import cfg_get, get_process_hermes_home
|
|
from utils import env_var_enabled
|
|
|
|
# Serve assets with explicit MIME types: the host's map (on Windows the
|
|
# registry can map .js -> text/plain) must not decide the Content-Type of
|
|
# dashboard bundles (#28987). Runs before any StaticFiles/FileResponse below
|
|
# serves a request, and is idempotent.
|
|
from hermes_cli.web_asset_mime_types import normalize_web_asset_mime_types
|
|
|
|
normalize_web_asset_mime_types()
|
|
|
|
# Same logger the code used before extraction (record parity).
|
|
_log = logging.getLogger("hermes_cli.web_server")
|
|
|
|
|
|
def _normalise_prefix(raw: Optional[str]) -> str:
|
|
"""Normalise an X-Forwarded-Prefix header value (single source of truth lives in
|
|
``hermes_cli.dashboard_auth.prefix`` so gate, OAuth, cookies and SPA mount agree)."""
|
|
from hermes_cli.dashboard_auth.prefix import normalise_prefix
|
|
return normalise_prefix(raw)
|
|
|
|
|
|
def _layer_hex(palette: dict[str, Any], key: str, default: str) -> str:
|
|
layer = palette.get(key) or {}
|
|
return layer.get("hex", default) if isinstance(layer, dict) else default
|
|
|
|
|
|
def _render_active_theme_bootstrap_css() -> str:
|
|
"""Critical-CSS ``<style>`` shim for the active *user* theme, so the first paint uses the
|
|
target palette instead of flashing the bundle's default Hermes Teal until
|
|
``ThemeProvider.applyTheme()`` runs. Built-in themes return "" (the bundle owns them).
|
|
|
|
Variable names MUST match what the bundle consumes (``layerVars()`` /
|
|
``typographyVars()`` in ``web/src/themes/context.tsx``). The ``html,body`` rule
|
|
references the variables rather than literals so runtime theme switches stay live:
|
|
``applyTheme()`` writes inline styles on ``documentElement`` which outrank this block.
|
|
"""
|
|
from hermes_cli.config import load_config
|
|
try:
|
|
active = cfg_get(load_config(), "dashboard", "theme", default="default")
|
|
if not active or not isinstance(active, str):
|
|
return ""
|
|
if any(b["name"] == active for b in _BUILTIN_DASHBOARD_THEMES):
|
|
return ""
|
|
for theme in _discover_user_themes():
|
|
if theme.get("name") != active:
|
|
continue
|
|
palette = theme.get("palette") or {}
|
|
typo = theme.get("typography") or {}
|
|
font_sans = typo.get("fontSans") or _THEME_DEFAULT_TYPOGRAPHY["fontSans"]
|
|
base_size = typo.get("baseSize") or _THEME_DEFAULT_TYPOGRAPHY["baseSize"]
|
|
|
|
def _esc(s: str) -> str: # defensive ``</style>`` escape
|
|
return str(s).replace("</", "<\\/")
|
|
return (
|
|
'<style id="hermes-theme-bootstrap">'
|
|
":root{"
|
|
f"--background-base:{_esc(_layer_hex(palette, 'background', '#0a0a0a'))};"
|
|
f"--midground-base:{_esc(_layer_hex(palette, 'midground', '#e5e5e5'))};"
|
|
f"--theme-font-sans:{_esc(font_sans)};"
|
|
f"--theme-base-size:{_esc(base_size)};"
|
|
"}"
|
|
"html,body{background-color:var(--background-base);"
|
|
"color:var(--midground-base);"
|
|
"font-family:var(--theme-font-sans);"
|
|
"font-size:var(--theme-base-size);}"
|
|
"</style>"
|
|
)
|
|
return ""
|
|
except Exception:
|
|
_log.debug("theme bootstrap render failed", exc_info=True)
|
|
return ""
|
|
|
|
|
|
# Hashed bundle assets are immutable by construction (content hash in the filename; index.html
|
|
# is served ``no-store`` and always references the current hashes).
|
|
_IMMUTABLE_ASSET_CACHE_CONTROL = "public, max-age=31536000, immutable"
|
|
_NO_STORE = {"Cache-Control": "no-store, no-cache, must-revalidate"}
|
|
_HEADLESS_MSG = (
|
|
"Headless backend (hermes serve): web UI disabled — use "
|
|
"`hermes dashboard` for the browser UI."
|
|
)
|
|
|
|
|
|
def mount_spa(application: FastAPI):
|
|
"""Mount the built SPA; unmatched paths fall back to index.html for client-side routing.
|
|
|
|
Dashboard injects its session token into index.html; Webapp instead receives
|
|
the same credential through an operator-only launch fragment. Behind a path-prefix reverse
|
|
proxy (``X-Forwarded-Prefix: /hermes``) the served index.html is rewritten so absolute
|
|
asset URLs and the runtime ``__HERMES_BASE_PATH__`` honour that prefix without a rebuild.
|
|
|
|
A missing WEB_DIST is deliberately NOT a mount-time terminal state: every route copes
|
|
with a missing dist per-request (404 JSON / ``check_dir=False``), so a long-lived
|
|
``--skip-build`` process recovers the moment a build appears on disk — no restart.
|
|
The dist is also chosen per request: this mount runs at import, and a host that serves
|
|
another renderer (the Webapp's ``dist-webapp``) records it on ``application.state.web_dist``
|
|
later.
|
|
"""
|
|
from hermes_cli.web_server import WEB_DIST, _DASHBOARD_EMBEDDED_CHAT_ENABLED
|
|
from hermes_cli.web_deps import _server
|
|
from hermes_cli.web_server_surface import policy
|
|
from hermes_cli.web_server_static_assets import is_static_asset_path, public_font_file
|
|
|
|
# `hermes serve` is the headless backend: it must NEVER serve the browser SPA, even if a
|
|
# dist is lying around, so only the JSON-RPC/WS/API surface is reachable.
|
|
if os.environ.get("HERMES_SERVE_HEADLESS") == "1":
|
|
|
|
@application.get("/{full_path:path}")
|
|
async def no_frontend(full_path: str):
|
|
# Desktop token handshake: the Electron shell boots by fetching `/` and reading
|
|
# ``window.__HERMES_SESSION_TOKEN__`` for /api/ws auth. When headless 404'd every
|
|
# path, a renderer whose spawn token no longer matched (e.g. after `hermes update`)
|
|
# white-screened. Serve a token-only page at the exact root, but ONLY when the auth
|
|
# gate is off: on a gated serve the token must never be readable without auth.
|
|
# See #94227, #95575.
|
|
gated = bool(getattr(application.state, "auth_required", False))
|
|
withheld = gated or bool(getattr(_server().app.state, "withhold_session_token", False))
|
|
if full_path == "" and not withheld:
|
|
return HTMLResponse(
|
|
"<!doctype html><html><head><script>"
|
|
f"window.__HERMES_SESSION_TOKEN__={json.dumps(_server()._SESSION_TOKEN)};"
|
|
"window.__HERMES_AUTH_REQUIRED__=false;"
|
|
f"</script></head><body>{_HEADLESS_MSG}</body></html>",
|
|
headers=_NO_STORE,
|
|
)
|
|
return JSONResponse({"error": _HEADLESS_MSG}, status_code=404)
|
|
return
|
|
|
|
# A missing WEB_DIST is deliberately NOT a mount-time terminal state (#82614): a long-lived `hermes
|
|
# dashboard --skip-build` process that survives a `git pull` (or starts before the first build) used to
|
|
# install a permanent no_frontend catch-all here and could never recover — every route answered 404
|
|
# "Frontend not built" until the process was restarted, even after `npm run build` completed. The SPA
|
|
# routes below all cope with a missing dist per-request (`_serve_index` returns the same 404 JSON when
|
|
# index.html is unreadable; the asset mounts use check_dir=False and 404 on missing files), so mounting
|
|
# them unconditionally makes the dashboard recover the moment a build appears on disk — no restart
|
|
# needed.
|
|
def web_dist() -> Path:
|
|
return getattr(application.state, "web_dist", None) or WEB_DIST
|
|
|
|
def _serve_index(request: Request, prefix: str = ""):
|
|
"""index.html with the session token + base-path injected.
|
|
|
|
When the OAuth auth gate is active (``app.state.auth_required``), the legacy
|
|
``_SESSION_TOKEN`` is NOT injected — the SPA reads identity from ``/api/auth/me`` over
|
|
cookie auth; ``__HERMES_AUTH_REQUIRED__`` tells it which scheme to use for /api/pty
|
|
and /api/ws (ticket vs token).
|
|
"""
|
|
try:
|
|
html = (web_dist() / "index.html").read_text(encoding="utf-8-sig")
|
|
except OSError:
|
|
# Partial build / wiped dist / permissions: same JSON 404 as a fully-missing dist. The
|
|
# Webapp's bundle is built by `hermes webapp`, not by the dashboard's web project.
|
|
webapp = getattr(application.state, "ui_surface", "dashboard") == "webapp"
|
|
missing = ("The Webapp is not built. Run: hermes webapp" if webapp
|
|
else "Frontend not built. Run: pnpm --dir web run build")
|
|
return JSONResponse({"error": missing}, status_code=404)
|
|
if getattr(application.state, "ui_surface", "dashboard") == "webapp":
|
|
# A bundle `hermes update` could not rebuild was built for the code before the update:
|
|
# refuse it until `hermes webapp` rebuilds it.
|
|
from gateway.webapp_listener import bundle_stale_reason
|
|
stale = bundle_stale_reason(web_dist())
|
|
if stale:
|
|
return JSONResponse({"error": stale}, status_code=503, headers=_NO_STORE)
|
|
chat_js = "true" if _DASHBOARD_EMBEDDED_CHAT_ENABLED else "false"
|
|
gated = bool(getattr(application.state, "auth_required", False))
|
|
# Only a host that records its surface (the gateway's Webapp) names it to the page.
|
|
surface = getattr(application.state, "ui_surface", None)
|
|
surface_js = "" if surface is None else f"window.__HERMES_UI_SURFACE__={json.dumps(surface)};"
|
|
# Webapp grants host files and shell.exec as well as PTYs: its existing
|
|
# session token must come from the operator, never anonymous HTML. The
|
|
# gateway-hosted listener (gateway/run_api.py) never publishes it either:
|
|
# local clients use tickets.
|
|
withheld = (gated or bool(getattr(_server().app.state, "withhold_session_token", False))
|
|
or not policy(application.state).html_token)
|
|
token_js = "" if withheld else f'window.__HERMES_SESSION_TOKEN__="{_server()._SESSION_TOKEN}";'
|
|
# Launcher-preselected profile (``--open-profile``): the SPA's fallback scope when the URL
|
|
# omits ``?profile=`` (#73085). ``</`` escaped so a hostile name cannot close the script tag.
|
|
initial_profile_js = json.dumps(str(getattr(application.state, "initial_profile", "") or "")).replace("</", "<\\/")
|
|
# This backend's OWN profile name (empty when it cannot be named unambiguously). The SPA
|
|
# falls back to it when neither the URL nor --open-profile names one, so requests carry an
|
|
# explicit scope from the first paint: destructive routes 400 on an unnamed profile as soon
|
|
# as the host serves more than one, and the switcher shows the same profile it writes. The
|
|
# Webapp reads it as the profile its gateway socket and REST answer for when a request names
|
|
# none, so a tab on any other profile names that profile on every request.
|
|
from hermes_cli.web_server_profiles import serving_profile_name as _serving_profile_name
|
|
serving_profile_js = json.dumps(_serving_profile_name()).replace("</", "<\\/")
|
|
bootstrap_script = (
|
|
f"<script>{token_js}"
|
|
f"window.__HERMES_DASHBOARD_EMBEDDED_CHAT__={chat_js};"
|
|
f'window.__HERMES_BASE_PATH__="{prefix}";'
|
|
f"window.__HERMES_AUTH_REQUIRED__={'true' if gated else 'false'};"
|
|
f"window.__HERMES_INITIAL_PROFILE__={initial_profile_js};"
|
|
f"{surface_js}"
|
|
f"window.__HERMES_DASHBOARD_PROFILE__={serving_profile_js};"
|
|
f"</script>"
|
|
)
|
|
if prefix:
|
|
# Rewrite absolute asset URLs baked into the Vite build to go through the proxy.
|
|
for attr in ('href="/assets/', 'src="/assets/', 'href="/favicon.ico"', 'href="/fonts/',
|
|
'href="/ds-assets/', 'src="/ds-assets/'):
|
|
html = html.replace(attr, attr.replace('"/', f'"{prefix}/', 1))
|
|
theme_bootstrap = _render_active_theme_bootstrap_css()
|
|
if theme_bootstrap:
|
|
html = html.replace("</head>", f"{theme_bootstrap}</head>", 1)
|
|
html = html.replace("</head>", f"{bootstrap_script}</head>", 1)
|
|
from hermes_cli.webapp_page_policy import page_headers
|
|
return HTMLResponse(html, headers=page_headers(request, html, _NO_STORE))
|
|
|
|
# Built CSS contains absolute ``url(/fonts/...)`` / ``url(/ds-assets/...)`` references that
|
|
# browsers resolve against the document origin — wrong under a proxy prefix. Intercept CSS
|
|
# BEFORE the StaticFiles mount and rewrite when a prefix is in play.
|
|
@application.get("/assets/{filename}.css")
|
|
async def serve_css(filename: str, request: Request):
|
|
dist = web_dist()
|
|
css_path = dist / "assets" / f"{filename}.css"
|
|
if not css_path.is_file() or not css_path.resolve().is_relative_to(dist.resolve()):
|
|
return JSONResponse({"error": "not found"}, status_code=404)
|
|
prefix = _normalise_prefix(request.headers.get("x-forwarded-prefix"))
|
|
css = css_path.read_text(encoding="utf-8-sig")
|
|
if prefix:
|
|
for asset_dir in ("/fonts/", "/fonts-terminal/", "/ds-assets/", "/assets/"):
|
|
for quote in ("", '"', "'"):
|
|
css = css.replace(f"url({quote}{asset_dir}", f"url({quote}{prefix}{asset_dir}")
|
|
return Response(
|
|
content=css, media_type="text/css", headers={"Cache-Control": _IMMUTABLE_ASSET_CACHE_CONTROL}
|
|
)
|
|
|
|
class _ImmutableAssetFiles(StaticFiles):
|
|
"""StaticFiles that marks hashed bundle assets immutable so reloads skip revalidation."""
|
|
|
|
async def get_response(self, path: str, scope):
|
|
response = await super().get_response(path, scope)
|
|
if response.status_code == 200:
|
|
response.headers["Cache-Control"] = _IMMUTABLE_ASSET_CACHE_CONTROL
|
|
return response
|
|
|
|
asset_files: dict[Path, StaticFiles] = {}
|
|
|
|
async def serve_assets(scope, receive, send):
|
|
# check_dir=False: the dist may not exist yet; StaticFiles 404s per-request until it does.
|
|
directory = web_dist() / "assets"
|
|
if directory not in asset_files:
|
|
asset_files[directory] = _ImmutableAssetFiles(directory=directory, check_dir=False)
|
|
await asset_files[directory](scope, receive, send)
|
|
|
|
from hermes_cli.webapp_previews import mount_preview_assets
|
|
mount_preview_assets(application)
|
|
application.mount("/assets", serve_assets, name="assets")
|
|
|
|
@application.get("/{full_path:path}")
|
|
async def serve_spa(full_path: str, request: Request):
|
|
prefix = _normalise_prefix(request.headers.get("x-forwarded-prefix"))
|
|
# An unmatched /api/* path is a missing endpoint, not a client-side route: return a
|
|
# real 404 JSON instead of index.html (which breaks JSON clients with a SyntaxError).
|
|
if full_path == "api" or full_path.startswith("api/"):
|
|
return JSONResponse({"detail": f"No such API endpoint: /{full_path}"}, status_code=404)
|
|
dist = web_dist()
|
|
file_path = dist / full_path
|
|
index = dist / "index.html"
|
|
if (not policy(application.state).html_token and file_path.is_file() and index.is_file()
|
|
and file_path.samefile(index)):
|
|
# No spelling of the entry document may bypass bootstrap or policy.
|
|
return _serve_index(request, prefix)
|
|
# Prevent path traversal via url-encoded sequences (%2e%2e/)
|
|
if (
|
|
full_path
|
|
and file_path.resolve().is_relative_to(dist.resolve())
|
|
and file_path.exists()
|
|
and file_path.is_file()
|
|
):
|
|
return FileResponse(file_path)
|
|
if is_static_asset_path(full_path):
|
|
font = public_font_file(dist, WEB_DIST, full_path)
|
|
if font is not None:
|
|
return FileResponse(font)
|
|
return JSONResponse({"error": "not found"}, status_code=404)
|
|
return _serve_index(request, prefix)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dashboard themes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Built-in themes — label + description only; colors live in web/src/themes/presets.ts.
|
|
_BUILTIN_DASHBOARD_THEMES = [
|
|
{"name": "default", "label": "Hermes Teal", "description": "Classic dark teal — the canonical Hermes look"},
|
|
{"name": "default-large", "label": "Hermes Teal (Large)", "description": "Hermes Teal with bigger fonts and roomier spacing"},
|
|
{"name": "nous-blue", "label": "Nous Blue", "description": "Light mode — vivid Nous-blue accents on cream canvas"},
|
|
{"name": "midnight", "label": "Midnight", "description": "Deep blue-violet with cool accents"},
|
|
{"name": "ember", "label": "Ember", "description": "Warm crimson and bronze — forge vibes"},
|
|
{"name": "mono", "label": "Mono", "description": "Clean grayscale — minimal and focused"},
|
|
{"name": "cyberpunk", "label": "Cyberpunk", "description": "Neon green on black — matrix terminal"},
|
|
{"name": "rose", "label": "Rosé", "description": "Soft pink and warm ivory — easy on the eyes"},
|
|
]
|
|
|
|
|
|
def _parse_theme_layer(value: Any, default_hex: str, default_alpha: float = 1.0) -> Optional[dict[str, Any]]:
|
|
"""Normalise a theme layer spec (bare hex shorthand or ``{hex, alpha}`` dict); ``None`` on
|
|
garbage so the caller falls back to a built-in default."""
|
|
if value is None:
|
|
return {"hex": default_hex, "alpha": default_alpha}
|
|
if isinstance(value, str):
|
|
return {"hex": value, "alpha": default_alpha}
|
|
if not isinstance(value, dict):
|
|
return None
|
|
hex_val = value.get("hex", default_hex)
|
|
if not isinstance(hex_val, str):
|
|
return None
|
|
try:
|
|
alpha_f = float(value.get("alpha", default_alpha))
|
|
except (TypeError, ValueError):
|
|
alpha_f = default_alpha
|
|
return {"hex": hex_val, "alpha": max(0.0, min(1.0, alpha_f))}
|
|
|
|
|
|
_THEME_DEFAULT_TYPOGRAPHY: dict[str, str] = {
|
|
"fontSans": 'system-ui, -apple-system, "Segoe UI", Roboto, "Helvetica Neue", Arial, sans-serif',
|
|
"fontMono": 'ui-monospace, "SF Mono", "Cascadia Mono", Menlo, Consolas, monospace',
|
|
"baseSize": "15px",
|
|
"lineHeight": "1.55",
|
|
"letterSpacing": "0",
|
|
}
|
|
_THEME_DEFAULT_LAYOUT: dict[str, str] = {
|
|
"radius": "0.5rem", "density": "comfortable"
|
|
}
|
|
_THEME_OVERRIDE_KEYS = {
|
|
"card", "cardForeground", "popover", "popoverForeground",
|
|
"primary", "primaryForeground", "secondary", "secondaryForeground",
|
|
"muted", "mutedForeground", "accent", "accentForeground",
|
|
"destructive", "destructiveForeground", "success", "warning",
|
|
"border", "input", "ring",
|
|
}
|
|
|
|
# Named asset slots; other keys under ``assets.custom`` become ``--theme-asset-custom-<key>``.
|
|
_THEME_NAMED_ASSET_KEYS = {"bg", "hero", "logo", "crest", "sidebar", "header"}
|
|
|
|
# Component-style buckets: each camelCase property under a bucket emits
|
|
# ``--component-<bucket>-<kebab-property>`` on :root, consumed by shell components.
|
|
_THEME_COMPONENT_BUCKETS = {
|
|
"card", "header", "footer", "sidebar", "tab", "progress", "badge", "backdrop", "page"
|
|
}
|
|
_THEME_LAYOUT_VARIANTS = {"standard", "cockpit", "tiled"}
|
|
|
|
# customCSS cap so an oversized theme YAML can't blow up the payload or <style> tag.
|
|
_THEME_CUSTOM_CSS_MAX = 32 * 1024
|
|
|
|
|
|
def _dict_field(data: dict[str, Any], key: str) -> dict[str, Any]:
|
|
value = data.get(key)
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def _nonempty_str(value: Any) -> bool:
|
|
return isinstance(value, str) and bool(value.strip())
|
|
|
|
|
|
def _css_ident(key: Any) -> bool:
|
|
return isinstance(key, str) and key.replace("-", "").replace("_", "").isalnum()
|
|
|
|
|
|
def _normalise_theme_definition(data: dict[str, Any]) -> Optional[dict[str, Any]]:
|
|
"""Normalise a user theme YAML into the wire format ``ThemeProvider`` expects; ``None`` if
|
|
unusable. Accepts the full schema and a loose form (top-level ``colors``, bare hex).
|
|
|
|
customCSS is clipped but intentionally NOT sanitised — themes are user-authored YAML in
|
|
~/.hermes/, the same trust level as config.yaml. Empty asset values are dropped so a
|
|
theme can explicitly clear a slot.
|
|
"""
|
|
if not isinstance(data, dict):
|
|
return None
|
|
name = data.get("name")
|
|
if not _nonempty_str(name):
|
|
return None
|
|
|
|
palette_src = _dict_field(data, "palette")
|
|
colors_src = _dict_field(data, "colors")
|
|
|
|
def _layer(key: str, default_hex: str, default_alpha: float = 1.0) -> dict[str, Any]:
|
|
parsed = _parse_theme_layer(palette_src.get(key, colors_src.get(key)), default_hex, default_alpha)
|
|
return parsed if parsed is not None else {"hex": default_hex, "alpha": default_alpha}
|
|
|
|
raw_noise = palette_src.get("noiseOpacity", data.get("noiseOpacity"))
|
|
try:
|
|
noise = float(raw_noise) if raw_noise is not None else 1.0
|
|
except (TypeError, ValueError):
|
|
noise = 1.0
|
|
palette = {
|
|
"background": _layer("background", "#041c1c", 1.0),
|
|
"midground": _layer("midground", "#ffe6cb", 1.0),
|
|
"foreground": _layer("foreground", "#ffffff", 0.0),
|
|
"warmGlow": palette_src.get("warmGlow") or data.get("warmGlow") or "rgba(255, 189, 56, 0.35)",
|
|
"noiseOpacity": noise,
|
|
}
|
|
|
|
typo_src = _dict_field(data, "typography")
|
|
typography = dict(_THEME_DEFAULT_TYPOGRAPHY)
|
|
for key in ("fontSans", "fontMono", "fontDisplay", "fontUrl", "baseSize", "lineHeight", "letterSpacing"):
|
|
if _nonempty_str(typo_src.get(key)):
|
|
typography[key] = typo_src[key]
|
|
|
|
layout_src = _dict_field(data, "layout")
|
|
layout = dict(_THEME_DEFAULT_LAYOUT)
|
|
if _nonempty_str(layout_src.get("radius")):
|
|
layout["radius"] = layout_src["radius"]
|
|
density = layout_src.get("density")
|
|
if isinstance(density, str) and density in {"compact", "comfortable", "spacious"}:
|
|
layout["density"] = density
|
|
|
|
color_overrides = {
|
|
k: v for k, v in _dict_field(data, "colorOverrides").items()
|
|
if k in _THEME_OVERRIDE_KEYS and _nonempty_str(v)
|
|
}
|
|
|
|
assets_src = _dict_field(data, "assets")
|
|
assets_out: dict[str, Any] = {k: assets_src[k] for k in _THEME_NAMED_ASSET_KEYS if _nonempty_str(assets_src.get(k))}
|
|
custom_assets = {k: v for k, v in _dict_field(assets_src, "custom").items() if _css_ident(k) and _nonempty_str(v)}
|
|
if custom_assets:
|
|
assets_out["custom"] = custom_assets
|
|
|
|
custom_css_val = data.get("customCSS")
|
|
custom_css = custom_css_val[:_THEME_CUSTOM_CSS_MAX] if _nonempty_str(custom_css_val) else None
|
|
|
|
component_styles: dict[str, dict[str, str]] = {}
|
|
for bucket, props in _dict_field(data, "componentStyles").items():
|
|
if bucket not in _THEME_COMPONENT_BUCKETS or not isinstance(props, dict):
|
|
continue
|
|
clean = {
|
|
prop: str(value) for prop, value in props.items()
|
|
if _css_ident(prop) and isinstance(value, (str, int, float)) and str(value).strip()
|
|
}
|
|
if clean:
|
|
component_styles[bucket] = clean
|
|
|
|
layout_variant = data.get("layoutVariant")
|
|
if not (isinstance(layout_variant, str) and layout_variant in _THEME_LAYOUT_VARIANTS):
|
|
layout_variant = "standard"
|
|
|
|
result: dict[str, Any] = {
|
|
"name": name,
|
|
"label": data.get("label") or name,
|
|
"description": data.get("description", ""),
|
|
"palette": palette,
|
|
"typography": typography,
|
|
"layout": layout,
|
|
"layoutVariant": layout_variant,
|
|
}
|
|
if color_overrides:
|
|
result["colorOverrides"] = color_overrides
|
|
if assets_out:
|
|
result["assets"] = assets_out
|
|
if custom_css is not None:
|
|
result["customCSS"] = custom_css
|
|
if component_styles:
|
|
result["componentStyles"] = component_styles
|
|
return result
|
|
|
|
|
|
def _discover_user_themes() -> list:
|
|
"""Fully-normalised user themes from ``<launch home>/dashboard-themes/*.yaml``.
|
|
|
|
Uses the process launch home, not ``get_hermes_home()``, so a transient profile override
|
|
from embedded chat does not hide themes under the server's own ``HERMES_HOME``.
|
|
"""
|
|
themes_dir = get_process_hermes_home() / "dashboard-themes"
|
|
if not themes_dir.is_dir():
|
|
return []
|
|
result = []
|
|
for f in sorted(themes_dir.glob("*.yaml")):
|
|
try:
|
|
data = yaml.safe_load(f.read_text(encoding="utf-8-sig"))
|
|
except Exception:
|
|
continue
|
|
normalised = _normalise_theme_definition(data)
|
|
if normalised is not None:
|
|
result.append(normalised)
|
|
return result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dashboard plugin system
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _safe_plugin_api_relpath(api_field: Any, *, dashboard_dir: Path) -> Optional[str]:
|
|
"""Validate the manifest's ``api`` field (later imported as a Python module — arbitrary
|
|
code execution by design).
|
|
|
|
An absolute path would swallow the plugin dir (``Path('safe') / '/tmp/evil.py'`` ->
|
|
``/tmp/evil.py``) and ``../`` could climb out of it (GHSA-5qr3-c538-wm9j). Returns the
|
|
original string when the resolved path stays under ``dashboard_dir``, else ``None`` so
|
|
the plugin still loads its static JS/CSS but its backend ``api`` is rejected.
|
|
|
|
The web server later imports this file as a Python module via ``plugin_isolation.import_user_code``
|
|
(arbitrary code execution by design — that's how plugins extend the backend). Pre-#29156 the field was
|
|
used as-is, which meant:
|
|
"""
|
|
if not isinstance(api_field, str) or not api_field.strip():
|
|
return None
|
|
candidate = Path(api_field)
|
|
if candidate.is_absolute():
|
|
return None
|
|
try:
|
|
(dashboard_dir / candidate).resolve().relative_to(dashboard_dir.resolve())
|
|
except (OSError, RuntimeError, ValueError):
|
|
return None
|
|
return api_field
|
|
|
|
|
|
def _dashboard_plugin_search_dirs() -> list[tuple]:
|
|
"""``(root, source)`` pairs to scan, in priority order (first name wins).
|
|
|
|
User dashboard plugins are a dashboard-owned asset (like theme YAML): resolved from the
|
|
process launch home so they don't vanish when a request is scoped to another profile.
|
|
When the process itself is profile-scoped (``HERMES_HOME=<root>/profiles/<name>``) the
|
|
launch home has no ``plugins/`` — user plugins live in the hermes root — so the default
|
|
root is scanned too; profile-local plugins stay authoritative over same-named root ones.
|
|
The project source is gated on shared truthy semantics (``1``/``true``/``yes``/``on``):
|
|
a bare non-empty check let ``=0``/``=false`` silently enable it (GHSA-5qr3-c538-wm9j).
|
|
"""
|
|
from hermes_cli.plugins import get_bundled_plugins_dir
|
|
from hermes_constants import get_default_hermes_root
|
|
|
|
bundled_root = get_bundled_plugins_dir()
|
|
# User dashboard plugins are a dashboard-owned asset (same category as theme YAML): resolve them from
|
|
# the process launch home so they don't vanish when a request is scoped to another profile via a
|
|
# context-local HERMES_HOME override (e.g. embedded /chat under --open-profile). #87197: when the
|
|
# process itself is profile-scoped (``--profile <name>`` sets ``HERMES_HOME=<root>/profiles/<name>``),
|
|
# the launch home is the profile directory, which has no ``plugins/`` — user plugins are installed in
|
|
# the hermes root (``~/.hermes/plugins``). Scan the default root as well (``get_default_hermes_root()``
|
|
# unwraps ``<root>/profiles/<name>`` → ``<root>`` and returns a custom ``HERMES_HOME`` unchanged when it
|
|
# *is* the root), mirroring how ``hermes_cli.plugins`` resolves plugin install locations. The
|
|
# ``seen_names`` dedupe below keeps profile-local plugins (if any) authoritative over same-named root
|
|
# plugins.
|
|
user_plugin_roots = [get_process_hermes_home() / "plugins"]
|
|
root_plugins = get_default_hermes_root() / "plugins"
|
|
if root_plugins.resolve(strict=False) != user_plugin_roots[0].resolve(strict=False):
|
|
user_plugin_roots.append(root_plugins)
|
|
search_dirs = [(d, "user") for d in user_plugin_roots]
|
|
search_dirs += [(bundled_root / "memory", "bundled"), (bundled_root, "bundled")]
|
|
# GHSA-5qr3-c538-wm9j (#29156): the previous ``os.environ.get(...)`` check treated *any* non-empty
|
|
# string as truthy, so ``=0``, ``=false``, and ``=no`` — all of which the agent loader and operators
|
|
# correctly read as "disabled" — silently *enabled* the untrusted project source in the web server.
|
|
# Combined with the absolute-path RCE primitive on the manifest's ``api`` field (now patched below),
|
|
# this turned the opt-in into a sticky always-on switch. Use the shared truthy semantics (``1`` /
|
|
# ``true`` / ``yes`` / ``on``) so the gate matches ``hermes_cli/plugins.py`` and the documented user
|
|
# contract.
|
|
if env_var_enabled("HERMES_ENABLE_PROJECT_PLUGINS"):
|
|
search_dirs.append((Path.cwd() / ".hermes" / "plugins", "project"))
|
|
return search_dirs
|
|
|
|
|
|
def _dashboard_plugin_entry(data: dict[str, Any], name: str, dashboard_dir: Path, source: str) -> dict[str, Any]:
|
|
# Tab options: ``path`` + ``position`` for a new tab, optional ``override`` to replace a
|
|
# built-in route, and ``hidden`` to register component/slots without adding a tab.
|
|
raw_tab = data.get("tab", {}) if isinstance(data.get("tab"), dict) else {}
|
|
tab_info = {"path": raw_tab.get("path", f"/{name}"), "position": raw_tab.get("position", "end")}
|
|
override_path = raw_tab.get("override")
|
|
if isinstance(override_path, str) and override_path.startswith("/"):
|
|
tab_info["override"] = override_path
|
|
if bool(raw_tab.get("hidden")):
|
|
tab_info["hidden"] = True
|
|
# Slots the plugin populates via ``window.registerSlot(pluginName, slotName, Component)``.
|
|
slots_src = data.get("slots")
|
|
slots = [s for s in slots_src if isinstance(s, str) and s] if isinstance(slots_src, list) else []
|
|
# Validate ``api`` at discovery time so the cached value is already safe for the importer.
|
|
raw_api = data.get("api")
|
|
safe_api = _safe_plugin_api_relpath(raw_api, dashboard_dir=dashboard_dir)
|
|
if raw_api and safe_api is None:
|
|
_log.warning(
|
|
"Plugin %s: refusing unsafe api path %r (must be a "
|
|
"relative file inside the plugin's dashboard/ "
|
|
"directory); backend routes from this plugin will "
|
|
"not be mounted",
|
|
name, raw_api,
|
|
)
|
|
return {
|
|
"name": name,
|
|
"label": data.get("label", name),
|
|
"description": data.get("description", ""),
|
|
"icon": data.get("icon", "Puzzle"),
|
|
"version": data.get("version", "0.0.0"),
|
|
"tab": tab_info,
|
|
"slots": slots,
|
|
"entry": data.get("entry", "dist/index.js"),
|
|
"css": data.get("css"),
|
|
"has_api": bool(safe_api),
|
|
"source": source,
|
|
"_dir": str(dashboard_dir),
|
|
"_api_file": safe_api,
|
|
}
|
|
|
|
|
|
def _discover_dashboard_plugins(search_dirs: Optional[list] = None) -> list:
|
|
"""Scan ``<plugins root>/*/dashboard/manifest.json`` across user, bundled and (opt-in)
|
|
project plugin sources — same three sources as ``hermes_cli.plugins`` — or ``search_dirs``."""
|
|
plugins = []
|
|
seen_names: set = set()
|
|
for plugins_root, source in (_dashboard_plugin_search_dirs() if search_dirs is None else search_dirs):
|
|
try:
|
|
if not plugins_root.is_dir():
|
|
continue
|
|
with os.scandir(plugins_root) as scan:
|
|
children = sorted((Path(e.path) for e in scan), key=lambda p: p.name)
|
|
except OSError as exc:
|
|
_log.warning("Skipping unreadable dashboard plugin root %s: %s", plugins_root, exc)
|
|
continue
|
|
for child in children:
|
|
manifest_file = child / "dashboard" / "manifest.json"
|
|
try:
|
|
if not child.is_dir() or not manifest_file.exists():
|
|
continue
|
|
data = json.loads(manifest_file.read_text(encoding="utf-8-sig"))
|
|
name = data.get("name", child.name)
|
|
if name in seen_names:
|
|
continue
|
|
seen_names.add(name)
|
|
plugins.append(_dashboard_plugin_entry(data, name, child / "dashboard", source))
|
|
except OSError as exc:
|
|
_log.warning("Skipping unreadable dashboard plugin %s: %s", manifest_file, exc)
|
|
continue
|
|
except Exception as exc:
|
|
_log.warning("Bad dashboard plugin manifest %s: %s", manifest_file, exc)
|
|
continue
|
|
return plugins
|
|
|
|
|
|
def _strip_dashboard_manifest(p: dict[str, Any]) -> dict[str, Any]:
|
|
return {k: v for k, v in p.items() if not k.startswith("_")}
|
|
|
|
|
|
_PLUGINS_HUB_CACHE_TTL_SECONDS = 5.0
|
|
_plugins_hub_cache: dict[str, dict[str, Any]] = {}
|
|
_plugins_hub_cache_expires_at: dict[str, float] = {}
|
|
_plugins_hub_cache_lock = threading.Lock()
|
|
|
|
|
|
def _invalidate_plugins_hub_cache() -> None:
|
|
with _plugins_hub_cache_lock:
|
|
_plugins_hub_cache.clear()
|
|
_plugins_hub_cache_expires_at.clear()
|
|
|
|
|
|
_plugins_hub_probe_inflight: set = set()
|
|
_plugins_hub_probe_lock = threading.Lock()
|
|
|
|
|
|
def _schedule_check_fn_probe(fn) -> Optional[threading.Thread]:
|
|
"""Warm a cold ``check_fn`` verdict off the request path.
|
|
|
|
The hub read path only consumes cached availability; the only other warmer is the
|
|
tool-schema build, which a dashboard-only session never runs — so a cold cache would
|
|
report ``auth_required=False`` forever. Daemon-thread probe, deduplicated per function;
|
|
the short hub TTL surfaces the verdict on the next fetch. Returns ``None`` when a probe
|
|
for *fn* is already in flight.
|
|
"""
|
|
with _plugins_hub_probe_lock:
|
|
if fn in _plugins_hub_probe_inflight:
|
|
return None
|
|
_plugins_hub_probe_inflight.add(fn)
|
|
|
|
def _probe():
|
|
try:
|
|
from tools.registry import _check_fn_cached
|
|
|
|
_check_fn_cached(fn)
|
|
except Exception:
|
|
pass
|
|
finally:
|
|
with _plugins_hub_probe_lock:
|
|
_plugins_hub_probe_inflight.discard(fn)
|
|
|
|
thread = threading.Thread(target=_probe, name="plugins-hub-checkfn-probe", daemon=True)
|
|
thread.start()
|
|
return thread
|
|
|
|
|
|
def _plugin_login_command(name: str) -> str:
|
|
"""A plugin that left core with its own login command (``hermes spotify login``), else ``hermes auth <name>``."""
|
|
from hermes_cli.left_core_migration import LEFT_CORE
|
|
return next((f"hermes {f.cli} login" for f in LEFT_CORE if f.plugin == name and f.cli), f"hermes auth {name}")
|
|
|
|
|
|
def _plugin_auth_hint(name: str, provides_tools: list) -> tuple:
|
|
"""``(auth_required, auth_command)`` from last-known cached tool availability only.
|
|
|
|
A missing cache entry is "unknown": schedule a background probe rather than probing
|
|
inline (which would starve the root event loop), so the short hub TTL picks it up.
|
|
"""
|
|
try:
|
|
from tools.registry import get_cached_check_fn_result, registry
|
|
for tname in provides_tools:
|
|
entry = registry.get_entry(tname)
|
|
if not entry or not entry.check_fn:
|
|
continue
|
|
cached_result = get_cached_check_fn_result(entry.check_fn)
|
|
if cached_result is None:
|
|
_schedule_check_fn_probe(entry.check_fn)
|
|
elif cached_result is False:
|
|
return True, _plugin_login_command(name)
|
|
except Exception:
|
|
pass
|
|
return False, ""
|
|
|
|
|
|
def _plugin_runtime_status(aliases: set, enabled_set: set, disabled_set: set) -> str:
|
|
"""enabled / disabled / inactive for a plugin's name+key alias set (disabled wins)."""
|
|
return "disabled" if aliases & disabled_set else "enabled" if aliases & enabled_set else "inactive"
|
|
|
|
|
|
def _merged_plugins_hub(force_refresh: bool = False) -> dict[str, Any]:
|
|
"""Agent discovery + dashboard manifests + provider picker metadata.
|
|
|
|
IMPORTANT: powers a dashboard request path, so it must stay read-only and cheap — never
|
|
execute tool ``check_fn`` probes here (imports, auth/network checks would starve the root
|
|
event loop). Only cached availability is consumed and the payload is memoized briefly to
|
|
collapse the dashboard's bursty duplicate fetches.
|
|
"""
|
|
from hermes_cli.web_server_memory import _discover_memory_provider_statuses, _normalize_memory_provider_name
|
|
from hermes_cli.web_server import _get_dashboard_plugins
|
|
from hermes_cli.config import get_hermes_home, load_config
|
|
from hermes_constants import hermes_home_key
|
|
|
|
cache_key = hermes_home_key(get_hermes_home())
|
|
now = time.monotonic()
|
|
if not force_refresh:
|
|
with _plugins_hub_cache_lock:
|
|
cached = _plugins_hub_cache.get(cache_key)
|
|
if cached is not None and now < _plugins_hub_cache_expires_at.get(cache_key, 0.0):
|
|
return cached
|
|
|
|
started_at = time.monotonic()
|
|
from hermes_cli.plugins_cmd import (
|
|
_category_active_names,
|
|
_discover_all_plugins,
|
|
_get_current_context_engine,
|
|
_get_current_memory_provider,
|
|
_discover_context_engines,
|
|
_get_disabled_set,
|
|
_get_enabled_set,
|
|
_plugin_status,
|
|
_read_manifest as _read_plugin_manifest_at,
|
|
)
|
|
from hermes_cli.plugins_cmd_catalog import removed_annotation
|
|
from hermes_cli.plugin_catalog import resolved_removed_entries
|
|
|
|
dashboard_list = _get_dashboard_plugins()
|
|
dash_by_name = {str(p["name"]): p for p in dashboard_list}
|
|
disabled_set = _get_disabled_set()
|
|
enabled_set = _get_enabled_set()
|
|
hidden_plugins: list = cfg_get(load_config(), "dashboard", "hidden_plugins", default=[]) or []
|
|
plugins_root_resolved = (get_hermes_home() / "plugins").resolve()
|
|
rows: list[dict[str, Any]] = []
|
|
|
|
# One kill-list resolution for the whole rebuild: resolving per row costs a live-catalog
|
|
# fetch per installed plugin when the catalog host is slow or unreachable.
|
|
removed_entries = resolved_removed_entries()
|
|
active = _category_active_names()
|
|
|
|
for name, version, description, source, dir_str, key in _discover_all_plugins():
|
|
# Same verdict as `hermes plugins list` / the TUI hub: name+key aliases for the lists, bundled
|
|
# backends/platforms/providers and the live memory provider count as enabled without a list
|
|
# entry (#73131, #82898).
|
|
runtime_status = _plugin_status(
|
|
name, enabled_set, disabled_set, key=key, source=source, dir_path=dir_str, active=active)
|
|
if runtime_status == "not enabled":
|
|
runtime_status = "inactive"
|
|
|
|
dir_path = Path(dir_str)
|
|
dm = dash_by_name.get(name)
|
|
try:
|
|
dir_path.resolve().relative_to(plugins_root_resolved)
|
|
under_user_tree = True
|
|
except ValueError:
|
|
under_user_tree = False
|
|
can_remove_update = source in {"user", "git"} and under_user_tree and dir_path.is_dir()
|
|
|
|
provides_tools = _read_plugin_manifest_at(dir_path).get("provides_tools") or []
|
|
auth_required, auth_command = _plugin_auth_hint(name, provides_tools) if provides_tools else (False, "")
|
|
|
|
rows.append({
|
|
"name": name,
|
|
"version": version or "",
|
|
"description": description or "",
|
|
"source": source,
|
|
"runtime_status": runtime_status,
|
|
"has_dashboard_manifest": dm is not None or (dir_path / "dashboard" / "manifest.json").exists(),
|
|
"dashboard_manifest": _strip_dashboard_manifest(dm) if dm else None,
|
|
"path": dir_str,
|
|
"can_remove": can_remove_update,
|
|
"can_update_git": can_remove_update and (dir_path / ".git").exists(),
|
|
"auth_required": auth_required,
|
|
"auth_command": auth_command,
|
|
"user_hidden": name in hidden_plugins,
|
|
"removed_reason": removed_annotation(name, dir_str, removed_entries),
|
|
})
|
|
|
|
agent_names = {r["name"] for r in rows}
|
|
orphan_dashboard = [_strip_dashboard_manifest(p) for p in dashboard_list if str(p["name"]) not in agent_names]
|
|
memory_providers = _discover_memory_provider_statuses()
|
|
try:
|
|
context_engines = [{"name": n, "description": desc} for n, desc in _discover_context_engines()]
|
|
except Exception:
|
|
context_engines = []
|
|
|
|
payload = {
|
|
"plugins": rows,
|
|
"orphan_dashboard_plugins": orphan_dashboard,
|
|
"providers": {
|
|
"memory_provider": _normalize_memory_provider_name(_get_current_memory_provider()),
|
|
"memory_options": memory_providers,
|
|
"context_engine": _get_current_context_engine(),
|
|
"context_options": context_engines,
|
|
},
|
|
}
|
|
duration = time.monotonic() - started_at
|
|
if duration >= 0.25:
|
|
_log.info(
|
|
"plugins/hub rebuilt in %.3fs (plugins=%d memory_options=%d)", duration, len(rows), len(memory_providers)
|
|
)
|
|
with _plugins_hub_cache_lock:
|
|
_plugins_hub_cache[cache_key] = payload
|
|
_plugins_hub_cache_expires_at[cache_key] = time.monotonic() + _PLUGINS_HUB_CACHE_TTL_SECONDS
|
|
return payload
|
|
|
|
|
|
def _plugin_api_mount_skip_reason(plugin: dict[str, Any], enabled_set: set, disabled_set: set) -> Optional[str]:
|
|
"""Why a plugin's backend ``api`` must NOT be imported, or None when it may be.
|
|
|
|
User plugins must be in ``plugins.enabled`` and not ``plugins.disabled`` before their
|
|
Python runs (GHSA-mcfc-hp25-cjv7); bundled plugins are trusted but respect an explicit
|
|
disable; project plugins (``./.hermes/plugins/``) ship with the CWD and are
|
|
attacker-controlled when opening a malicious repo — never auto-imported (GHSA-5qr3-c538-wm9j).
|
|
"""
|
|
source, plugin_name = plugin.get("source"), plugin.get("name", "")
|
|
if source in ("user", "bundled") and plugin_name in disabled_set:
|
|
return "explicitly disabled"
|
|
if source == "user" and plugin_name not in enabled_set:
|
|
return "not in plugins.enabled"
|
|
return None
|
|
|
|
|
|
@contextmanager
|
|
def _launch_profile_scope():
|
|
"""Resolve config as the launch profile's, whichever profile the request is scoped to: home and
|
|
secrets switch together, so the launch config's ``${VAR}`` refs (and its last-good copy) never
|
|
expand through another profile's secrets."""
|
|
from hermes_constants import reset_hermes_home_override, set_hermes_home_override
|
|
from hermes_cli.web_server_profiles import _config_profile_scope
|
|
token = set_hermes_home_override(None)
|
|
try:
|
|
with _config_profile_scope(None):
|
|
yield
|
|
finally:
|
|
reset_hermes_home_override(token)
|
|
|
|
|
|
def _plugin_enable_sets_here() -> tuple[set, set]:
|
|
"""``(plugins.enabled, plugins.disabled)`` of the active profile scope from one cached read-only
|
|
config read: the gates run on the event loop for every plugin request and upgrade."""
|
|
from hermes_cli.config import load_config_readonly
|
|
try:
|
|
config = load_config_readonly()
|
|
except Exception: # health: allow BLE001 -- an unreadable config grants no consent (empty sets), as plugins_cmd reads it
|
|
config = {}
|
|
enabled, disabled = (cfg_get(config, "plugins", key, default=[]) for key in ("enabled", "disabled"))
|
|
return (set(enabled) if isinstance(enabled, list) else set(),
|
|
set(disabled) if isinstance(disabled, list) else set())
|
|
|
|
|
|
def _plugin_consent_withheld(plugin: dict[str, Any]) -> bool:
|
|
"""Whether a request must not reach ``plugin``'s backend or assets (#46435, GHSA-mcfc-hp25-cjv7).
|
|
|
|
``plugin`` is the active scope's own copy (a secondary profile's plugins are discovered under
|
|
its home and run in its plugin host, :mod:`hermes_cli.web_server_plugin_profiles`), so the
|
|
profile the request acts for decides alone: a native ticket's profile, a ``?profile=``
|
|
selection, else the launch profile (#134889, PLG-3).
|
|
"""
|
|
return _plugin_api_mount_skip_reason(plugin, *_plugin_enable_sets_here()) is not None
|
|
|
|
|
|
def _plugin_api_name(scope) -> str:
|
|
"""``<name>`` of an ``/api/plugins/<name>/...`` request (the path the router matches), else ``""``."""
|
|
path = get_route_path(scope)
|
|
parts = path.split("/") # ['', 'api', 'plugins', '<name>', ...]
|
|
return parts[3] if path.startswith("/api/plugins/") and len(parts) >= 4 else ""
|
|
|
|
|
|
async def _plugin_api_request_refused(name: str) -> bool:
|
|
"""The runtime gate for one ``/api/plugins/<name>/`` request, HTTP and WebSocket alike.
|
|
|
|
:func:`_mount_plugin_api_routes` gates at import time; a plugin disabled while the server runs
|
|
keeps its router mounted until restart, so every request is checked again. An unknown name
|
|
counts as a user plugin (safe default: refused).
|
|
"""
|
|
from hermes_cli.web_server import _get_dashboard_plugins
|
|
from hermes_cli.web_server_plugin_profiles import rescanned_plugin, secondary_profile_home
|
|
plugin = next((p for p in _get_dashboard_plugins() if p.get("name") == name), None)
|
|
home = secondary_profile_home()
|
|
if plugin is None and home is not None: # installed since the last scan (rate-limited per name)
|
|
plugin = await asyncio.to_thread(rescanned_plugin, home, name) # a disk scan: off the event loop
|
|
return _plugin_consent_withheld(plugin or {"name": name, "source": "user"})
|
|
|
|
|
|
class PluginApiWebSocketGate:
|
|
""":func:`_plugin_api_request_refused` for WebSocket upgrades. ``_plugin_api_runtime_gate`` is
|
|
HTTP middleware and never sees one, so a plugin disabled at runtime kept accepting connections
|
|
on its websocket routes until restart (#134889).
|
|
|
|
It refuses before the route's own credential check, which is the disabled plugin's code, so it
|
|
cannot wait for auth (a gated-mode ticket is single-use: only the route may redeem it). That is
|
|
no plugin-name oracle: a close before accept reaches the client as the same 403 handshake
|
|
answer whether the plugin is disabled, unknown, or the credential was refused.
|
|
|
|
A Desktop's native grant is the exception: it names the profile the socket acts for, so it is
|
|
redeemed here, once (:func:`_native_ws_scope`), and consent is checked in that profile. A
|
|
secondary profile's socket to a plugin its own host serves is answered here too (PLG-3).
|
|
|
|
First it runs the core sockets' Host/Origin/peer gate (``_ws_request_is_allowed``, 4403), which
|
|
HTTP middleware never applies to an upgrade. Here it covers every plugin socket, whatever form
|
|
declares it: a router dependency reaches ``@router.websocket`` routes only.
|
|
"""
|
|
|
|
def __init__(self, app) -> None:
|
|
self.app = app
|
|
|
|
async def __call__(self, scope, receive, send) -> None:
|
|
from hermes_cli.web_server_chat import _ws_request_is_allowed
|
|
from hermes_cli.web_server_plugin_profiles import plugin_request_scope, refuse_secondary_plugin_socket
|
|
name = _plugin_api_name(scope) if scope["type"] == "websocket" else ""
|
|
if not name:
|
|
await self.app(scope, receive, send)
|
|
return
|
|
if not _ws_request_is_allowed(WebSocket(scope, receive, send)):
|
|
await send({"type": "websocket.close", "code": 4403})
|
|
return
|
|
with _native_ws_scope(scope) as admitted, plugin_request_scope(scope):
|
|
if admitted and not await _plugin_api_request_refused(name):
|
|
if not await refuse_secondary_plugin_socket(scope, receive, send, name):
|
|
await self.app(scope, receive, send)
|
|
return
|
|
await send({"type": "websocket.close", "code": status.WS_1008_POLICY_VIOLATION})
|
|
|
|
|
|
@contextmanager
|
|
def _native_ws_scope(scope):
|
|
"""Yield whether a plugin socket's native grant (``X-Hermes-Gateway-Ticket``) was admitted,
|
|
with that grant's home and secret scope bound for the rest of the connection; without the
|
|
header, True and the ambient scope. The grant is redeemed here only: the route's credential
|
|
check reads the principal stamped on the connection, and its scope dependency resolves the
|
|
same profile, so consent, home and ``get_secret`` all follow the grant, as on HTTP
|
|
(``native_profile_scope``).
|
|
"""
|
|
from hermes_cli.dashboard_auth.native_http import HEADER, native_profile_scope, redeem_native_ws
|
|
from hermes_cli.web_server_profiles import _config_profile_scope
|
|
conn = HTTPConnection(scope)
|
|
if HEADER not in conn.headers:
|
|
yield True
|
|
return
|
|
grant = redeem_native_ws(conn)
|
|
if grant is None:
|
|
yield False
|
|
return
|
|
conn.state.native_http_principal = grant
|
|
with native_profile_scope(conn), _config_profile_scope(None):
|
|
yield True
|
|
|
|
|
|
# Serializes plugin API mounts: concurrent first requests for a freshly enabled plugin must import
|
|
# and mount it exactly once.
|
|
_plugin_api_mount_lock = threading.Lock()
|
|
|
|
|
|
async def _plugin_route_secret_scope(conn: HTTPConnection, profile: Optional[str] = None):
|
|
"""Home + secret scope for one ``/api/plugins/<name>/`` request: the launch profile's, or the
|
|
``?profile=``-requested one — the same ``_config_profile_scope`` seam the built-in routers use.
|
|
Without it plugin handlers ran unscoped, so under multi-profile hosting every ``get_secret``
|
|
/ ``resolve_runtime_provider`` failed closed with ``UnscopedSecretError`` (#120310; the bundled
|
|
kanban plugin's Decompose / Specify / Estimate aux-LLM calls, #123372). ``async`` on purpose:
|
|
a sync yield-dependency's setup and teardown run on different threadpool threads, so the
|
|
scope token would be reset in a foreign context; sync handlers still see the scope because
|
|
``run_in_threadpool`` copies the request context into the worker."""
|
|
from hermes_cli.web_server_plugin_profiles import in_process_router_refused
|
|
from hermes_cli.web_server_profiles import _config_profile_scope, _is_current_profile
|
|
name = _plugin_api_name(conn.scope)
|
|
with _config_profile_scope(profile):
|
|
# The plugin must not run on a profile that disabled it (#134889), and this router is the
|
|
# launch profile's copy: a secondary profile reaches it only for bundled plugins (PLG-3).
|
|
if (not _is_current_profile(profile) and await _plugin_api_request_refused(name)) or in_process_router_refused(name):
|
|
if conn.scope["type"] == "websocket":
|
|
raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)
|
|
raise HTTPException(status_code=404, detail="Plugin not found")
|
|
yield
|
|
|
|
|
|
def _mount_hosted_plugin_api(app, plugin: dict, api_file_name: str) -> None:
|
|
"""``plugins.isolation: host``: the plugin's router runs in the requesting profile's plugin host;
|
|
this process only forwards each ``/api/plugins/<name>/`` request to it (auth and the profile
|
|
scope still apply here first). Responses are buffered: streaming and websockets need in-process."""
|
|
name = plugin["name"]
|
|
|
|
async def forward(request: Request, path: str = "") -> Response:
|
|
# Resolve the plugin again in the REQUESTING profile (the route is mounted once, from the
|
|
# launch profile): gated by its own plugins.enabled, served by its own host. The copy is
|
|
# still the launch profile's (discovery scans only the launch home and the root).
|
|
target = await asyncio.to_thread(_hosted_plugin_for_request, name)
|
|
if target is None:
|
|
return JSONResponse({"detail": f"plugin {name!r} is not enabled in this profile"}, status_code=404)
|
|
host, dashboard_dir, api_file = target
|
|
from hermes_cli.web_server_plugin_profiles import run_hosted_request
|
|
query, headers = request.url.query, [(k, v) for k, v in request.headers.items() if k.lower() != "host"]
|
|
body = await request.body()
|
|
from hermes_cli.plugin_host_http_auth import admitted_for_sensitive_route
|
|
authenticated = admitted_for_sensitive_route(request)
|
|
return await run_hosted_request(host.home, name, lambda timeout: host.asgi_request(
|
|
name, dashboard_dir, api_file, request.method, "/" + path, query, headers, body,
|
|
timeout=timeout, authenticated=authenticated))
|
|
|
|
app.add_api_route(f"/api/plugins/{name}/{{path:path}}", forward, include_in_schema=False,
|
|
methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "HEAD"],
|
|
dependencies=[Depends(_plugin_route_secret_scope)])
|
|
_log.info("Mounted plugin API routes via the plugin host: /api/plugins/%s/", name)
|
|
|
|
|
|
def _hosted_plugin_for_request(name: str) -> Optional[tuple]:
|
|
"""``(plugin host, dashboard dir, api file)`` for plugin ``name`` in the active profile, or None
|
|
when that profile has no enabled user copy of it."""
|
|
from hermes_cli.plugins import get_plugin_manager
|
|
from hermes_cli.plugins_cmd import _get_disabled_set, _get_enabled_set
|
|
for plugin in _discover_dashboard_plugins():
|
|
if plugin.get("name") != name or not plugin.get("_api_file") or plugin.get("source") != "user":
|
|
continue
|
|
if _plugin_api_mount_skip_reason(plugin, _get_enabled_set(), _get_disabled_set()):
|
|
return None
|
|
return get_plugin_manager()._plugin_host(), str(plugin["_dir"]), str(plugin["_api_file"])
|
|
return None
|
|
|
|
|
|
def _mount_plugin_api_routes():
|
|
"""Import and mount backend API routes from plugins that declare them.
|
|
|
|
Each plugin's ``api`` file must expose a ``router`` (FastAPI APIRouter), mounted under
|
|
``/api/plugins/<name>/``. See ``_plugin_api_mount_skip_reason`` for the trust gates.
|
|
|
|
Backend import is restricted to ``bundled`` and ``user`` sources. Project plugins
|
|
(``./.hermes/plugins/``) ship with the CWD and are therefore attacker-controlled in any threat model
|
|
where the user opens a malicious repo; they can extend the dashboard UI via static JS/CSS but their
|
|
Python ``api`` file is never auto-imported by the web server. See GHSA-5qr3-c538-wm9j (#29156).
|
|
Additionally, user plugins must be explicitly enabled via the ``plugins.enabled`` allow-list in
|
|
config.yaml before their backend code is imported. Without this gate, an installed-but-not-enabled
|
|
plugin's Python code would execute at dashboard startup — a code execution vector that bypasses the
|
|
user's intent. (#46435, GHSA-mcfc-hp25-cjv7)
|
|
"""
|
|
from hermes_cli.web_server import _get_dashboard_plugins, app
|
|
try:
|
|
from hermes_cli.plugins_cmd import _get_enabled_set, _get_disabled_set
|
|
enabled_set = _get_enabled_set()
|
|
disabled_set = _get_disabled_set()
|
|
except Exception:
|
|
enabled_set = set()
|
|
disabled_set = set()
|
|
|
|
for plugin in _get_dashboard_plugins():
|
|
with _plugin_api_mount_lock:
|
|
_mount_one_plugin_api(app, plugin, enabled_set, disabled_set)
|
|
_register_plugin_api_fallback(app)
|
|
|
|
|
|
def _try_hot_mount_plugin_api(name: str) -> bool:
|
|
"""Mount plugin ``name``'s backend API on the running app; True when new routes went live.
|
|
|
|
A plugin installed or enabled after server start had no route until restart: the startup
|
|
sweep is the only other caller of ``_mount_one_plugin_api``. Discovery is re-run (the plugin
|
|
dir may not have existed at startup) and the decision is the launch profile's, exactly as at
|
|
startup (its enable sets and its ``plugins.isolation``), so a plugin the launch profile did not
|
|
enable is never imported or forwarded (#134889); a secondary profile's request (its grant or a
|
|
``?profile=`` selection) may mount only its own bundled copy, by its own lists (PLG-3). New
|
|
routes are spliced in just ahead of the hot-mount fallback, which sits before the SPA
|
|
catch-all: appended at the end they would land behind ``/{full_path:path}`` and every GET
|
|
would keep missing them.
|
|
"""
|
|
if not name or any(c in name for c in "/\\{}") or name == "..":
|
|
return False
|
|
from hermes_cli.plugins_cmd import _get_disabled_set, _get_enabled_set
|
|
from hermes_cli.web_server import _get_dashboard_plugins, app
|
|
from hermes_cli.web_server_plugin_profiles import plugin_api_path, secondary_plugin, secondary_profile_home
|
|
# A secondary profile reaches this router only for its own bundled copy (Hermes code), and its
|
|
# own lists decide that one (PLG-3).
|
|
secondary, own, requester_sets = secondary_profile_home() is not None, secondary_plugin(name), _plugin_enable_sets_here()
|
|
with _launch_profile_scope(), _plugin_api_mount_lock:
|
|
enabled_set, disabled_set = _get_enabled_set(), _get_disabled_set()
|
|
if _plugin_api_mounted(app, name): # a mounted plugin's unknown subpath: no disk rescan
|
|
return False
|
|
plugin = next((p for p in _get_dashboard_plugins(force_rescan=True) if p.get("name") == name), None)
|
|
if secondary:
|
|
if own is None or own.get("source") != "bundled" or plugin_api_path(plugin) != plugin_api_path(own):
|
|
return False
|
|
enabled_set, disabled_set = requester_sets
|
|
routes = app.router.routes
|
|
before = len(routes)
|
|
if plugin is None or not _mount_one_plugin_api(app, plugin, enabled_set, disabled_set):
|
|
return False
|
|
fresh, rest = routes[before:], routes[:before]
|
|
at = next((i for i, r in enumerate(rest) if getattr(r, "endpoint", None) is _plugin_api_hot_mount), len(rest))
|
|
# One slice assignment: a request iterating the list on the loop thread sees routes
|
|
# shift right (re-checks one), never skips one.
|
|
routes[:] = rest[:at] + fresh + rest[at:]
|
|
return True
|
|
|
|
|
|
def _plugin_api_mounted(app, name: str) -> bool:
|
|
"""Whether ``app`` already serves ``/api/plugins/<name>/`` — read from the route table itself,
|
|
so anything that rebuilds the routes (a test restoring its snapshot) cannot leave a stale
|
|
"mounted" record that blocks the remount."""
|
|
prefix = f"/api/plugins/{name}"
|
|
return any(
|
|
getattr(r, "endpoint", None) is not _plugin_api_hot_mount
|
|
and (getattr(r, "path", "") == prefix or getattr(r, "path", "").startswith(prefix + "/"))
|
|
for r in app.router.routes
|
|
)
|
|
|
|
|
|
class _PluginApiHotMount:
|
|
"""ASGI fallback for ``/api/plugins/{name}/{path}`` requests no mounted router matched.
|
|
|
|
Registered after every startup router, so it only sees plugins that are not mounted yet
|
|
(auth and ``_plugin_api_runtime_gate`` already ran). On a successful hot mount the SAME
|
|
request is re-routed to the fresh routes in-process — no redirect: Desktop's transport
|
|
never follows a 3xx and would fail the first poll. A second miss lands back here, finds the
|
|
plugin mounted and gets the plain 404, so re-routing cannot loop. A class instance, not a
|
|
function: Starlette wraps plain functions as ``request -> response`` endpoints.
|
|
"""
|
|
|
|
async def __call__(self, scope, receive, send) -> None:
|
|
from hermes_cli.web_server import app
|
|
from hermes_cli.web_server_plugin_profiles import plugin_request_scope
|
|
name = str(scope.get("path_params", {}).get("name", ""))
|
|
with plugin_request_scope(scope): # a ``?profile=`` selection decides its own cold mount (PLG-3)
|
|
mounted = await asyncio.to_thread(_try_hot_mount_plugin_api, name)
|
|
if mounted:
|
|
_log.info("Hot-mounted plugin API routes on first request: /api/plugins/%s/", name)
|
|
await app.router(dict(scope, path_params={}), receive, send)
|
|
return
|
|
await JSONResponse({"detail": "Not Found"}, status_code=404)(scope, receive, send)
|
|
|
|
|
|
_plugin_api_hot_mount = _PluginApiHotMount()
|
|
|
|
|
|
def _register_plugin_api_fallback(app) -> None:
|
|
"""Register the hot-mount fallback once, after the startup plugin routers (so a mounted
|
|
router always matches first) and before the SPA catch-all."""
|
|
if any(getattr(r, "endpoint", None) is _plugin_api_hot_mount for r in app.router.routes):
|
|
return
|
|
app.router.add_route("/api/plugins/{name}/{path:path}", _plugin_api_hot_mount,
|
|
methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "HEAD"],
|
|
include_in_schema=False)
|
|
|
|
|
|
def _mount_one_plugin_api(app, plugin: dict[str, Any], enabled_set: set, disabled_set: set) -> bool:
|
|
"""Mount one plugin's backend API routes on ``app`` if its trust gates pass.
|
|
|
|
Returns True when routes were added. Callers must hold ``_plugin_api_mount_lock``.
|
|
Shared by the import-time sweep and the hot-mount fallback so both paths
|
|
apply the exact same gates (GHSA-mcfc-hp25-cjv7, GHSA-5qr3-c538-wm9j).
|
|
"""
|
|
api_file_name = plugin.get("_api_file")
|
|
if not api_file_name:
|
|
return False
|
|
if _plugin_api_mounted(app, plugin.get("name", "")):
|
|
return False
|
|
skip = _plugin_api_mount_skip_reason(plugin, enabled_set, disabled_set)
|
|
if skip:
|
|
_log.debug("Plugin %s: skipping API mount (%s)", plugin.get("name", ""), skip)
|
|
return False
|
|
if plugin.get("source") not in ("bundled", "project"):
|
|
from hermes_cli.plugin_isolation import ISOLATION_HOST, isolation_mode
|
|
if isolation_mode() == ISOLATION_HOST:
|
|
_mount_hosted_plugin_api(app, plugin, api_file_name)
|
|
return True
|
|
if plugin.get("source") == "project":
|
|
_log.warning(
|
|
"Plugin %s: ignoring backend api=%s (project plugins may "
|
|
"not auto-import Python code; move the plugin to "
|
|
"~/.hermes/plugins/ if you trust it)",
|
|
plugin["name"], api_file_name,
|
|
)
|
|
return False
|
|
dashboard_dir = Path(plugin["_dir"])
|
|
api_path = dashboard_dir / api_file_name
|
|
try:
|
|
api_path.resolve().relative_to(dashboard_dir.resolve())
|
|
except (OSError, RuntimeError, ValueError):
|
|
# Discovery already filters this; defence in depth in case ``_dir`` was tampered
|
|
# with after caching or a future caller bypasses the validator.
|
|
_log.warning(
|
|
"Plugin %s: refusing to import api file outside its "
|
|
"dashboard directory (%s)", plugin["name"], api_path,
|
|
)
|
|
return False
|
|
if not api_path.exists():
|
|
_log.warning("Plugin %s declares api=%s but file not found", plugin["name"], api_file_name)
|
|
return False
|
|
try:
|
|
from hermes_cli.plugin_isolation import import_user_code
|
|
# Registered in sys.modules while the body runs so pydantic/FastAPI can resolve
|
|
# string annotations (``from __future__ import annotations``) by module name.
|
|
mod = import_user_code(api_path, f"hermes_dashboard_plugin_{plugin['name']}")
|
|
if mod is None:
|
|
return False
|
|
router = getattr(mod, "router", None)
|
|
if router is None:
|
|
_log.warning("Plugin %s api file has no 'router' attribute", plugin["name"])
|
|
return False
|
|
app.include_router(
|
|
router,
|
|
prefix=f"/api/plugins/{plugin['name']}",
|
|
dependencies=[Depends(_plugin_route_secret_scope)],
|
|
)
|
|
_log.info("Mounted plugin API routes: /api/plugins/%s/", plugin["name"])
|
|
return True
|
|
except Exception as exc: # health: allow BLE001 -- executes third-party plugin code; any failure must skip that plugin, not crash the server
|
|
_log.warning("Failed to load plugin %s API routes: %s", plugin["name"], exc, exc_info=True)
|
|
return False
|