"""Opt-in WebUI extension hooks. This module intentionally provides a small, self-hosted extension surface:configured same-origin script/style injection plus sandboxed static file serving.It is disabled by default and never executes or fetches third-party URLs.""" import htmlimport http.clientimport jsonimport mathimport loggingimport osimport reimport socketimport threadingfrom pathlib import Pathfrom typing import Any, Dict, List, Optional, Set, Tuplefrom urllib.parse import unquote, urlsplitimport hashlibimport ioimport timeimport zipfilefrom urllib.request import (    HTTPRedirectHandler,    HTTPSHandler,    build_opener,) from api.helpers import _security_headers, j _log = logging.getLogger(__name__) # Sane bound on configured URLs — real extensions ship 1-3 files. Higher values# typically indicate a misconfiguration (one giant unsplit string, or a runaway# generator script that wrote an env-var template without filtering). Capping# avoids rendering tens of thousands of <script> tags into every page load._MAX_URL_LIST = 32 # Keep extension manifests small and auditable. The manifest is a convenience for# bundling static assets, not a package manager or dependency lockfile._MAX_MANIFEST_BYTES = 64 * 1024 # Tracks rejected URL strings we've already warned about so a misconfigured env# var doesn't spam the log on every request that re-reads it._warned_urls: set = set()  class _ManifestTooLarge(ValueError):    pass  class ExtensionToggleError(Exception):    """Sanitized extension mutation error safe to return to the browser."""     def __init__(self, message: str, status: int = 400):        super().__init__(message)        self.status = status  class ExtensionInstallError(Exception):    """Sanitized extension install/uninstall error safe to return to the browser."""     def __init__(self, message: str, status: int = 400):        super().__init__(message)        self.status = status  class ExtensionSidecarProxyError(Exception):    """Sanitized sidecar proxy error safe to return to the browser."""     def __init__(self, message: str, status: int = 400):        super().__init__(message)        self.status = status  EXTENSION_ROUTE_PREFIX = "/extensions/"_EXTENSION_DIR_ENV = "HERMES_WEBUI_EXTENSION_DIR"_EXTENSION_SCRIPT_URLS_ENV = "HERMES_WEBUI_EXTENSION_SCRIPT_URLS"_EXTENSION_STYLESHEET_URLS_ENV = "HERMES_WEBUI_EXTENSION_STYLESHEET_URLS"_EXTENSION_MANIFEST_ENV = "HERMES_WEBUI_EXTENSION_MANIFEST"_ALLOWED_ASSET_PREFIXES = ("/extensions/", "/static/")_SIDECAR_WARNING_SOURCE = "manifest:sidecars"_DEFAULT_SIDECAR_HEALTH_PATH = "/health"_LOOPBACK_SIDECAR_HOSTS = {"127.0.0.1", "localhost", "::1"}_EXTENSION_STATE_FILENAME = "extension-overrides.json"_MAX_EXTENSION_STATE_BYTES = 32 * 1024_MAX_DISABLED_EXTENSION_IDS = 512_MAX_SIDECAR_PROXY_CONSENTS = 512_EXTENSION_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")_EXTENSION_SETTINGS_KEY_RE = re.compile(r"^[A-Za-z][A-Za-z0-9._-]{0,63}$")_EXTENSION_STATE_WARNING_SOURCE = "extension_state"_EXTENSION_STATE_LOCK = threading.Lock()_EXTENSION_SETTING_TYPES = {"boolean", "string", "number", "integer", "enum"} _GALLERY_INSTALL_STATE_FILENAME = "extension-install-manifest.json"_MAX_INSTALL_MANIFEST_BYTES = 128 * 1024_MAX_GALLERY_INSTALLED_IDS = 256_MAX_ZIP_DOWNLOAD_BYTES = 32 * 1024 * 1024_REGISTRY_URL = "https://hermes-webui.github.io/hermes-webui-extensions/registry.json"_REGISTRY_ALLOWED_DOWNLOAD_HOSTS = frozenset({"hermes-webui.github.io"})_REGISTRY_CACHE: dict = {}_REGISTRY_LOCK = threading.Lock()_REGISTRY_TTL_SECONDS = 300  class _AllowlistRedirectHandler(HTTPRedirectHandler):    """Reject redirects to hosts not in the download allowlist."""     def redirect_request(self, req, fp, code, msg, headers, newurl):        parsed = urlsplit(newurl)        if parsed.scheme != "https" or parsed.hostname not in _REGISTRY_ALLOWED_DOWNLOAD_HOSTS:            raise ExtensionInstallError("Download redirected to disallowed host")        return super().redirect_request(req, fp, code, msg, headers, newurl)  def _connect_ipv4_first(address, timeout=socket._GLOBAL_DEFAULT_TIMEOUT, source_address=None):    """Connect to *address*, preferring IPv4 to avoid dead-route stalls.     Some networks advertise IPv6 routes (router advertisements, tunnel brokers)    that do not actually reach the public internet.  ``socket.create_connection``    follows the OS address ordering, which typically tries IPv6 first; each dead    address stalls for the full TCP timeout (~75–130 s) before falling back to    IPv4.  This makes gallery installs and registry fetches appear to hang until    the browser or curl timeout kills them.     We iterate families in IPv4→IPv6 order so the common case (IPv4 works, IPv6    is a broken route) completes instantly.  IPv6 is still tried as a fallback so    IPv6-only networks are unaffected.     Signature matches ``socket.create_connection`` including the    ``_GLOBAL_DEFAULT_TIMEOUT`` sentinel so callers that rely on the stdlib    default-timeout behaviour are not broken.    """    host, port = address    if timeout is socket._GLOBAL_DEFAULT_TIMEOUT:        timeout = socket.getdefaulttimeout()    last_error: OSError | None = None    for family in (socket.AF_INET, socket.AF_INET6):        try:            infos = socket.getaddrinfo(host, port, family, socket.SOCK_STREAM)        except socket.gaierror:            continue        for _fam, socktype, proto, _canon, sockaddr in infos:            sock = None            try:                sock = socket.socket(_fam, socktype, proto)                if timeout is not None:                    sock.settimeout(timeout)                if source_address is not None:                    sock.bind(source_address)                sock.connect(sockaddr)                return sock            except OSError as exc:                last_error = exc                if sock is not None:                    sock.close()    if last_error is not None:        raise last_error    raise OSError(f"Could not connect to {host}:{port}")  class _IPv4FirstHTTPSConnection(http.client.HTTPSConnection):    """HTTPS connection that resolves IPv4 addresses before IPv6.     Overrides the class-level ``_create_connection`` that    ``HTTPConnection.connect()`` calls — the stdlib extension point designed for    exactly this kind of per-class customisation (added in Python 3.9,    bpo-37830).  No global state is mutated, so the change is thread-safe under    ``ThreadingHTTPServer``.    """     _create_connection = staticmethod(_connect_ipv4_first)  class _IPv4FirstHTTPSHandler(HTTPSHandler):    """HTTPS handler that builds IPv4-first connections for gallery downloads."""     def https_open(self, req):        return self.do_open(_IPv4FirstHTTPSConnection, req)  def _build_gallery_opener():    """Opener with IPv4-first downloads and redirect-host allowlist."""    return build_opener(_IPv4FirstHTTPSHandler, _AllowlistRedirectHandler)  def _safe_download(url: str, max_bytes: int, timeout: int = 30) -> bytes:    """Download from an allowlisted host, rejecting cross-host redirects."""    opener = _build_gallery_opener()    resp = opener.open(url, timeout=timeout)    try:        return resp.read(max_bytes + 1)    finally:        resp.close()  _EXTENSION_MIME = {    "css": "text/css",    "js": "application/javascript",    "html": "text/html",    "svg": "image/svg+xml",    "png": "image/png",    "jpg": "image/jpeg",    "jpeg": "image/jpeg",    "ico": "image/x-icon",    "gif": "image/gif",    "webp": "image/webp",    "woff": "font/woff",    "woff2": "font/woff2",    "ttf": "font/ttf",    "otf": "font/otf",    "wasm": "application/wasm",}_TEXT_MIME_TYPES = {"text/css", "application/javascript", "text/html", "image/svg+xml", "text/plain"}  def _default_extension_root() -> Path:    """WebUI-managed default extension directory under the state dir.     Used when ``HERMES_WEBUI_EXTENSION_DIR`` is unset so one-click gallery    install works out of the box on a single-user self-hosted instance with no    environment setup. It lives alongside sessions/settings in the WebUI-owned    state dir, which is a different trust domain from "a user-writable directory    on a shared box" — the loaded code still runs with full session authority,    so the trust model is unchanged (see docs/EXTENSIONS.md).    """    return _extension_state_dir() / "extensions"  def _extension_root() -> Optional[Path]:    """Return the active extension directory, or None when none is available.     Resolution order:    1. ``HERMES_WEBUI_EXTENSION_DIR`` when set — must be an existing directory,       otherwise None (the admin owns that path; we never auto-create it).    2. Otherwise the WebUI-managed default (``STATE_DIR/extensions``) when it       already exists. The first gallery install creates it on demand       (see ``_writable_extension_root``); until then this stays None and the       UI reports the same "nothing installed yet" state as before.    """    raw = os.getenv(_EXTENSION_DIR_ENV, "").strip()    if raw:        root = Path(raw).expanduser().resolve()        if not root.exists() or not root.is_dir():            return None        return root    default_root = _default_extension_root()    try:        if default_root.is_dir() and not default_root.is_symlink():            return default_root.resolve()    except OSError:        return None    return None  def _writable_extension_root() -> Optional[Path]:    """Resolve the extension root for writes, bootstrapping the managed default.     When ``HERMES_WEBUI_EXTENSION_DIR`` is set we use it as-is (the admin owns    it; it must already exist). When unset we create and return the    WebUI-managed default so a fresh install can install an extension with zero    configuration — plug and play.    """    raw = os.getenv(_EXTENSION_DIR_ENV, "").strip()    if raw:        return _extension_root()    default_root = _default_extension_root()    try:        default_root.mkdir(parents=True, exist_ok=True)    except OSError:        return None    try:        if default_root.is_symlink() or not default_root.is_dir():            return None        return default_root.resolve()    except OSError:        return None  def _extension_root_status() -> Tuple[Optional[Path], bool, bool]:    """Return (root, configured, valid) without exposing the configured path.     With no ``HERMES_WEBUI_EXTENSION_DIR`` the WebUI-managed default is always    available as an install target, so ``configured`` is True (extensions are    no longer "not configured" out of the box). ``valid`` reflects whether that    managed directory currently exists — it is created on the first install.    """    raw = os.getenv(_EXTENSION_DIR_ENV, "").strip()    if raw:        root = Path(raw).expanduser().resolve()        if not root.exists() or not root.is_dir():            return None, True, False        return root, True, True    root = _extension_root()    return root, True, root is not None  def _new_diagnostics() -> Dict[str, Any]:    return {"warnings": []}  def _add_diagnostic_warning(    diagnostics: Optional[Dict[str, Any]], code: str, source: str) -> None:    """Record a sanitized diagnostic warning.     Warnings intentionally carry only stable codes and coarse sources. They never    include filesystem paths, raw environment values, or rejected URL strings.    """    if diagnostics is None:        return    warnings = diagnostics.setdefault("warnings", [])    if not isinstance(warnings, list):        return    warning = {"code": code, "source": source}    if warning not in warnings:        warnings.append(warning)  def _valid_extension_id(value: object) -> bool:    return isinstance(value, str) and bool(_EXTENSION_ID_RE.fullmatch(value.strip()))  def _extension_state_dir() -> Path:    """Return the WebUI-managed state directory for extension overrides."""    try:        from api.config import STATE_DIR         return Path(STATE_DIR)    except Exception:        return Path(os.getenv("HERMES_WEBUI_STATE_DIR", str(Path.home() / ".hermes" / "webui"))).expanduser()  def _extension_state_file() -> Path:    return _extension_state_dir() / _EXTENSION_STATE_FILENAME  def _empty_extension_state() -> Dict[str, Any]:    return {"version": 1, "disabled_extensions": [], "sidecar_proxy_consents": {}}  def _load_extension_state(diagnostics: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:    """Load UI-managed extension overrides, failing safe without path leaks."""    state_file = _extension_state_file()    try:        if not state_file.exists() or not state_file.is_file():            return _empty_extension_state()        with state_file.open("rb") as fh:            raw = fh.read(_MAX_EXTENSION_STATE_BYTES + 1)        if len(raw) > _MAX_EXTENSION_STATE_BYTES:            _add_diagnostic_warning(                diagnostics, "extension_state_oversized", _EXTENSION_STATE_WARNING_SOURCE            )            return _empty_extension_state()        parsed = json.loads(raw.decode("utf-8"))    except (OSError, UnicodeDecodeError, json.JSONDecodeError, RecursionError):        _add_diagnostic_warning(            diagnostics, "extension_state_unreadable", _EXTENSION_STATE_WARNING_SOURCE        )        return _empty_extension_state()    if not isinstance(parsed, dict):        _add_diagnostic_warning(            diagnostics, "extension_state_invalid", _EXTENSION_STATE_WARNING_SOURCE        )        return _empty_extension_state()    disabled_raw = parsed.get("disabled_extensions", [])    if not isinstance(disabled_raw, list):        _add_diagnostic_warning(            diagnostics, "extension_state_invalid", _EXTENSION_STATE_WARNING_SOURCE        )        return _empty_extension_state()    disabled: List[str] = []    seen: Set[str] = set()    invalid = False    for value in disabled_raw:        if not _valid_extension_id(value):            invalid = True            continue        ext_id = str(value).strip()        if ext_id in seen:            continue        seen.add(ext_id)        disabled.append(ext_id)        if len(disabled) >= _MAX_DISABLED_EXTENSION_IDS:            _add_diagnostic_warning(                diagnostics, "extension_state_truncated", _EXTENSION_STATE_WARNING_SOURCE            )            break    consents_raw = parsed.get("sidecar_proxy_consents", {})    consents: Dict[str, str] = {}    consents_invalid = False    if consents_raw is not None:        if not isinstance(consents_raw, dict):            invalid = True            consents_invalid = True        else:            for raw_ext_id, raw_origin in consents_raw.items():                if not _valid_extension_id(raw_ext_id):                    invalid = True                    consents_invalid = True                    continue                origin = _normalize_loopback_sidecar_origin(raw_origin)                if origin is None:                    invalid = True                    consents_invalid = True                    continue                ext_id = str(raw_ext_id).strip()                if ext_id in consents:                    continue                consents[ext_id] = origin                if len(consents) >= _MAX_SIDECAR_PROXY_CONSENTS:                    _add_diagnostic_warning(                        diagnostics, "extension_state_truncated", _EXTENSION_STATE_WARNING_SOURCE                    )                    break    if consents_invalid:        consents = {}    if invalid:        _add_diagnostic_warning(            diagnostics, "extension_state_invalid_entries", _EXTENSION_STATE_WARNING_SOURCE        )    return {        "version": 1,        "disabled_extensions": disabled,        "sidecar_proxy_consents": consents,    }  def _write_extension_state(state: Dict[str, Any]) -> None:    """Persist extension overrides with an atomic same-directory replace."""    disabled_raw = state.get("disabled_extensions", [])    disabled: List[str] = []    seen: Set[str] = set()    if isinstance(disabled_raw, list):        for value in disabled_raw:            if not _valid_extension_id(value):                continue            ext_id = str(value).strip()            if ext_id in seen:                continue            seen.add(ext_id)            disabled.append(ext_id)            if len(disabled) >= _MAX_DISABLED_EXTENSION_IDS:                break    consents_raw = state.get("sidecar_proxy_consents", {})    consents: Dict[str, str] = {}    if isinstance(consents_raw, dict):        for raw_ext_id, raw_origin in consents_raw.items():            if not _valid_extension_id(raw_ext_id):                continue            origin = _normalize_loopback_sidecar_origin(raw_origin)            if origin is None:                continue            ext_id = str(raw_ext_id).strip()            if ext_id in consents:                continue            consents[ext_id] = origin            if len(consents) >= _MAX_SIDECAR_PROXY_CONSENTS:                break    payload = {        "version": 1,        "disabled_extensions": disabled,        "sidecar_proxy_consents": consents,    }    target = _extension_state_file()    target.parent.mkdir(parents=True, exist_ok=True)    tmp = target.with_name(f".{target.name}.{os.getpid()}.{threading.get_ident()}.tmp")    data = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8")    try:        with tmp.open("wb") as fh:            fh.write(data)            fh.flush()            os.fsync(fh.fileno())        os.replace(tmp, target)    finally:        try:            if tmp.exists():                tmp.unlink()        except OSError:            pass  def _fully_unquote_path(path: str) -> str:    """Decode percent-encoding until stable so encoded dot-segments cannot hide.     Iterates up to 10 times so even quadruple-encoded inputs like    ``%2525252e%2525252e`` collapse to literal ``..`` and are rejected by    the segment-level safety check downstream. URL strings stabilize in    fewer than 5 iterations in practice; the cap is defensive.    """    previous = path    for _ in range(10):        current = unquote(previous)        if current == previous:            return current        previous = current    return previous  

Lines 1–500 of 2014. Load more