#5228: opt-in extension loopback proxy (#4747), rebased on master; union urllib imports

This commit is contained in:
nesquena-hermes committed 2026-07-01 23:49:25 +00:00
1 parent 086d082149
commit 3d99763f3a
7 files changed
+1840 -88

No files matched your search

+320 -19
View File
@@ -60,6 +60,14 @@ class ExtensionInstallError(Exception):
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"
@@ -72,6 +80,7 @@ _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"
@@ -251,7 +260,7 @@ def _extension_state_file() -> Path:
def _empty_extension_state() -> Dict[str, Any]:
return {"version": 1, "disabled_extensions": []}
return {"version": 1, "disabled_extensions": [], "sidecar_proxy_consents": {}}
def _load_extension_state(diagnostics: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
@@ -301,11 +310,44 @@ def _load_extension_state(diagnostics: Optional[Dict[str, Any]] = None) -> Dict[
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}
return {
"version": 1,
"disabled_extensions": disabled,
"sidecar_proxy_consents": consents,
}
def _write_extension_state(state: Dict[str, Any]) -> None:
@@ -324,7 +366,26 @@ def _write_extension_state(state: Dict[str, Any]) -> None:
disabled.append(ext_id)
if len(disabled) >= _MAX_DISABLED_EXTENSION_IDS:
break
payload = {"version": 1, "disabled_extensions": disabled}
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")
@@ -712,6 +773,22 @@ def _normalize_sidecar_health_path(value: object) -> Optional[str]:
return decoded_path
def _is_valid_sidecar_proxy_path(decoded_path: str) -> bool:
if any(ch in decoded_path for ch in ("?", "#")):
return False
if any(ch.isspace() for ch in decoded_path):
return False
if not decoded_path.startswith("/") or decoded_path.startswith("//"):
return False
segments = decoded_path.split("/")[1:]
if not segments:
return False
for segment in segments:
if not segment or segment in (".", ".."):
return False
return True
def _sidecar_from_manifest_entry(
entry: Dict[str, object], diagnostics: Optional[Dict[str, Any]] = None
) -> Optional[Dict[str, str]]:
@@ -931,7 +1008,10 @@ def _load_manifest_with_status(
def _manifest_extension_state(
manifest: object, disabled_ids: Set[str], diagnostics: Optional[Dict[str, Any]] = None
manifest: object,
disabled_ids: Set[str],
diagnostics: Optional[Dict[str, Any]] = None,
consent_ids: Optional[Set[str]] = None,
) -> Dict[str, Any]:
"""Return sanitized per-extension state for manifest extension entries."""
extension_entries: List[Dict[str, Any]] = []
@@ -982,7 +1062,7 @@ def _manifest_extension_state(
_add_diagnostic_warning(diagnostics, "manifest_extension_id_invalid", "manifest:extensions")
if duplicate_seen:
_add_diagnostic_warning(diagnostics, "manifest_extension_id_duplicate", "manifest:extensions")
stale_ids = sorted(disabled_ids - known_ids)
stale_ids = sorted((disabled_ids | (consent_ids or set())) - known_ids)
if stale_ids:
_add_diagnostic_warning(diagnostics, "extension_state_unknown_ids", _EXTENSION_STATE_WARNING_SOURCE)
return {
@@ -991,6 +1071,98 @@ def _manifest_extension_state(
"manifest_disabled_ids": manifest_disabled_ids,
}
def _extension_sidecar_proxy_path(extension_id: str) -> str:
return f"/api/extensions/{extension_id}/sidecar/"
def _sidecar_proxy_public_status(
extension_id: str,
origin: str,
approved_origin: Optional[str],
*,
available: bool,
) -> Dict[str, Any]:
consented = bool(available and approved_origin == origin)
origin_changed = bool(available and approved_origin and approved_origin != origin)
return {
"available": available,
"consented": consented,
"consent_required": bool(available and not consented),
"path": _extension_sidecar_proxy_path(extension_id),
"origin_changed": origin_changed,
}
def _extension_sidecar_records(
manifest: object,
disabled_ids: Optional[Set[str]] = None,
state: Optional[Dict[str, Any]] = None,
diagnostics: Optional[Dict[str, Any]] = None,
) -> Tuple[List[Dict[str, Any]], Dict[str, Dict[str, Any]]]:
disabled_ids = disabled_ids or set()
consent_map = {}
if isinstance(state, dict) and isinstance(state.get("sidecar_proxy_consents"), dict):
consent_map = state["sidecar_proxy_consents"]
id_counts: Dict[str, int] = {}
by_id: Dict[str, Dict[str, Any]] = {}
for _source, _index, entry in _manifest_extension_entries(manifest):
raw_id = _manifest_entry_text(entry, "id")
if not _valid_extension_id(raw_id):
continue
ext_id = raw_id.strip()
id_counts[ext_id] = id_counts.get(ext_id, 0) + 1
if ext_id in by_id:
continue
manifest_enabled = entry.get("enabled", True) is not False
user_disabled = ext_id in disabled_ids
effective_enabled = manifest_enabled and not user_disabled
sidecar = _sidecar_from_manifest_entry(entry, diagnostics) if effective_enabled else None
approved_origin = consent_map.get(ext_id) if isinstance(consent_map.get(ext_id), str) else None
by_id[ext_id] = {
"id": ext_id,
"name": _manifest_entry_text(entry, "name"),
"manifest_enabled": manifest_enabled,
"user_disabled": user_disabled,
"effective_enabled": effective_enabled,
"sidecar": sidecar,
"approved_origin": approved_origin,
}
records: List[Dict[str, Any]] = []
for ext_id, item in by_id.items():
sidecar = item.get("sidecar")
if sidecar is None:
continue
available = bool(item["effective_enabled"] and id_counts.get(ext_id, 0) == 1)
proxy = _sidecar_proxy_public_status(
ext_id,
sidecar["origin"],
item.get("approved_origin"),
available=available,
)
item["duplicate_id"] = id_counts.get(ext_id, 0) > 1
item["proxy"] = proxy
if len(records) < _MAX_URL_LIST:
records.append({**sidecar, "proxy": proxy})
else:
_add_diagnostic_warning(diagnostics, "sidecar_list_truncated", _SIDECAR_WARNING_SOURCE)
break
return records, by_id
def _normalize_sidecar_proxy_path(value: object) -> Optional[str]:
if value is None:
return "/"
raw = str(value)
if raw == "":
return "/"
if raw.startswith("/"):
return None
candidate = f"/{raw}"
if not _is_valid_sidecar_proxy_path(_fully_unquote_path(candidate)):
return None
return candidate
def _extension_runtime_entries(
manifest: object, disabled_ids: Optional[Set[str]] = None
) -> List[Dict[str, object]]:
@@ -1025,7 +1197,8 @@ def _read_manifest_urls_with_diagnostics(
disabled_ids: Optional[Set[str]] = None,
manifest: Optional[object] = None,
manifest_status: Optional[Dict[str, Any]] = None,
) -> Tuple[List[str], List[str], List[Dict[str, str]], Dict[str, Any]]:
state: Optional[Dict[str, Any]] = None,
) -> Tuple[List[str], List[str], List[Dict[str, Any]], Dict[str, Any]]:
disabled_ids = disabled_ids or set()
if manifest is None or manifest_status is None:
manifest, manifest_status = _load_manifest_with_status(root, diagnostics)
@@ -1034,7 +1207,7 @@ def _read_manifest_urls_with_diagnostics(
scripts: List[str] = []
stylesheets: List[str] = []
sidecars: List[Dict[str, str]] = []
sidecars: List[Dict[str, Any]] = []
asset_base = str(manifest_status.get("_asset_base", "") or "")
entries = _iter_manifest_entries(manifest, disabled_ids=disabled_ids)
manifest_status["entry_count"] = len(entries)
@@ -1043,15 +1216,6 @@ def _read_manifest_urls_with_diagnostics(
for _source, entry in entries:
if not isinstance(entry, dict):
continue
if _source.startswith("manifest.extensions["):
sidecar = _sidecar_from_manifest_entry(entry, diagnostics)
if sidecar is not None:
if len(sidecars) < _MAX_URL_LIST:
sidecars.append(sidecar)
else:
_add_diagnostic_warning(
diagnostics, "sidecar_list_truncated", _SIDECAR_WARNING_SOURCE
)
script_source = "manifest:scripts"
stylesheet_source = "manifest:stylesheets"
if not scripts_full:
@@ -1074,6 +1238,12 @@ def _read_manifest_urls_with_diagnostics(
):
stylesheets_full = True
break
sidecars, _ = _extension_sidecar_records(
manifest,
disabled_ids=disabled_ids,
state=state,
diagnostics=diagnostics,
)
manifest_status.update(
{
"loaded": True,
@@ -1170,7 +1340,13 @@ def get_extension_status() -> Dict[str, Any]:
}
manifest, manifest_status = _load_manifest_with_status(root, diagnostics)
extension_state = _manifest_extension_state(manifest, disabled_ids, diagnostics) if manifest is not None else {
consent_ids = set((state.get("sidecar_proxy_consents") or {}).keys())
extension_state = _manifest_extension_state(
manifest,
disabled_ids,
diagnostics,
consent_ids=consent_ids,
) if manifest is not None else {
"extensions": [],
"known_ids": set(),
"manifest_disabled_ids": set(),
@@ -1181,6 +1357,7 @@ def get_extension_status() -> Dict[str, Any]:
disabled_ids=disabled_ids,
manifest=manifest,
manifest_status=manifest_status,
state=state,
)
extensions = extension_state["extensions"]
known_ids = extension_state["known_ids"]
@@ -1236,7 +1413,13 @@ def set_extension_user_enabled(extension_id: object, enabled: object) -> Dict[st
manifest, manifest_status = _load_manifest_with_status(root, diagnostics)
if manifest is None or not manifest_status.get("loaded", False):
raise ExtensionToggleError("Extension manifest is not loaded", status=409)
extension_state = _manifest_extension_state(manifest, disabled_ids, diagnostics)
consent_map = state.get("sidecar_proxy_consents") or {}
extension_state = _manifest_extension_state(
manifest,
disabled_ids,
diagnostics,
consent_ids=set(consent_map.keys()),
)
known_ids: Set[str] = extension_state["known_ids"]
manifest_disabled_ids: Set[str] = extension_state["manifest_disabled_ids"]
if ext_id not in known_ids:
@@ -1247,7 +1430,16 @@ def set_extension_user_enabled(extension_id: object, enabled: object) -> Dict[st
disabled_ids.discard(ext_id)
else:
disabled_ids.add(ext_id)
_write_extension_state({"disabled_extensions": sorted(disabled_ids)})
_write_extension_state(
{
"disabled_extensions": sorted(disabled_ids),
"sidecar_proxy_consents": {
consent_ext_id: origin
for consent_ext_id, origin in consent_map.items()
if consent_ext_id in known_ids
},
}
)
# Return a fresh status snapshot after the atomic write is visible. Keeping
# the readback outside the lock avoids doing the full manifest/status parse
# while blocking other toggles; a concurrent toggle may be reflected too,
@@ -1255,6 +1447,115 @@ def set_extension_user_enabled(extension_id: object, enabled: object) -> Dict[st
return get_extension_status()
def set_extension_sidecar_proxy_consent(extension_id: object, approved: object) -> Dict[str, Any]:
"""Persist or revoke proxy consent for the current sidecar origin."""
if not _valid_extension_id(extension_id):
raise ExtensionSidecarProxyError("Invalid extension id", status=400)
ext_id = str(extension_id).strip()
if not isinstance(approved, bool):
raise ExtensionSidecarProxyError("approved must be a boolean", status=400)
root = _extension_root()
if root is None:
raise ExtensionSidecarProxyError("Extensions are not configured", status=404)
with _EXTENSION_STATE_LOCK:
diagnostics = _new_diagnostics()
state = _load_extension_state(diagnostics)
disabled_ids = set(state.get("disabled_extensions") or [])
consent_map = dict(state.get("sidecar_proxy_consents") or {})
manifest, manifest_status = _load_manifest_with_status(root, diagnostics)
if manifest is None or not manifest_status.get("loaded", False):
raise ExtensionSidecarProxyError("Extension manifest is not loaded", status=409)
extension_state = _manifest_extension_state(
manifest,
disabled_ids,
diagnostics,
consent_ids=set(consent_map.keys()),
)
known_ids: Set[str] = extension_state["known_ids"]
if ext_id not in known_ids:
raise ExtensionSidecarProxyError("Extension not found", status=404)
_sidecars, by_id = _extension_sidecar_records(
manifest,
disabled_ids=disabled_ids,
state=state,
diagnostics=diagnostics,
)
item = by_id.get(ext_id) or {}
sidecar = item.get("sidecar")
proxy = item.get("proxy") or {}
if approved:
if sidecar is None or proxy.get("available") is not True:
raise ExtensionSidecarProxyError("Extension sidecar proxy is unavailable", status=409)
consent_map[ext_id] = sidecar["origin"]
else:
consent_map.pop(ext_id, None)
_write_extension_state(
{
"disabled_extensions": sorted(disabled_ids),
"sidecar_proxy_consents": {
consent_ext_id: origin
for consent_ext_id, origin in consent_map.items()
if consent_ext_id in known_ids
},
}
)
return get_extension_status()
def resolve_extension_sidecar_proxy_target(
extension_id: object,
proxy_path: object,
query: str = "",
) -> Dict[str, Any]:
"""Resolve the current approved sidecar proxy target for an extension."""
if not _valid_extension_id(extension_id):
raise ExtensionSidecarProxyError("Invalid extension id", status=400)
normalized_path = _normalize_sidecar_proxy_path(proxy_path)
if normalized_path is None:
raise ExtensionSidecarProxyError("Invalid sidecar proxy path", status=400)
ext_id = str(extension_id).strip()
root = _extension_root()
if root is None:
raise ExtensionSidecarProxyError("Extensions are not configured", status=404)
diagnostics = _new_diagnostics()
state = _load_extension_state(diagnostics)
disabled_ids = set(state.get("disabled_extensions") or [])
manifest, manifest_status = _load_manifest_with_status(root, diagnostics)
if manifest is None or not manifest_status.get("loaded", False):
raise ExtensionSidecarProxyError("Extension manifest is not loaded", status=409)
consent_ids = set((state.get("sidecar_proxy_consents") or {}).keys())
extension_state = _manifest_extension_state(
manifest,
disabled_ids,
diagnostics,
consent_ids=consent_ids,
)
if ext_id not in extension_state["known_ids"]:
raise ExtensionSidecarProxyError("Extension not found", status=404)
_sidecars, by_id = _extension_sidecar_records(
manifest,
disabled_ids=disabled_ids,
state=state,
diagnostics=diagnostics,
)
item = by_id.get(ext_id) or {}
sidecar = item.get("sidecar")
proxy = item.get("proxy") or {}
if sidecar is None or proxy.get("available") is not True:
raise ExtensionSidecarProxyError("Extension sidecar proxy is unavailable", status=409)
if proxy.get("consented") is not True:
raise ExtensionSidecarProxyError("Extension sidecar proxy consent required", status=403)
upstream_url = f"{sidecar['origin']}{normalized_path}"
if query:
upstream_url = f"{upstream_url}?{query}"
return {
"extension_id": ext_id,
"origin": sidecar["origin"],
"proxy_path": proxy["path"],
"upstream_url": upstream_url,
}
def _install_manifest_file() -> Path:
return _extension_state_dir() / _GALLERY_INSTALL_STATE_FILENAME
+342 -45
View File
@@ -28,7 +28,9 @@ import uuid
from collections import defaultdict
from pathlib import Path
from contextlib import closing
from urllib.parse import parse_qs, quote, urlsplit
from urllib.parse import parse_qs, quote, urljoin, urlsplit
from urllib.error import HTTPError, URLError
from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener
from api.agent_sessions import (
MESSAGING_SOURCES,
_looks_like_default_cli_title,
@@ -261,6 +263,7 @@ _CLIENT_EVENT_RATE_LIMIT_LOCK = threading.Lock()
_CLIENT_EVENT_RATE_LIMIT_WINDOW_SECONDS = 60
_CLIENT_EVENT_RATE_LIMIT_MAX = 30
_CLIENT_EVENT_MAX_BODY_BYTES = 4 * 1024
_EXTENSION_SIDECAR_PROXY_MAX_RESPONSE_BYTES = 512 * 1024
_CLIENT_EVENT_ALLOWED_FIELDS = {
"event": 64,
"source": 80,
@@ -2683,6 +2686,7 @@ from api.helpers import (
j,
t,
read_body,
MAX_BODY_BYTES,
_security_headers,
_sanitize_error,
redact_session_data,
@@ -4939,6 +4943,57 @@ def _is_browser_unsafe_request(handler) -> bool:
return bool(handler.headers.get("Origin") or handler.headers.get("Referer"))
def _check_same_origin_browser_request(handler, *, require_provenance: bool = False) -> bool:
_clear_csrf_failure_reason(handler)
origin = handler.headers.get("Origin", "")
referer = handler.headers.get("Referer", "")
host = handler.headers.get("Host", "")
sec_fetch_site = handler.headers.get("Sec-Fetch-Site", "").strip().lower()
if not (origin or referer or sec_fetch_site):
return not require_provenance or _set_csrf_failure_reason(handler, "origin_mismatch")
if sec_fetch_site == "cross-site":
return _set_csrf_failure_reason(handler, "origin_mismatch")
target = origin or referer
if not target:
if sec_fetch_site == "none":
return True
if sec_fetch_site == "same-origin":
return not require_provenance or _set_csrf_failure_reason(
handler, "origin_mismatch"
)
return _set_csrf_failure_reason(handler, "origin_mismatch")
m = _re.match(r"^https?://([^/]+)", target)
if not m:
return _set_csrf_failure_reason(handler, "origin_mismatch")
origin_host = m.group(1)
origin_scheme = m.group(0).split('://')[0].lower()
origin_name, origin_port = _normalize_host_port(origin_host)
origin_allowed = False
origin_value = m.group(0).rstrip('/').lower()
if origin_value in _allowed_public_origins():
origin_allowed = True
if not origin_allowed:
allowed_hosts = [h.strip() for h in [host] if h.strip()]
trust_forwarded_host = os.getenv("HERMES_WEBUI_TRUST_FORWARDED_HOST", "").strip().lower()
if trust_forwarded_host in ("1", "true", "yes", "on"):
allowed_hosts.extend(
h.strip()
for h in [
handler.headers.get("X-Forwarded-Host", ""),
handler.headers.get("X-Real-Host", ""),
]
if h.strip()
)
for allowed in allowed_hosts:
allowed_name, allowed_port = _normalize_host_port(allowed)
if origin_name == allowed_name and _ports_match(origin_scheme, origin_port, allowed_port):
origin_allowed = True
break
if not origin_allowed:
return _set_csrf_failure_reason(handler, "origin_mismatch")
return True
def _csrf_exempt_path(path: str) -> bool:
"""Paths that cannot or must not carry a session CSRF token."""
return path in {
@@ -4979,50 +5034,10 @@ def _csrf_rejection_error(handler) -> str:
def _check_csrf(handler) -> bool:
"""Reject cross-origin or tokenless authenticated browser unsafe requests."""
_clear_csrf_failure_reason(handler)
origin = handler.headers.get("Origin", "")
referer = handler.headers.get("Referer", "")
host = handler.headers.get("Host", "")
if not _check_same_origin_browser_request(handler):
return False
if not _is_browser_unsafe_request(handler):
return True # non-browser clients (curl, MCP, agent) have no Origin
target = origin or referer
# Extract host:port from origin/referer
m = _re.match(r"^https?://([^/]+)", target)
if not m:
return _set_csrf_failure_reason(handler, "origin_mismatch")
origin_host = m.group(1)
origin_scheme = m.group(0).split('://')[0].lower() # 'http' or 'https'
origin_name, origin_port = _normalize_host_port(origin_host)
origin_allowed = False
# Check against explicitly allowed public origins (env var)
origin_value = m.group(0).rstrip('/').lower()
if origin_value in _allowed_public_origins():
origin_allowed = True
if not origin_allowed:
# Allow same-origin Host by default. Forwarded host headers are only
# trustworthy behind a proxy that strips untrusted inbound copies.
allowed_hosts = [
h.strip()
for h in [host]
if h.strip()
]
trust_forwarded_host = os.getenv("HERMES_WEBUI_TRUST_FORWARDED_HOST", "").strip().lower()
if trust_forwarded_host in ("1", "true", "yes", "on"):
allowed_hosts.extend(
h.strip()
for h in [
handler.headers.get("X-Forwarded-Host", ""),
handler.headers.get("X-Real-Host", ""),
]
if h.strip()
)
for allowed in allowed_hosts:
allowed_name, allowed_port = _normalize_host_port(allowed)
if origin_name == allowed_name and _ports_match(origin_scheme, origin_port, allowed_port):
origin_allowed = True
break
if not origin_allowed:
return _set_csrf_failure_reason(handler, "origin_mismatch")
return True # non-browser clients (curl, MCP, agent) have no Origin/Referer
from api.auth import CSRF_HEADER_NAME, is_auth_enabled, parse_cookie, verify_csrf_token
@@ -5035,6 +5050,231 @@ def _check_csrf(handler) -> bool:
return _set_csrf_failure_reason(handler, "token_mismatch")
_EXTENSION_SIDECAR_PROXY_RE = _re.compile(
r"^/api/extensions/(?P<extension_id>[^/]+)/sidecar(?:/(?P<proxy_path>.*))?$"
)
_HOP_BY_HOP_HEADERS = {
"connection",
"keep-alive",
"proxy-connection",
"proxy-authenticate",
"proxy-authorization",
"te",
"trailer",
"transfer-encoding",
"upgrade",
}
def _connection_bound_header_names(headers) -> set[str]:
names = set(_HOP_BY_HOP_HEADERS)
if not headers or not hasattr(headers, "items"):
return names
connection_values = []
if hasattr(headers, "get_all"):
connection_values.extend(headers.get_all("Connection", []))
else:
for name, value in headers.items():
if str(name).lower() == "connection":
connection_values.append(value)
for value in connection_values:
for token in str(value).split(","):
normalized = token.strip().lower()
if normalized:
names.add(normalized)
return names
def _match_extension_sidecar_proxy_path(path: str) -> tuple[str, str] | None:
match = _EXTENSION_SIDECAR_PROXY_RE.match(path or "")
if not match:
return None
return match.group("extension_id"), match.group("proxy_path") or ""
def _read_body_bytes(handler) -> bytes:
raw_length = handler.headers.get("Content-Length", 0)
try:
length = int(raw_length)
except (TypeError, ValueError):
try:
handler.close_connection = True
except Exception:
pass
raise ValueError(f"Invalid Content-Length: {raw_length!r}") from None
if length < 0:
try:
handler.close_connection = True
except Exception:
pass
raise ValueError(f"Invalid Content-Length: {length}")
if length > MAX_BODY_BYTES:
try:
handler.close_connection = True
except Exception:
pass
raise ValueError(f"Request body too large ({length} bytes, max {MAX_BODY_BYTES})")
return handler.rfile.read(length) if length else b""
def _extension_sidecar_proxy_request_headers(handler) -> dict[str, str]:
headers = {}
raw_headers = getattr(handler, "headers", None)
if not raw_headers or not hasattr(raw_headers, "items"):
return headers
blocked_headers = _connection_bound_header_names(raw_headers)
for name, value in raw_headers.items():
lower = str(name).lower()
if (
lower in blocked_headers
or lower in {"authorization", "cookie", "content-length", "host", "origin", "referer"}
or lower.startswith("x-csrf")
):
continue
headers[str(name)] = str(value)
return headers
def _send_extension_sidecar_proxy_response(handler, status: int, body: bytes, headers) -> bool:
handler.send_response(status)
sent_content_type = False
blocked_headers = _connection_bound_header_names(headers)
if headers and hasattr(headers, "items"):
for name, value in headers.items():
lower = str(name).lower()
if lower in blocked_headers or lower in {"content-length", "set-cookie"}:
continue
if lower == "content-type":
sent_content_type = True
handler.send_header(str(name), str(value))
if not sent_content_type:
handler.send_header("Content-Type", "application/octet-stream")
handler.send_header("Content-Length", str(len(body)))
handler.send_header("Cache-Control", "no-store")
_security_headers(handler)
handler.end_headers()
handler.wfile.write(body)
return True
def _read_extension_sidecar_proxy_body(stream) -> bytes:
body = stream.read(_EXTENSION_SIDECAR_PROXY_MAX_RESPONSE_BYTES + 1)
if len(body) > _EXTENSION_SIDECAR_PROXY_MAX_RESPONSE_BYTES:
raise ValueError("Extension sidecar response too large")
return body
def _extension_sidecar_proxy_redirect_url(
allowed_origin: str,
request_url: str,
redirect_url: str,
) -> str | None:
resolved = urljoin(request_url, redirect_url or "")
allowed = urlsplit(allowed_origin or "")
parts = urlsplit(resolved)
if not allowed.scheme or not allowed.netloc or not parts.scheme or not parts.netloc:
return None
allowed_scheme = allowed.scheme.lower()
redirect_scheme = parts.scheme.lower()
if redirect_scheme != allowed_scheme:
return None
allowed_name, allowed_port = _normalize_host_port(allowed.netloc)
redirect_name, redirect_port = _normalize_host_port(parts.netloc)
if redirect_name != allowed_name or not _ports_match(
allowed_scheme,
redirect_port,
allowed_port,
):
return None
return resolved
def _extension_sidecar_proxy_same_origin_opener(allowed_origin: str):
class _SameOriginRedirectHandler(HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
resolved = _extension_sidecar_proxy_redirect_url(
allowed_origin,
req.full_url,
newurl,
)
if not resolved:
raise URLError("Extension sidecar redirect crossed declared origin")
return super().redirect_request(req, fp, code, msg, headers, resolved)
return build_opener(ProxyHandler({}), _SameOriginRedirectHandler)
def _handle_extension_sidecar_proxy(
handler,
parsed,
method: str,
*,
read_request_body: bool = False,
):
matched = _match_extension_sidecar_proxy_path(parsed.path)
if matched is None:
return False
if method == "GET" and not _check_same_origin_browser_request(
handler, require_provenance=True
):
return j(handler, {"error": _csrf_rejection_error(handler)}, status=403)
try:
request_body = _read_body_bytes(handler) if read_request_body else None
except ValueError as exc:
status = 413 if "too large" in str(exc).lower() else 400
return bad(handler, str(exc), status=status)
from api.extensions import (
ExtensionSidecarProxyError,
resolve_extension_sidecar_proxy_target,
)
extension_id, proxy_path = matched
try:
target = resolve_extension_sidecar_proxy_target(
extension_id,
proxy_path,
query=parsed.query,
)
request = Request(
target["upstream_url"],
data=request_body,
headers=_extension_sidecar_proxy_request_headers(handler),
method=method,
)
opener = _extension_sidecar_proxy_same_origin_opener(target["origin"])
with opener.open(request, timeout=10) as response:
body = _read_extension_sidecar_proxy_body(response)
return _send_extension_sidecar_proxy_response(
handler,
getattr(response, "status", 200),
body,
response.headers,
)
except ExtensionSidecarProxyError as exc:
return bad(handler, str(exc), status=exc.status)
except ValueError as exc:
return bad(handler, str(exc), status=502)
except HTTPError as exc:
try:
body = _read_extension_sidecar_proxy_body(exc)
except ValueError as read_exc:
return bad(handler, str(read_exc), status=502)
return _send_extension_sidecar_proxy_response(
handler,
exc.code,
body,
exc.headers,
)
except (TimeoutError, URLError, OSError):
logger.warning(
"extension sidecar proxy failed for %s %s",
method,
parsed.path,
exc_info=True,
)
return bad(handler, "Failed to reach extension sidecar", status=502)
def _client_ip_for_rate_limit(handler) -> str:
try:
address = getattr(handler, "client_address", None)
@@ -5196,7 +5436,7 @@ def _safe_content_length(handler, max_bytes: int) -> int:
handler.close_connection = True
except Exception:
pass
raise ValueError(f"Invalid Content-Length: {raw_length!r}")
raise ValueError(f"Invalid Content-Length: {raw_length!r}") from None
if length < 0:
try:
handler.close_connection = True
@@ -10652,6 +10892,9 @@ def _render_index_shell_base() -> str:
def handle_get(handler, parsed) -> bool:
"""Handle all GET routes. Returns True if handled, False for 404."""
proxy_result = _handle_extension_sidecar_proxy(handler, parsed, "GET")
if proxy_result is not False:
return proxy_result
if parsed.path.startswith("/session/static/"):
# Strip the leading "/session" so _serve_static() sees a path that
@@ -12430,6 +12673,16 @@ def handle_post(handler, parsed) -> bool:
finally:
if diag:
diag.finish()
proxy_result = _handle_extension_sidecar_proxy(
handler,
parsed,
"POST",
read_request_body=True,
)
if proxy_result is not False:
if diag:
diag.finish()
return proxy_result
if parsed.path == "/api/shutdown":
return _handle_shutdown(handler)
@@ -12500,6 +12753,26 @@ def handle_post(handler, parsed) -> bool:
logger.exception("extension toggle failed")
return bad(handler, "Failed to update extension state", status=500)
if parsed.path == "/api/extensions/sidecar-proxy-consent":
from api.extensions import (
ExtensionSidecarProxyError,
set_extension_sidecar_proxy_consent,
)
try:
return j(
handler,
set_extension_sidecar_proxy_consent(
body.get("id"),
body.get("approved"),
),
)
except ExtensionSidecarProxyError as exc:
return bad(handler, str(exc), status=exc.status)
except Exception:
logger.exception("extension sidecar proxy consent update failed")
return bad(handler, "Failed to update extension state", status=500)
if parsed.path == "/api/extensions/install":
from api.extensions import ExtensionInstallError, install_extension
@@ -14659,6 +14932,14 @@ def handle_patch(handler, parsed) -> bool:
"""Handle all PATCH routes. Returns True if handled, False for 404."""
if not _check_csrf(handler):
return j(handler, {"error": _csrf_rejection_error(handler)}, status=403)
proxy_result = _handle_extension_sidecar_proxy(
handler,
parsed,
"PATCH",
read_request_body=True,
)
if proxy_result is not False:
return proxy_result
body = read_body(handler)
if not _guard_request_session_visibility(handler, parsed, body=body, method="PATCH"):
return True
@@ -14679,6 +14960,14 @@ def handle_delete(handler, parsed) -> bool:
"""Handle all DELETE routes. Returns True if handled, False for 404."""
if not _check_csrf(handler):
return j(handler, {"error": _csrf_rejection_error(handler)}, status=403)
proxy_result = _handle_extension_sidecar_proxy(
handler,
parsed,
"DELETE",
read_request_body=True,
)
if proxy_result is not False:
return proxy_result
body = read_body(handler)
if not _guard_request_session_visibility(handler, parsed, body=body, method="DELETE"):
return True
@@ -14707,6 +14996,14 @@ def handle_put(handler, parsed) -> bool:
"""Handle all PUT routes. Returns True if handled, False for 404."""
if not _check_csrf(handler):
return j(handler, {"error": "Cross-origin request rejected"}, status=403)
proxy_result = _handle_extension_sidecar_proxy(
handler,
parsed,
"PUT",
read_request_body=True,
)
if proxy_result is not False:
return proxy_result
body = read_body(handler)
if not _guard_request_session_visibility(handler, parsed, body=body, method="PUT"):
return True
+17 -21
View File
@@ -44,6 +44,7 @@ Extensions cannot, by themselves:
- serve files outside the configured extension directory
- load third-party scripts/styles through the built-in injection config
- register new WebUI backend routes or proxy arbitrary sidecar/backend traffic
outside the fixed consented extension sidecar path described below
- change Hermes Agent permissions, models, memory, or tools unless they call
existing authenticated APIs that already allow those changes
@@ -150,7 +151,8 @@ manifest such as `extensions.json` keeps the existing
`desktop-companion/manifest.json` resolves relative assets under
`/extensions/desktop-companion/`.
Extension entries may also declare a read-only loopback sidecar for diagnostics:
Extension entries may also declare a loopback sidecar for diagnostics and the
opt-in proxy:
```json
{
@@ -170,9 +172,11 @@ Extension entries may also declare a read-only loopback sidecar for diagnostics:
}
```
Loopback sidecars do **not** change asset injection behavior. They are only
reported by diagnostics so an operator can see that a local companion service was
declared and optionally check its health from the browser.
Loopback sidecars do **not** change asset injection behavior. They are reported
by diagnostics so an operator can see that a local companion service was
declared and optionally check its health from the browser. If the operator later
approves the proxy in **Settings → Extensions**, WebUI may proxy requests only
through the fixed per-extension sidecar path for that extension.
Extension entries may declare browser-local settings when they also request
extension-owned storage:
@@ -270,9 +274,10 @@ URLs are ignored rather than injected.
Manifest-bundled extensions may integrate with a trusted local sidecar process,
such as a desktop companion listening on `http://127.0.0.1:17787`. The injected
extension JavaScript talks to that sidecar directly from the browser; Hermes
WebUI does not proxy those requests and does not create extension-owned backend
routes.
extension JavaScript can talk to that sidecar directly from the browser, and
WebUI diagnostics still use that direct browser path. WebUI may also proxy the
same sidecar through a fixed per-extension sidecar path after explicit persisted
user consent. WebUI does not create arbitrary extension-owned backend routes.
Loopback sidecar origins are already included in WebUI's enforced CSP
`connect-src` directive:
@@ -606,8 +611,8 @@ for operators who prefer to inspect extension state from the browser. Installed
manifest entries can be enabled or disabled from that panel through the
authenticated `POST /api/extensions/toggle` endpoint. The toggle writes only a
WebUI-managed override in the WebUI state directory; it does not edit extension
manifests, fetch new extension assets, uninstall files, proxy sidecars, or add
extension-owned backend routes. Manifest entries with `"enabled": false` remain
manifests, fetch new extension assets, uninstall files, or add extension-owned
backend routes. Manifest entries with `"enabled": false` remain
manifest-disabled and cannot be re-enabled from WebUI.
The diagnostics return the same public asset URLs that can already be injected
@@ -628,18 +633,9 @@ return `HERMES_WEBUI_EXTENSION_DIR`, resolved manifest paths, raw environment
values, rejected URL strings, rejected sidecar origins, rejected health paths, or
the override state-file path.
When sanitized loopback sidecars are present, **Settings → Extensions** renders a
read-only sidecar monitor card. The browser checks each declared `health_url`
directly with `fetch(..., { credentials: 'omit', cache: 'no-store' })` and a
short timeout. WebUI does **not** proxy sidecar requests and does not send WebUI
cookies to sidecars. A successful HTTP response is shown as healthy, a non-OK
HTTP response as unhealthy, and CORS/network/timeouts as unreachable or blocked;
raw health response bodies are never rendered. If a healthy response includes an
optional top-level `runtime` object, the panel may parse it and render only
allowlisted scalar fields such as `sidecar`, `native_host`, `bridge`,
`last_seen_at`, and `webui_origin`. This keeps sidecar-specific diagnostics
machine-readable without making WebUI depend on any one extension's private
payload shape.
When sanitized loopback sidecars are present, **Settings → Extensions** renders a sidecar monitor card. The browser checks each declared `health_url` directly with `fetch(..., { credentials: 'omit', cache: 'no-store' })` and a short timeout. A successful HTTP response is shown as healthy, a non-OK HTTP response as unhealthy, and CORS/network/timeouts as unreachable or blocked; raw health response bodies are never rendered. If a healthy response includes an optional top-level `runtime` object, the panel may parse it and render only allowlisted scalar fields such as `sidecar`, `native_host`, `bridge`, `last_seen_at`, and `webui_origin`. This keeps sidecar-specific diagnostics machine-readable without making WebUI depend on any one extension's private payload shape.
The same card also exposes proxy consent through `POST /api/extensions/sidecar-proxy-consent` and reports the fixed per-extension sidecar path `/api/extensions/<extension-id>/sidecar/<relative-path>`. WebUI strips `Cookie`, `Authorization`, and CSRF headers before contacting the sidecar, and sidecar `Set-Cookie` headers are stripped before the browser sees the response. WebUI does not create arbitrary extension-owned backend routes; the proxy surface stays on that fixed per-extension sidecar path.
When sanitized settings are present, Settings -> Extensions renders
browser-local controls for installed manifest entries. Save, reset, and clear
+46
View File
@@ -8809,6 +8809,22 @@ function _extensionSidecarCard(sidecars){
const origin=(sidecar&&sidecar.origin)||'';
const healthPath=(sidecar&&sidecar.health_path)||'';
const healthUrl=(sidecar&&sidecar.health_url)||'';
const proxy=(sidecar&&sidecar.proxy&&typeof sidecar.proxy==='object')?sidecar.proxy:{};
const proxyAvailable=proxy.available===true;
const proxyConsented=proxy.consented===true;
const proxyConsentRequired=proxy.consent_required===true;
const proxyOriginChanged=proxy.origin_changed===true;
const proxyPath=(proxy&&proxy.path)||'';
const proxyStatus=proxyConsented
?'consented'
:(proxyOriginChanged
?'reconfirm required'
:(proxyConsentRequired
?'approval required'
:'unavailable'));
const proxyButton=(proxyAvailable&&id)
?`<button class="sm-btn extension-toggle-btn" type="button" data-extension-sidecar-proxy-id="${esc(id)}" data-extension-sidecar-proxy-approved="${proxyConsented?'false':'true'}">${esc(proxyConsented?'Revoke proxy consent':'Approve proxy consent')}</button>`
:'';
return `<div class="extension-sidecar-row" data-sidecar-index="${index}">
<div class="extension-sidecar-row-head">
<div class="extension-sidecar-title">${esc(title)}</div>
@@ -8819,7 +8835,10 @@ function _extensionSidecarCard(sidecars){
<div><span>Origin</span><code>${esc(origin)}</code></div>
<div><span>Health path</span><code>${esc(healthPath)}</code></div>
<div><span>Health URL</span><code>${esc(healthUrl)}</code></div>
<div><span>Proxy</span><code>${esc(proxyStatus)}</code></div>
<div><span>Proxy path</span><code>${esc(proxyPath)}</code></div>
</div>
<div class="extension-sidecar-actions">${proxyButton}</div>
<div class="extension-sidecar-runtime" data-sidecar-runtime-index="${index}" hidden></div>
</div>`;
}).join('')}</div>`:'<div class="extension-url-empty">No loopback sidecars declared.</div>';
@@ -8983,6 +9002,7 @@ function _renderExtensionsPanel(data,seq){
</div>
`;
_bindExtensionToggleButtons(target);
_bindExtensionSidecarProxyButtons(target);
_bindExtensionSettingsButtons(target);
_monitorExtensionSidecars(sidecars,seq);
}
@@ -8994,6 +9014,13 @@ function _bindExtensionToggleButtons(root){
});
}
function _bindExtensionSidecarProxyButtons(root){
if(!root) return;
root.querySelectorAll('[data-extension-sidecar-proxy-id]').forEach(btn=>{
btn.addEventListener('click',()=>handleExtensionSidecarProxyConsent(btn));
});
}
async function handleExtensionToggle(btn){
if(!btn||btn.disabled) return;
const id=btn.dataset.extensionToggleId||'';
@@ -9013,6 +9040,25 @@ async function handleExtensionToggle(btn){
}
}
async function handleExtensionSidecarProxyConsent(btn){
if(!btn||btn.disabled) return;
const id=btn.dataset.extensionSidecarProxyId||'';
const approved=btn.dataset.extensionSidecarProxyApproved==='true';
if(!id) return;
const previousText=btn.textContent;
btn.disabled=true;
btn.textContent=approved?'Approving…':'Revoking…';
try{
const data=await api('/api/extensions/sidecar-proxy-consent',{method:'POST',body:JSON.stringify({id,approved})});
showToast(approved?'Extension sidecar proxy approved.':'Extension sidecar proxy consent revoked.');
_renderExtensionsPanel(data,++_extensionsSidecarMonitorSeq);
}catch(e){
btn.disabled=false;
btn.textContent=previousText;
showToast('Failed to update extension sidecar proxy consent: '+(e&&e.message?e.message:String(e)));
}
}
function _readExtensionSettingsForm(row){
const values={};
row.querySelectorAll('[data-extension-setting-input]').forEach(input=>{
+931
View File
@@ -0,0 +1,931 @@
"""Focused tests for the extension sidecar proxy contract."""
from types import SimpleNamespace
import io
import json
import urllib.request
import pytest
class FakeHandler:
def __init__(self, body: bytes = b""):
self.status = None
self.headers = {}
self.sent_headers = []
self.body = bytearray()
self.wfile = self
self.rfile = io.BytesIO(body)
def send_response(self, status):
self.status = status
def send_header(self, name, value):
self.sent_headers.append((name, value))
def end_headers(self):
pass
def write(self, data):
self.body.extend(data)
def header(self, name):
for key, value in self.sent_headers:
if key.lower() == name.lower():
return value
return None
@pytest.fixture(autouse=True)
def _clear_extension_env(monkeypatch):
from api import auth as auth_mod
for name in (
"HERMES_WEBUI_EXTENSION_DIR",
"HERMES_WEBUI_EXTENSION_MANIFEST",
"HERMES_WEBUI_PASSWORD",
):
monkeypatch.delenv(name, raising=False)
auth_mod._invalidate_password_hash_cache()
yield
auth_mod._invalidate_password_hash_cache()
def _use_extension_state_dir(monkeypatch, tmp_path):
tmp_path.mkdir(parents=True, exist_ok=True)
state_dir = tmp_path / "webui-state"
state_dir.mkdir()
monkeypatch.setenv("HERMES_WEBUI_STATE_DIR", str(state_dir))
import api.extensions as extensions
monkeypatch.setattr(extensions, "_extension_state_dir", lambda: state_dir)
return state_dir
def _write_manifest(root, payload):
(root / "extensions.json").write_text(json.dumps(payload), encoding="utf-8")
def _configure_manifest_extension(monkeypatch, tmp_path, payload):
state_dir = _use_extension_state_dir(monkeypatch, tmp_path)
root = tmp_path / "extensions"
root.mkdir(parents=True, exist_ok=True)
_write_manifest(root, payload)
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_DIR", str(root))
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_MANIFEST", "extensions.json")
return state_dir, root
def test_extension_sidecar_proxy_requires_webui_auth(monkeypatch):
monkeypatch.setenv("HERMES_WEBUI_PASSWORD", "test-password")
from api.auth import check_auth
handler = FakeHandler()
assert check_auth(
handler,
SimpleNamespace(path="/api/extensions/templates/sidecar/health", query=""),
) is False
assert handler.status == 401
assert handler.header("Location") is None
def test_extension_sidecar_proxy_requires_consent_and_reconfirms_after_origin_change(
tmp_path, monkeypatch
):
state_dir, root = _configure_manifest_extension(
monkeypatch,
tmp_path,
{
"extensions": [
{
"id": "templates",
"sidecar": {
"type": "loopback",
"origin": "http://127.0.0.1:17787",
},
}
]
},
)
from api.extensions import (
ExtensionSidecarProxyError,
resolve_extension_sidecar_proxy_target,
set_extension_sidecar_proxy_consent,
)
with pytest.raises(ExtensionSidecarProxyError) as unapproved:
resolve_extension_sidecar_proxy_target("templates", "v1/ping", "debug=1")
assert unapproved.value.status == 403
approved = set_extension_sidecar_proxy_consent("templates", True)
assert approved["sidecars"][0]["proxy"]["consented"] is True
assert json.loads((state_dir / "extension-overrides.json").read_text(encoding="utf-8")) == {
"version": 1,
"disabled_extensions": [],
"sidecar_proxy_consents": {
"templates": "http://127.0.0.1:17787",
},
}
target = resolve_extension_sidecar_proxy_target("templates", "v1/ping", "debug=1")
assert target == {
"extension_id": "templates",
"origin": "http://127.0.0.1:17787",
"proxy_path": "/api/extensions/templates/sidecar/",
"upstream_url": "http://127.0.0.1:17787/v1/ping?debug=1",
}
encoded_target = resolve_extension_sidecar_proxy_target("templates", "v1%2Fprivate")
assert encoded_target["upstream_url"] == "http://127.0.0.1:17787/v1%2Fprivate"
_write_manifest(
root,
{
"extensions": [
{
"id": "templates",
"sidecar": {
"type": "loopback",
"origin": "http://127.0.0.1:17788",
},
}
]
},
)
changed = set_extension_sidecar_proxy_consent("templates", False)
assert changed["sidecars"][0]["proxy"]["origin_changed"] is False
with pytest.raises(ExtensionSidecarProxyError) as changed_origin:
resolve_extension_sidecar_proxy_target("templates", "v1/ping")
assert changed_origin.value.status == 403
def test_extension_sidecar_proxy_rejects_unavailable_surfaces(tmp_path, monkeypatch):
from api.extensions import ExtensionSidecarProxyError, resolve_extension_sidecar_proxy_target
_configure_manifest_extension(
monkeypatch,
tmp_path / "duplicate",
{
"extensions": [
{
"id": "templates",
"sidecar": {
"type": "loopback",
"origin": "http://127.0.0.1:17787",
},
},
{
"id": "templates",
"sidecar": {
"type": "loopback",
"origin": "http://127.0.0.1:17788",
},
},
]
},
)
with pytest.raises(ExtensionSidecarProxyError) as duplicate:
resolve_extension_sidecar_proxy_target("templates", "v1/ping")
assert duplicate.value.status == 409
_configure_manifest_extension(
monkeypatch,
tmp_path / "manifest_disabled",
{
"extensions": [
{
"id": "templates",
"enabled": False,
"sidecar": {
"type": "loopback",
"origin": "http://127.0.0.1:17787",
},
}
]
},
)
with pytest.raises(ExtensionSidecarProxyError) as manifest_disabled:
resolve_extension_sidecar_proxy_target("templates", "v1/ping")
assert manifest_disabled.value.status == 409
state_dir, _root = _configure_manifest_extension(
monkeypatch,
tmp_path / "user_disabled",
{
"extensions": [
{
"id": "templates",
"sidecar": {
"type": "loopback",
"origin": "http://127.0.0.1:17787",
},
}
]
},
)
(state_dir / "extension-overrides.json").write_text(
json.dumps(
{
"version": 1,
"disabled_extensions": ["templates"],
"sidecar_proxy_consents": {
"templates": "http://127.0.0.1:17787",
},
}
),
encoding="utf-8",
)
with pytest.raises(ExtensionSidecarProxyError) as user_disabled:
resolve_extension_sidecar_proxy_target("templates", "v1/ping")
assert user_disabled.value.status == 409
_configure_manifest_extension(
monkeypatch,
tmp_path / "unsupported",
{
"extensions": [
{
"id": "templates",
"sidecar": {
"type": "unix-socket",
"origin": "http://127.0.0.1:17787",
},
}
]
},
)
with pytest.raises(ExtensionSidecarProxyError) as unsupported:
resolve_extension_sidecar_proxy_target("templates", "v1/ping")
assert unsupported.value.status == 409
def test_extension_sidecar_proxy_malformed_consents_fail_closed(tmp_path, monkeypatch):
from api.extensions import ExtensionSidecarProxyError, resolve_extension_sidecar_proxy_target
state_dir, _root = _configure_manifest_extension(
monkeypatch,
tmp_path / "malformed_consents",
{
"extensions": [
{
"id": "templates",
"sidecar": {
"type": "loopback",
"origin": "http://127.0.0.1:17787",
},
}
]
},
)
(state_dir / "extension-overrides.json").write_text(
json.dumps(
{
"version": 1,
"disabled_extensions": [],
"sidecar_proxy_consents": {
"templates": "http://127.0.0.1:17787",
"../bad": "http://127.0.0.1:17788",
},
}
),
encoding="utf-8",
)
with pytest.raises(ExtensionSidecarProxyError) as exc:
resolve_extension_sidecar_proxy_target("templates", "v1/ping")
assert exc.value.status == 403
def test_extension_sidecar_proxy_route_uses_shared_resolver_and_strips_headers(monkeypatch):
from api import routes
captured = {}
class FakeResponse:
def __init__(self):
self.status = 202
self.headers = {
"Content-Type": "application/json",
"Set-Cookie": "sidecar=1",
"Connection": "close, X-Upstream-Hop",
"X-Sidecar": "ok",
"X-Upstream-Hop": "strip-me",
}
def read(self, *_args):
return b'{"ok":true}'
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
class FakeOpener:
def open(self, request, timeout=10):
captured["url"] = request.full_url
captured["method"] = request.get_method()
captured["data"] = request.data
captured["headers"] = {k.lower(): v for k, v in request.header_items()}
captured["timeout"] = timeout
return FakeResponse()
monkeypatch.setattr(routes, "_check_csrf", lambda handler: True)
monkeypatch.setattr(
"api.extensions.resolve_extension_sidecar_proxy_target",
lambda extension_id, proxy_path, query="": {
"extension_id": extension_id,
"origin": "http://127.0.0.1:17787",
"proxy_path": "/api/extensions/templates/sidecar/",
"upstream_url": f"http://127.0.0.1:17787/{proxy_path}?{query}",
},
)
monkeypatch.setattr(
routes,
"_extension_sidecar_proxy_same_origin_opener",
lambda allowed_origin: FakeOpener(),
)
raw_body = b'{"ping":"pong"}'
handler = FakeHandler(raw_body)
handler.headers = {
"Accept": "application/json",
"If-None-Match": '"abc123"',
"Range": "bytes=0-64",
"Content-Type": "application/json",
"Content-Length": str(len(raw_body)),
"Cookie": "webui=secret",
"Authorization": "Bearer secret",
"Host": "webui.local",
"Origin": "http://webui.local",
"Referer": "http://webui.local/settings",
"X-CSRF-Token": "secret",
"X-Sidecar-Auth": "local-token",
"Connection": "keep-alive, X-Client-Hop",
"X-Client-Hop": "strip-me",
}
result = routes.handle_post(
handler,
SimpleNamespace(path="/api/extensions/templates/sidecar/v1/ping", query="debug=1"),
)
assert result is True
assert captured == {
"url": "http://127.0.0.1:17787/v1/ping?debug=1",
"method": "POST",
"data": raw_body,
"headers": {
"accept": "application/json",
"content-type": "application/json",
"if-none-match": '"abc123"',
"range": "bytes=0-64",
"x-sidecar-auth": "local-token",
},
"timeout": 10,
}
assert handler.status == 202
assert handler.body == b'{"ok":true}'
assert handler.header("Content-Type") == "application/json"
assert handler.header("X-Sidecar") == "ok"
assert handler.header("Set-Cookie") is None
assert handler.header("Connection") is None
assert handler.header("X-Upstream-Hop") is None
def test_extension_sidecar_proxy_get_rejects_cross_site_browser_request(monkeypatch):
from api import routes
monkeypatch.setattr(
"api.extensions.resolve_extension_sidecar_proxy_target",
lambda extension_id, proxy_path, query="": {
"extension_id": extension_id,
"origin": "http://127.0.0.1:17787",
"proxy_path": "/api/extensions/templates/sidecar/",
"upstream_url": "http://127.0.0.1:17787/v1/ping",
},
)
handler = FakeHandler()
handler.headers = {
"Origin": "https://evil.example",
"Host": "webui.local",
"Sec-Fetch-Site": "cross-site",
}
result = routes.handle_get(
handler,
SimpleNamespace(path="/api/extensions/templates/sidecar/v1/ping", query=""),
)
assert result is None
assert handler.status == 403
assert json.loads(handler.body.decode("utf-8")) == {
"error": "Cross-origin mismatch - check reverse proxy headers"
}
def test_extension_sidecar_proxy_get_requires_browser_provenance(monkeypatch):
from api import routes
monkeypatch.setattr(
"api.extensions.resolve_extension_sidecar_proxy_target",
lambda extension_id, proxy_path, query="": {
"extension_id": extension_id,
"origin": "http://127.0.0.1:17787",
"proxy_path": "/api/extensions/templates/sidecar/",
"upstream_url": "http://127.0.0.1:17787/v1/ping",
},
)
handler = FakeHandler()
result = routes.handle_get(
handler,
SimpleNamespace(path="/api/extensions/templates/sidecar/v1/ping", query=""),
)
assert result is None
assert handler.status == 403
assert json.loads(handler.body.decode("utf-8")) == {
"error": "Cross-origin mismatch - check reverse proxy headers"
}
def test_extension_sidecar_proxy_get_allows_same_origin_browser_request_without_csrf_token(monkeypatch):
from api import routes
class FakeResponse:
def __init__(self):
self.status = 200
self.headers = {"Content-Type": "application/json"}
def read(self, *_args):
return b'{"ok":true}'
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
class FakeOpener:
def open(self, request, timeout=10):
return FakeResponse()
monkeypatch.setattr(
"api.extensions.resolve_extension_sidecar_proxy_target",
lambda extension_id, proxy_path, query="": {
"extension_id": extension_id,
"origin": "http://127.0.0.1:17787",
"proxy_path": "/api/extensions/templates/sidecar/",
"upstream_url": "http://127.0.0.1:17787/v1/ping",
},
)
monkeypatch.setattr(
routes,
"_extension_sidecar_proxy_same_origin_opener",
lambda allowed_origin: FakeOpener(),
)
handler = FakeHandler()
handler.headers = {
"Origin": "http://webui.local",
"Host": "webui.local",
"Sec-Fetch-Site": "same-origin",
}
result = routes.handle_get(
handler,
SimpleNamespace(path="/api/extensions/templates/sidecar/v1/ping", query=""),
)
assert result is True
assert handler.status == 200
assert json.loads(handler.body.decode("utf-8")) == {"ok": True}
def test_extension_sidecar_proxy_route_preserves_upstream_http_errors(monkeypatch):
from api import routes
from urllib.error import HTTPError
monkeypatch.setattr(
"api.extensions.resolve_extension_sidecar_proxy_target",
lambda extension_id, proxy_path, query="": {
"extension_id": extension_id,
"origin": "http://127.0.0.1:17787",
"proxy_path": "/api/extensions/templates/sidecar/",
"upstream_url": "http://127.0.0.1:17787/v1/ping",
},
)
class ErrorHeaders(dict):
pass
error = HTTPError(
"http://127.0.0.1:17787/v1/ping",
418,
"teapot",
ErrorHeaders({"Content-Type": "text/plain", "Set-Cookie": "drop=1"}),
io.BytesIO(b"sidecar said no"),
)
class FakeOpener:
def open(self, request, timeout=10):
raise error
monkeypatch.setattr(
routes,
"_extension_sidecar_proxy_same_origin_opener",
lambda allowed_origin: FakeOpener(),
)
handler = FakeHandler()
handler.headers = {
"Origin": "http://webui.local",
"Host": "webui.local",
"Sec-Fetch-Site": "same-origin",
}
result = routes.handle_get(
handler,
SimpleNamespace(path="/api/extensions/templates/sidecar/v1/ping", query=""),
)
assert result is True
assert handler.status == 418
assert handler.body == b"sidecar said no"
assert handler.header("Content-Type") == "text/plain"
assert handler.header("Set-Cookie") is None
def test_extension_sidecar_proxy_get_rejects_sec_fetch_same_origin_without_origin(monkeypatch):
from api import routes
monkeypatch.setattr(
"api.extensions.resolve_extension_sidecar_proxy_target",
lambda extension_id, proxy_path, query="": {
"extension_id": extension_id,
"origin": "http://127.0.0.1:17787",
"proxy_path": "/api/extensions/templates/sidecar/",
"upstream_url": "http://127.0.0.1:17787/v1/ping",
},
)
handler = FakeHandler()
handler.headers = {
"Host": "webui.local",
"Sec-Fetch-Site": "same-origin",
}
result = routes.handle_get(
handler,
SimpleNamespace(path="/api/extensions/templates/sidecar/v1/ping", query=""),
)
assert result is None
assert handler.status == 403
assert json.loads(handler.body.decode("utf-8")) == {
"error": "Cross-origin mismatch - check reverse proxy headers"
}
def test_extension_sidecar_proxy_get_allows_top_level_navigation_provenance(monkeypatch):
from api import routes
class FakeResponse:
def __init__(self):
self.status = 200
self.headers = {"Content-Type": "application/json"}
def read(self, *_args):
return b'{"ok":true}'
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
class FakeOpener:
def open(self, request, timeout=10):
return FakeResponse()
monkeypatch.setattr(
"api.extensions.resolve_extension_sidecar_proxy_target",
lambda extension_id, proxy_path, query="": {
"extension_id": extension_id,
"origin": "http://127.0.0.1:17787",
"proxy_path": "/api/extensions/templates/sidecar/",
"upstream_url": "http://127.0.0.1:17787/v1/ping",
},
)
monkeypatch.setattr(
routes,
"_extension_sidecar_proxy_same_origin_opener",
lambda allowed_origin: FakeOpener(),
)
handler = FakeHandler()
handler.headers = {
"Host": "webui.local",
"Sec-Fetch-Site": "none",
}
result = routes.handle_get(
handler,
SimpleNamespace(path="/api/extensions/templates/sidecar/v1/ping", query=""),
)
assert result is True
assert handler.status == 200
assert json.loads(handler.body.decode("utf-8")) == {"ok": True}
def test_extension_sidecar_proxy_route_rejects_oversized_upstream_response(monkeypatch):
from api import routes
monkeypatch.setattr(
"api.extensions.resolve_extension_sidecar_proxy_target",
lambda extension_id, proxy_path, query="": {
"extension_id": extension_id,
"origin": "http://127.0.0.1:17787",
"proxy_path": "/api/extensions/templates/sidecar/",
"upstream_url": "http://127.0.0.1:17787/v1/ping",
},
)
class FakeResponse:
def __init__(self):
self.status = 200
self.headers = {"Content-Type": "application/octet-stream"}
def read(self, size=-1):
assert size == routes._EXTENSION_SIDECAR_PROXY_MAX_RESPONSE_BYTES + 1
return b"x" * size
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
class FakeOpener:
def open(self, request, timeout=10):
return FakeResponse()
monkeypatch.setattr(
routes,
"_extension_sidecar_proxy_same_origin_opener",
lambda allowed_origin: FakeOpener(),
)
handler = FakeHandler()
handler.headers = {
"Origin": "http://webui.local",
"Host": "webui.local",
"Sec-Fetch-Site": "same-origin",
}
result = routes.handle_get(
handler,
SimpleNamespace(path="/api/extensions/templates/sidecar/v1/ping", query=""),
)
assert result is None
assert handler.status == 502
assert json.loads(handler.body.decode("utf-8")) == {
"error": "Extension sidecar response too large"
}
def test_extension_sidecar_proxy_route_rejects_oversized_upstream_http_error(monkeypatch):
from api import routes
from urllib.error import HTTPError
monkeypatch.setattr(
"api.extensions.resolve_extension_sidecar_proxy_target",
lambda extension_id, proxy_path, query="": {
"extension_id": extension_id,
"origin": "http://127.0.0.1:17787",
"proxy_path": "/api/extensions/templates/sidecar/",
"upstream_url": "http://127.0.0.1:17787/v1/ping",
},
)
class OversizedBody(io.BytesIO):
def read(self, size=-1):
assert size == routes._EXTENSION_SIDECAR_PROXY_MAX_RESPONSE_BYTES + 1
return b"x" * size
error = HTTPError(
"http://127.0.0.1:17787/v1/ping",
502,
"bad gateway",
{},
OversizedBody(),
)
class FakeOpener:
def open(self, request, timeout=10):
raise error
monkeypatch.setattr(
routes,
"_extension_sidecar_proxy_same_origin_opener",
lambda allowed_origin: FakeOpener(),
)
handler = FakeHandler()
handler.headers = {
"Origin": "http://webui.local",
"Host": "webui.local",
"Sec-Fetch-Site": "same-origin",
}
result = routes.handle_get(
handler,
SimpleNamespace(path="/api/extensions/templates/sidecar/v1/ping", query=""),
)
assert result is None
assert handler.status == 502
assert json.loads(handler.body.decode("utf-8")) == {
"error": "Extension sidecar response too large"
}
def test_extension_sidecar_proxy_route_returns_sanitized_502(monkeypatch):
from api import routes
monkeypatch.setattr(
"api.extensions.resolve_extension_sidecar_proxy_target",
lambda extension_id, proxy_path, query="": {
"extension_id": extension_id,
"origin": "http://127.0.0.1:17787",
"proxy_path": "/api/extensions/templates/sidecar/",
"upstream_url": "http://127.0.0.1:17787/v1/ping",
},
)
class FakeOpener:
def open(self, request, timeout=10):
raise OSError("no route")
monkeypatch.setattr(
routes,
"_extension_sidecar_proxy_same_origin_opener",
lambda allowed_origin: FakeOpener(),
)
handler = FakeHandler()
handler.headers = {
"Origin": "http://webui.local",
"Host": "webui.local",
"Sec-Fetch-Site": "same-origin",
}
result = routes.handle_get(
handler,
SimpleNamespace(path="/api/extensions/templates/sidecar/v1/ping", query=""),
)
assert result is None
assert handler.status == 502
assert json.loads(handler.body.decode("utf-8")) == {
"error": "Failed to reach extension sidecar"
}
def test_extension_sidecar_proxy_redirect_guard_preserves_origin_only():
from api import routes
assert routes._extension_sidecar_proxy_redirect_url(
"http://127.0.0.1:17787",
"http://127.0.0.1:17787/v1/ping",
"/v1/next?debug=1",
) == "http://127.0.0.1:17787/v1/next?debug=1"
assert routes._extension_sidecar_proxy_redirect_url(
"http://127.0.0.1:17787",
"http://127.0.0.1:17787/v1/ping",
"http://evil.example/steal",
) is None
assert routes._extension_sidecar_proxy_redirect_url(
"http://127.0.0.1:17787",
"http://127.0.0.1:17787/v1/ping",
"http://127.0.0.1:17788/other-port",
) is None
assert routes._extension_sidecar_proxy_redirect_url(
"http://localhost",
"http://localhost/v1/ping",
"http://LOCALHOST:80/v1/next",
) == "http://LOCALHOST:80/v1/next"
assert routes._extension_sidecar_proxy_redirect_url(
"https://localhost:443",
"https://localhost/v1/ping",
"https://localhost/v1/next",
) == "https://localhost/v1/next"
def test_extension_sidecar_proxy_route_uses_same_origin_redirect_opener(monkeypatch):
from api import routes
captured = {}
class FakeResponse:
status = 200
headers = {"Content-Type": "application/json"}
def read(self, *_args):
return b'{"ok":true}'
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
class FakeOpener:
def open(self, request, timeout=10):
captured["url"] = request.full_url
captured["timeout"] = timeout
return FakeResponse()
monkeypatch.setattr(
"api.extensions.resolve_extension_sidecar_proxy_target",
lambda extension_id, proxy_path, query="": {
"extension_id": extension_id,
"origin": "http://127.0.0.1:17787",
"proxy_path": "/api/extensions/templates/sidecar/",
"upstream_url": "http://127.0.0.1:17787/v1/ping",
},
)
monkeypatch.setattr(
routes,
"_extension_sidecar_proxy_same_origin_opener",
lambda allowed_origin: (
captured.__setitem__("allowed_origin", allowed_origin),
FakeOpener(),
)[1],
)
handler = FakeHandler()
handler.headers = {
"Origin": "http://webui.local",
"Host": "webui.local",
"Sec-Fetch-Site": "same-origin",
}
result = routes.handle_get(
handler,
SimpleNamespace(path="/api/extensions/templates/sidecar/v1/ping", query=""),
)
assert result is True
assert captured == {
"allowed_origin": "http://127.0.0.1:17787",
"url": "http://127.0.0.1:17787/v1/ping",
"timeout": 10,
}
assert handler.status == 200
assert handler.body == b'{"ok":true}'
def test_extension_sidecar_proxy_opener_disables_ambient_proxies(monkeypatch):
from api import routes
captured = {}
def fake_build_opener(*handlers):
captured["handlers"] = handlers
return object()
monkeypatch.setattr(routes, "build_opener", fake_build_opener)
opener = routes._extension_sidecar_proxy_same_origin_opener("http://127.0.0.1:17787")
assert opener is not None
proxy_handlers = [
handler
for handler in captured["handlers"]
if isinstance(handler, urllib.request.ProxyHandler)
]
assert len(proxy_handlers) == 1
assert proxy_handlers[0].proxies == {}
def test_extension_sidecar_proxy_consent_route_is_wired(monkeypatch):
from api import routes
captured = {}
def fake_j(handler, data, status=200, headers=None):
captured["data"] = data
captured["status"] = status
return True
monkeypatch.setattr(routes, "_check_csrf", lambda handler: True)
monkeypatch.setattr(routes, "read_body", lambda handler: {"id": "templates", "approved": True})
monkeypatch.setattr(routes, "j", fake_j)
monkeypatch.setattr(
"api.extensions.set_extension_sidecar_proxy_consent",
lambda extension_id, approved: {
"ok": True,
"id": extension_id,
"approved": approved,
},
)
handler = FakeHandler()
assert routes.handle_post(
handler,
SimpleNamespace(path="/api/extensions/sidecar-proxy-consent"),
) is True
assert captured == {
"status": 200,
"data": {"ok": True, "id": "templates", "approved": True},
}
+158 -1
View File
@@ -515,6 +515,13 @@ def test_extension_status_reports_sanitized_loopback_sidecars(tmp_path, monkeypa
"origin": "http://127.0.0.1:17787",
"health_path": "/health",
"health_url": "http://127.0.0.1:17787/health",
"proxy": {
"available": True,
"consented": False,
"consent_required": True,
"path": "/api/extensions/desktop-companion/sidecar/",
"origin_changed": False,
},
},
{
"id": "implicit-health",
@@ -523,6 +530,13 @@ def test_extension_status_reports_sanitized_loopback_sidecars(tmp_path, monkeypa
"origin": "http://localhost:17788",
"health_path": "/health",
"health_url": "http://localhost:17788/health",
"proxy": {
"available": True,
"consented": False,
"consent_required": True,
"path": "/api/extensions/implicit-health/sidecar/",
"origin_changed": False,
},
},
{
"id": "ipv6-loopback",
@@ -531,6 +545,13 @@ def test_extension_status_reports_sanitized_loopback_sidecars(tmp_path, monkeypa
"origin": "http://[::1]:17789",
"health_path": "/ready",
"health_url": "http://[::1]:17789/ready",
"proxy": {
"available": True,
"consented": False,
"consent_required": True,
"path": "/api/extensions/ipv6-loopback/sidecar/",
"origin_changed": False,
},
},
]
assert status["counts"]["sidecars"] == 3
@@ -538,6 +559,79 @@ def test_extension_status_reports_sanitized_loopback_sidecars(tmp_path, monkeypa
assert status["warnings"] == []
def test_extension_status_reports_sidecar_proxy_consent_and_origin_change(tmp_path, monkeypatch):
state_dir = _use_extension_state_dir(monkeypatch, tmp_path)
root = tmp_path / "extensions"
root.mkdir()
manifest_file = root / "extensions.json"
manifest_file.write_text(
json.dumps(
{
"extensions": [
{
"id": "desktop-companion",
"sidecar": {
"type": "loopback",
"origin": "http://127.0.0.1:17787",
},
}
]
}
),
encoding="utf-8",
)
(state_dir / "extension-overrides.json").write_text(
json.dumps(
{
"version": 1,
"disabled_extensions": [],
"sidecar_proxy_consents": {
"desktop-companion": "http://127.0.0.1:17787",
},
}
),
encoding="utf-8",
)
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_DIR", str(root))
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_MANIFEST", "extensions.json")
from api.extensions import get_extension_status
current = get_extension_status()
assert current["sidecars"][0]["proxy"] == {
"available": True,
"consented": True,
"consent_required": False,
"path": "/api/extensions/desktop-companion/sidecar/",
"origin_changed": False,
}
manifest_file.write_text(
json.dumps(
{
"extensions": [
{
"id": "desktop-companion",
"sidecar": {
"type": "loopback",
"origin": "http://127.0.0.1:17788",
},
}
]
}
),
encoding="utf-8",
)
changed = get_extension_status()
assert changed["sidecars"][0]["proxy"] == {
"available": True,
"consented": False,
"consent_required": True,
"path": "/api/extensions/desktop-companion/sidecar/",
"origin_changed": True,
}
def test_extension_status_skips_disabled_sidecar_entries(tmp_path, monkeypatch):
root = tmp_path / "extensions"
root.mkdir()
@@ -874,6 +968,7 @@ def test_set_extension_user_enabled_persists_override_and_reenables(tmp_path, mo
assert json.loads((state_dir / "extension-overrides.json").read_text(encoding="utf-8")) == {
"version": 1,
"disabled_extensions": ["templates"],
"sidecar_proxy_consents": {},
}
enabled = set_extension_user_enabled("templates", True)
@@ -882,6 +977,66 @@ def test_set_extension_user_enabled_persists_override_and_reenables(tmp_path, mo
assert json.loads((state_dir / "extension-overrides.json").read_text(encoding="utf-8")) == {
"version": 1,
"disabled_extensions": [],
"sidecar_proxy_consents": {},
}
def test_set_extension_user_enabled_preserves_sidecar_proxy_consent(tmp_path, monkeypatch):
state_dir = _use_extension_state_dir(monkeypatch, tmp_path)
root = tmp_path / "extensions"
root.mkdir()
(root / "extensions.json").write_text(
json.dumps(
{
"extensions": [
{
"id": "templates",
"scripts": ["templates.js"],
"sidecar": {
"type": "loopback",
"origin": "http://127.0.0.1:17787",
},
}
]
}
),
encoding="utf-8",
)
(state_dir / "extension-overrides.json").write_text(
json.dumps(
{
"version": 1,
"disabled_extensions": [],
"sidecar_proxy_consents": {
"templates": "http://127.0.0.1:17787",
},
}
),
encoding="utf-8",
)
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_DIR", str(root))
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_MANIFEST", "extensions.json")
from api.extensions import set_extension_user_enabled
disabled = set_extension_user_enabled("templates", False)
assert disabled["sidecars"] == []
assert json.loads((state_dir / "extension-overrides.json").read_text(encoding="utf-8")) == {
"version": 1,
"disabled_extensions": ["templates"],
"sidecar_proxy_consents": {
"templates": "http://127.0.0.1:17787",
},
}
enabled = set_extension_user_enabled("templates", True)
assert enabled["sidecars"][0]["proxy"]["consented"] is True
assert json.loads((state_dir / "extension-overrides.json").read_text(encoding="utf-8")) == {
"version": 1,
"disabled_extensions": [],
"sidecar_proxy_consents": {
"templates": "http://127.0.0.1:17787",
},
}
@@ -1003,7 +1158,7 @@ def test_extension_state_recursion_error_fails_safe(tmp_path, monkeypatch):
monkeypatch.setattr(extensions.json, "loads", raise_recursion_error)
state = extensions._load_extension_state({"warnings": []})
assert state == {"version": 1, "disabled_extensions": []}
assert state == {"version": 1, "disabled_extensions": [], "sidecar_proxy_consents": {}}
diagnostics = {"warnings": []}
extensions._load_extension_state(diagnostics)
@@ -1063,6 +1218,7 @@ def test_set_extension_user_enabled_is_idempotent_and_preserves_stale_ids(tmp_pa
assert json.loads((state_dir / "extension-overrides.json").read_text(encoding="utf-8")) == {
"version": 1,
"disabled_extensions": ["stale", "templates"],
"sidecar_proxy_consents": {},
}
status = set_extension_user_enabled("templates", True)
@@ -1070,6 +1226,7 @@ def test_set_extension_user_enabled_is_idempotent_and_preserves_stale_ids(tmp_pa
assert json.loads((state_dir / "extension-overrides.json").read_text(encoding="utf-8")) == {
"version": 1,
"disabled_extensions": ["stale"],
"sidecar_proxy_consents": {},
}
assert status["warnings"] == [
{"code": "extension_state_unknown_ids", "source": "extension_state"}
+26 -2
View File
@@ -167,12 +167,21 @@ def test_extensions_panel_renders_loopback_sidecar_monitor_safely():
assert "esc(origin)" in sidecar_block
assert "esc(healthPath)" in sidecar_block
assert "esc(healthUrl)" in sidecar_block
assert "sidecar&&sidecar.proxy" in sidecar_block
assert "proxy.available===true" in sidecar_block
assert "proxy.consented===true" in sidecar_block
assert "proxy.consent_required===true" in sidecar_block
assert "proxy.origin_changed===true" in sidecar_block
assert "Proxy path" in sidecar_block
assert "data-extension-sidecar-proxy-id" in sidecar_block
assert "data-extension-sidecar-proxy-approved" in sidecar_block
assert 'data-sidecar-runtime-index="${index}"' in sidecar_block
assert "fetch(healthUrl,{credentials:'omit',cache:'no-store'" in monitor_block
assert "function _monitorExtensionSidecars(sidecars,seq)" in monitor_block
assert "const seq=_extensionsSidecarMonitorSeq" not in monitor_block
assert "_monitorExtensionSidecars(sidecars,seq)" in render_block
assert "function _renderExtensionsPanel(data,seq)" in render_block
assert "_bindExtensionSidecarProxyButtons(target)" in render_block
assert "const seq=++_extensionsSidecarMonitorSeq" in load_block
assert "opts&&opts.preserveExisting&&target.innerHTML.trim()" in load_block
# A failed refresh must NOT be preserved as "existing content": the Loading/
@@ -204,6 +213,19 @@ def test_extensions_panel_renders_loopback_sidecar_monitor_safely():
assert not _contains_post_method(monitor_block)
def test_extensions_panel_sidecar_proxy_consent_uses_dedicated_endpoint():
bind_block = _between("function _bindExtensionSidecarProxyButtons", "async function handleExtensionToggle")
consent_block = _between("async function handleExtensionSidecarProxyConsent", "function _readExtensionSettingsForm")
assert "handleExtensionSidecarProxyConsent" in bind_block
assert "data-extension-sidecar-proxy-id" in bind_block
assert "api('/api/extensions/sidecar-proxy-consent',{method:'POST',body:JSON.stringify({id,approved})})" in consent_block
assert "Extension sidecar proxy approved." in consent_block
assert "Extension sidecar proxy consent revoked." in consent_block
assert "Failed to update extension sidecar proxy consent" in consent_block
assert "api('/api/extensions/toggle'" not in consent_block
def test_extensions_panel_toggle_uses_dedicated_endpoint_without_settings_or_install():
installed_block = _between("function _extensionInstalledList", "function _extensionSidecarHealthBadge")
toggle_block = _between("async function handleExtensionToggle", "async function loadExtensionsPanel")
@@ -415,11 +437,13 @@ def test_extensions_docs_mentions_settings_panel_without_install_or_proxy_claims
assert "manifests" in diagnostics_section
assert "fetch new extension assets" in diagnostics_section
assert "uninstall files" in diagnostics_section
assert "proxy sidecars" in diagnostics_section
assert "GET /api/extensions/status" in diagnostics_section
assert "`POST /api/extensions/sidecar-proxy-consent`" in diagnostics_section
assert "sanitized loopback sidecars" in diagnostics_section
assert "credentials: 'omit'" in diagnostics_section
assert "does **not** proxy sidecar requests" in diagnostics_section
assert "fixed per-extension sidecar path" in diagnostics_section
assert "WebUI strips `Cookie`, `Authorization`, and CSRF headers" in diagnostics_section
assert "does not create arbitrary extension-owned backend routes" in diagnostics_section
assert "optional top-level `runtime` object" in diagnostics_section
assert "allowlisted scalar fields" in diagnostics_section
assert "browser-local controls" in diagnostics_section