santastabber: authenticated /api/extensions/status endpoint reporting WHY an extension didn't load (stable codes + coarse sources only, no path/env/URL leak). Co-authored-by: santastabber <santastabber@users.noreply.github.com>
This commit is contained in:
1 parent
075b579a9c
commit
1b7c1fe619
5 files changed
+564
-22
No files matched your search
@@ -353,6 +353,8 @@ Full list of environment variables:
|
||||
| `HERMES_WEBUI_AGENT_CACHE_MAX` | `25` | Max live agent instances kept warm in the in-memory LRU. Each pins a full conversation transcript, so this is the dominant lever on resident memory — lower it on installs with many long sessions to cap RAM (at the cost of more cold reloads) |
|
||||
| `HERMES_WEBUI_SESSIONS_MAX` | `100` | Max compact `Session` objects held in the in-memory LRU. Lighter than the agent cache; lower it on installs with hundreds of sessions |
|
||||
|
||||
Extension deployments can inspect sanitized, authenticated diagnostics at `GET /api/extensions/status`; see [WebUI Extensions](docs/EXTENSIONS.md#diagnostics).
|
||||
|
||||
---
|
||||
|
||||
### Remote access (SSH tunnel, Tailscale, phone)
|
||||
|
||||
+177
-22
@@ -10,7 +10,7 @@ import json
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
from urllib.parse import unquote, urlsplit
|
||||
|
||||
from api.helpers import _security_headers, j
|
||||
@@ -78,6 +78,39 @@ def _extension_root() -> Optional[Path]:
|
||||
return root
|
||||
|
||||
|
||||
def _extension_root_status() -> Tuple[Optional[Path], bool, bool]:
|
||||
"""Return (root, configured, valid) without exposing the configured path."""
|
||||
raw = os.getenv(_EXTENSION_DIR_ENV, "").strip()
|
||||
if not raw:
|
||||
return None, False, False
|
||||
root = Path(raw).expanduser().resolve()
|
||||
if not root.exists() or not root.is_dir():
|
||||
return None, True, False
|
||||
return root, True, True
|
||||
|
||||
|
||||
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 _fully_unquote_path(path: str) -> str:
|
||||
"""Decode percent-encoding until stable so encoded dot-segments cannot hide.
|
||||
|
||||
@@ -130,7 +163,12 @@ def _warn_rejected_url(value: str, source: str) -> None:
|
||||
|
||||
|
||||
def _append_safe_asset_url(
|
||||
urls: List[str], value: str, source: str, *, dedupe: bool = True
|
||||
urls: List[str],
|
||||
value: str,
|
||||
source: str,
|
||||
*,
|
||||
dedupe: bool = True,
|
||||
diagnostics: Optional[Dict[str, Any]] = None,
|
||||
) -> bool:
|
||||
"""Append a validated URL while preserving order and the global cap.
|
||||
|
||||
@@ -143,6 +181,7 @@ def _append_safe_asset_url(
|
||||
return True
|
||||
if not _is_safe_asset_url(value):
|
||||
_warn_rejected_url(value, source)
|
||||
_add_diagnostic_warning(diagnostics, "asset_url_rejected", source)
|
||||
return True
|
||||
if dedupe and value in urls:
|
||||
return True
|
||||
@@ -153,12 +192,18 @@ def _append_safe_asset_url(
|
||||
"Extension URL list %s truncated at %d entries",
|
||||
source, _MAX_URL_LIST,
|
||||
)
|
||||
_add_diagnostic_warning(diagnostics, "asset_url_list_truncated", source)
|
||||
return False
|
||||
urls.append(value)
|
||||
return True
|
||||
|
||||
|
||||
def _read_url_list(env_name: str, existing: Optional[List[str]] = None) -> List[str]:
|
||||
def _read_url_list(
|
||||
env_name: str,
|
||||
existing: Optional[List[str]] = None,
|
||||
*,
|
||||
diagnostics: Optional[Dict[str, Any]] = None,
|
||||
) -> List[str]:
|
||||
raw = os.getenv(env_name, "")
|
||||
urls = list(existing or [])
|
||||
# Preserve legacy env-only behavior: duplicate env URLs injected twice before
|
||||
@@ -166,28 +211,35 @@ def _read_url_list(env_name: str, existing: Optional[List[str]] = None) -> List[
|
||||
# so bundle manifests and explicit overrides do not double-load an asset.
|
||||
dedupe = existing is not None
|
||||
for item in raw.split(","):
|
||||
if not _append_safe_asset_url(urls, item, env_name, dedupe=dedupe):
|
||||
if not _append_safe_asset_url(
|
||||
urls, item, env_name, dedupe=dedupe, diagnostics=diagnostics
|
||||
):
|
||||
break
|
||||
return urls
|
||||
|
||||
|
||||
def _manifest_path(root: Path) -> Optional[Path]:
|
||||
def _manifest_path_with_status(root: Path) -> Tuple[Optional[Path], str]:
|
||||
raw = os.getenv(_EXTENSION_MANIFEST_ENV, "").strip()
|
||||
if not raw:
|
||||
return None
|
||||
return None, "not_configured"
|
||||
if raw.startswith(("/", "~")):
|
||||
_log.warning("Rejected extension manifest path from %s", _EXTENSION_MANIFEST_ENV)
|
||||
return None
|
||||
return None, "invalid_path"
|
||||
rel = _fully_unquote_path(raw)
|
||||
if not _is_safe_relative_path(rel):
|
||||
_log.warning("Rejected extension manifest path from %s", _EXTENSION_MANIFEST_ENV)
|
||||
return None
|
||||
return None, "invalid_path"
|
||||
manifest = (root / rel).resolve()
|
||||
try:
|
||||
manifest.relative_to(root)
|
||||
except ValueError:
|
||||
_log.warning("Rejected extension manifest path from %s", _EXTENSION_MANIFEST_ENV)
|
||||
return None
|
||||
return None, "invalid_path"
|
||||
return manifest, "configured"
|
||||
|
||||
|
||||
def _manifest_path(root: Path) -> Optional[Path]:
|
||||
manifest, _ = _manifest_path_with_status(root)
|
||||
return manifest
|
||||
|
||||
|
||||
@@ -219,7 +271,9 @@ def _iter_manifest_entries(manifest: object) -> List[Tuple[str, object]]:
|
||||
extension_entries = manifest
|
||||
if isinstance(extension_entries, list):
|
||||
for index, extension in enumerate(extension_entries):
|
||||
if isinstance(extension, dict) and extension.get("enabled", True) is False:
|
||||
if not isinstance(extension, dict):
|
||||
continue
|
||||
if extension.get("enabled", True) is False:
|
||||
continue
|
||||
entries.append((f"manifest.extensions[{index}]", extension))
|
||||
return entries
|
||||
@@ -238,58 +292,103 @@ def _read_manifest_text(manifest_file: Path) -> str:
|
||||
return data.decode("utf-8")
|
||||
|
||||
|
||||
def _read_manifest_urls(root: Path) -> Tuple[List[str], List[str]]:
|
||||
manifest_file = _manifest_path(root)
|
||||
def _read_manifest_urls_with_diagnostics(
|
||||
root: Path, diagnostics: Optional[Dict[str, Any]] = None
|
||||
) -> Tuple[List[str], List[str], Dict[str, Any]]:
|
||||
manifest_file, path_status = _manifest_path_with_status(root)
|
||||
manifest_status: Dict[str, Any] = {
|
||||
"configured": path_status != "not_configured",
|
||||
"loaded": False,
|
||||
"status": path_status,
|
||||
"entry_count": 0,
|
||||
"script_count": 0,
|
||||
"stylesheet_count": 0,
|
||||
}
|
||||
if manifest_file is None:
|
||||
return [], []
|
||||
if path_status == "invalid_path":
|
||||
_add_diagnostic_warning(diagnostics, "manifest_invalid_path", "manifest")
|
||||
return [], [], manifest_status
|
||||
try:
|
||||
if not manifest_file.exists() or not manifest_file.is_file():
|
||||
_log.warning("Configured extension manifest was not found")
|
||||
return [], []
|
||||
manifest_status["status"] = "missing"
|
||||
_add_diagnostic_warning(diagnostics, "manifest_missing", "manifest")
|
||||
return [], [], manifest_status
|
||||
manifest = json.loads(_read_manifest_text(manifest_file))
|
||||
except _ManifestTooLarge:
|
||||
_log.warning("Configured extension manifest exceeds %d bytes", _MAX_MANIFEST_BYTES)
|
||||
return [], []
|
||||
manifest_status["status"] = "oversized"
|
||||
_add_diagnostic_warning(diagnostics, "manifest_oversized", "manifest")
|
||||
return [], [], manifest_status
|
||||
except json.JSONDecodeError:
|
||||
_log.warning("Configured extension manifest is not valid JSON")
|
||||
return [], []
|
||||
manifest_status["status"] = "malformed"
|
||||
_add_diagnostic_warning(diagnostics, "manifest_malformed", "manifest")
|
||||
return [], [], manifest_status
|
||||
except RecursionError:
|
||||
# A <=64KB but deeply-nested manifest makes json.loads exceed the
|
||||
# interpreter recursion limit. Without this, the RecursionError escapes
|
||||
# into the app-shell route and every page load 503s. Fail safe.
|
||||
_log.warning("Configured extension manifest is too deeply nested")
|
||||
return [], []
|
||||
manifest_status["status"] = "too_deeply_nested"
|
||||
_add_diagnostic_warning(diagnostics, "manifest_too_deeply_nested", "manifest")
|
||||
return [], [], manifest_status
|
||||
except (OSError, UnicodeDecodeError):
|
||||
_log.warning("Configured extension manifest could not be read")
|
||||
return [], []
|
||||
manifest_status["status"] = "unreadable"
|
||||
_add_diagnostic_warning(diagnostics, "manifest_unreadable", "manifest")
|
||||
return [], [], manifest_status
|
||||
|
||||
scripts: List[str] = []
|
||||
stylesheets: List[str] = []
|
||||
entries = _iter_manifest_entries(manifest)
|
||||
manifest_status["entry_count"] = len(entries)
|
||||
scripts_full = False
|
||||
stylesheets_full = False
|
||||
for _source, entry in _iter_manifest_entries(manifest):
|
||||
for _source, entry in entries:
|
||||
if not isinstance(entry, dict):
|
||||
continue
|
||||
script_source = "manifest:scripts"
|
||||
stylesheet_source = "manifest:stylesheets"
|
||||
if not scripts_full:
|
||||
for value in _entry_asset_values(entry, "scripts"):
|
||||
if not _append_safe_asset_url(
|
||||
scripts, _manifest_asset_url(value), "manifest:scripts"
|
||||
scripts,
|
||||
_manifest_asset_url(value),
|
||||
script_source,
|
||||
diagnostics=diagnostics,
|
||||
):
|
||||
scripts_full = True
|
||||
break
|
||||
if not stylesheets_full:
|
||||
for value in _entry_asset_values(entry, "stylesheets"):
|
||||
if not _append_safe_asset_url(
|
||||
stylesheets, _manifest_asset_url(value), "manifest:stylesheets"
|
||||
stylesheets,
|
||||
_manifest_asset_url(value),
|
||||
stylesheet_source,
|
||||
diagnostics=diagnostics,
|
||||
):
|
||||
stylesheets_full = True
|
||||
break
|
||||
if scripts_full and stylesheets_full:
|
||||
break
|
||||
manifest_status.update(
|
||||
{
|
||||
"loaded": True,
|
||||
"status": "loaded",
|
||||
"script_count": len(scripts),
|
||||
"stylesheet_count": len(stylesheets),
|
||||
}
|
||||
)
|
||||
return scripts, stylesheets, manifest_status
|
||||
|
||||
|
||||
def _read_manifest_urls(root: Path) -> Tuple[List[str], List[str]]:
|
||||
scripts, stylesheets, _ = _read_manifest_urls_with_diagnostics(root)
|
||||
return scripts, stylesheets
|
||||
|
||||
|
||||
def get_extension_config() -> Dict[str, object]:
|
||||
def get_extension_config() -> Dict[str, Any]:
|
||||
"""Return public extension config without exposing filesystem paths."""
|
||||
root = _extension_root()
|
||||
if root is None:
|
||||
@@ -306,6 +405,62 @@ def get_extension_config() -> Dict[str, object]:
|
||||
}
|
||||
|
||||
|
||||
def get_extension_status() -> Dict[str, Any]:
|
||||
"""Return sanitized read-only extension diagnostics for administrators."""
|
||||
diagnostics = _new_diagnostics()
|
||||
root, dir_configured, dir_valid = _extension_root_status()
|
||||
manifest_configured = bool(os.getenv(_EXTENSION_MANIFEST_ENV, "").strip())
|
||||
manifest_status: Dict[str, Any] = {
|
||||
"configured": manifest_configured,
|
||||
"loaded": False,
|
||||
"status": "extension_disabled" if manifest_configured else "not_configured",
|
||||
"entry_count": 0,
|
||||
"script_count": 0,
|
||||
"stylesheet_count": 0,
|
||||
}
|
||||
if dir_configured and not dir_valid:
|
||||
_add_diagnostic_warning(diagnostics, "extension_dir_unavailable", "extension_dir")
|
||||
|
||||
if root is None:
|
||||
return {
|
||||
"enabled": False,
|
||||
"extension_dir_configured": dir_configured,
|
||||
"extension_dir_valid": False,
|
||||
"script_urls": [],
|
||||
"stylesheet_urls": [],
|
||||
"counts": {"script_urls": 0, "stylesheet_urls": 0},
|
||||
"manifest": manifest_status,
|
||||
"warnings": diagnostics["warnings"],
|
||||
}
|
||||
|
||||
manifest_scripts, manifest_stylesheets, manifest_status = _read_manifest_urls_with_diagnostics(
|
||||
root, diagnostics
|
||||
)
|
||||
script_urls = _read_url_list(
|
||||
_EXTENSION_SCRIPT_URLS_ENV,
|
||||
manifest_scripts or None,
|
||||
diagnostics=diagnostics,
|
||||
)
|
||||
stylesheet_urls = _read_url_list(
|
||||
_EXTENSION_STYLESHEET_URLS_ENV,
|
||||
manifest_stylesheets or None,
|
||||
diagnostics=diagnostics,
|
||||
)
|
||||
return {
|
||||
"enabled": True,
|
||||
"extension_dir_configured": True,
|
||||
"extension_dir_valid": True,
|
||||
"script_urls": script_urls,
|
||||
"stylesheet_urls": stylesheet_urls,
|
||||
"counts": {
|
||||
"script_urls": len(script_urls),
|
||||
"stylesheet_urls": len(stylesheet_urls),
|
||||
},
|
||||
"manifest": manifest_status,
|
||||
"warnings": diagnostics["warnings"],
|
||||
}
|
||||
|
||||
|
||||
def inject_extension_tags(index_html: str) -> str:
|
||||
"""Inject configured extension tags into the app shell.
|
||||
|
||||
|
||||
@@ -8889,6 +8889,11 @@ def handle_get(handler, parsed) -> bool:
|
||||
if parsed.path == "/api/onboarding/status":
|
||||
return j(handler, get_onboarding_status())
|
||||
|
||||
if parsed.path == "/api/extensions/status":
|
||||
from api.extensions import get_extension_status
|
||||
|
||||
return j(handler, get_extension_status())
|
||||
|
||||
if parsed.path.startswith("/extensions/"):
|
||||
from api.extensions import serve_extension_static
|
||||
|
||||
|
||||
@@ -252,3 +252,23 @@ HERMES_WEBUI_EXTENSION_SCRIPT_URLS=/extensions/app.js \
|
||||
```
|
||||
|
||||
Open the WebUI and confirm the badge appears.
|
||||
|
||||
## Diagnostics
|
||||
|
||||
Authenticated administrators can inspect sanitized extension configuration at:
|
||||
|
||||
```text
|
||||
GET /api/extensions/status
|
||||
```
|
||||
|
||||
The endpoint is read-only and follows the normal WebUI authentication rules. It
|
||||
returns the same public asset URLs that can already be injected into the HTML,
|
||||
plus coarse manifest status, asset counts, and warning codes for rejected or
|
||||
unavailable configuration. `manifest.script_count` and
|
||||
`manifest.stylesheet_count` count accepted assets from the manifest only;
|
||||
`counts.script_urls` and `counts.stylesheet_urls` count the final post-env-merge
|
||||
URLs. `manifest.entry_count` counts the loaded top-level manifest object and
|
||||
enabled extension entries that were inspected, not every extension object in the
|
||||
file. The endpoint does **not** return
|
||||
`HERMES_WEBUI_EXTENSION_DIR`, resolved manifest paths, raw environment values, or
|
||||
rejected URL strings.
|
||||
@@ -0,0 +1,360 @@
|
||||
"""Tests for sanitized WebUI extension diagnostics."""
|
||||
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
class FakeHandler:
|
||||
def __init__(self):
|
||||
self.status = None
|
||||
self.headers = {}
|
||||
self.sent_headers = []
|
||||
self.body = bytearray()
|
||||
self.wfile = self
|
||||
|
||||
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_EXTENSION_SCRIPT_URLS",
|
||||
"HERMES_WEBUI_EXTENSION_STYLESHEET_URLS",
|
||||
"HERMES_WEBUI_PASSWORD",
|
||||
):
|
||||
monkeypatch.delenv(name, raising=False)
|
||||
auth_mod._invalidate_password_hash_cache()
|
||||
yield
|
||||
auth_mod._invalidate_password_hash_cache()
|
||||
|
||||
|
||||
def test_extension_status_disabled_by_default():
|
||||
from api.extensions import get_extension_status
|
||||
|
||||
assert get_extension_status() == {
|
||||
"enabled": False,
|
||||
"extension_dir_configured": False,
|
||||
"extension_dir_valid": False,
|
||||
"script_urls": [],
|
||||
"stylesheet_urls": [],
|
||||
"counts": {"script_urls": 0, "stylesheet_urls": 0},
|
||||
"manifest": {
|
||||
"configured": False,
|
||||
"loaded": False,
|
||||
"status": "not_configured",
|
||||
"entry_count": 0,
|
||||
"script_count": 0,
|
||||
"stylesheet_count": 0,
|
||||
},
|
||||
"warnings": [],
|
||||
}
|
||||
|
||||
|
||||
def test_extension_status_reports_invalid_extension_dir_without_path(tmp_path, monkeypatch):
|
||||
missing = tmp_path / "missing-extension-dir"
|
||||
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_DIR", str(missing))
|
||||
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_MANIFEST", "extensions.json")
|
||||
|
||||
from api.extensions import get_extension_status
|
||||
|
||||
status = get_extension_status()
|
||||
assert status["enabled"] is False
|
||||
assert status["extension_dir_configured"] is True
|
||||
assert status["extension_dir_valid"] is False
|
||||
assert status["manifest"]["status"] == "extension_disabled"
|
||||
assert status["warnings"] == [
|
||||
{"code": "extension_dir_unavailable", "source": "extension_dir"}
|
||||
]
|
||||
assert str(missing) not in repr(status)
|
||||
|
||||
|
||||
def test_extension_status_reports_loaded_manifest_counts_and_urls(tmp_path, monkeypatch):
|
||||
root = tmp_path / "extensions"
|
||||
root.mkdir()
|
||||
(root / "extensions.json").write_text(
|
||||
"""
|
||||
{
|
||||
"scripts": ["runtime.js"],
|
||||
"stylesheets": ["base.css"],
|
||||
"extensions": [
|
||||
{"id": "templates", "scripts": ["templates/app.js"], "stylesheets": ["templates/app.css"]},
|
||||
{"id": "off", "enabled": false, "scripts": ["off.js"]}
|
||||
]
|
||||
}
|
||||
""",
|
||||
encoding="utf-8",
|
||||
)
|
||||
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_DIR", str(root))
|
||||
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_MANIFEST", "extensions.json")
|
||||
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_SCRIPT_URLS", "/extensions/env.js")
|
||||
|
||||
from api.extensions import get_extension_status
|
||||
|
||||
status = get_extension_status()
|
||||
assert status["enabled"] is True
|
||||
assert status["extension_dir_configured"] is True
|
||||
assert status["extension_dir_valid"] is True
|
||||
assert status["script_urls"] == [
|
||||
"/extensions/runtime.js",
|
||||
"/extensions/templates/app.js",
|
||||
"/extensions/env.js",
|
||||
]
|
||||
assert status["stylesheet_urls"] == [
|
||||
"/extensions/base.css",
|
||||
"/extensions/templates/app.css",
|
||||
]
|
||||
assert status["counts"] == {"script_urls": 3, "stylesheet_urls": 2}
|
||||
assert status["manifest"] == {
|
||||
"configured": True,
|
||||
"loaded": True,
|
||||
"status": "loaded",
|
||||
"entry_count": 2,
|
||||
"script_count": 2,
|
||||
"stylesheet_count": 2,
|
||||
}
|
||||
assert status["warnings"] == []
|
||||
|
||||
|
||||
def test_extension_status_ignores_non_dict_manifest_extensions_in_entry_count(
|
||||
tmp_path, monkeypatch
|
||||
):
|
||||
root = tmp_path / "extensions"
|
||||
root.mkdir()
|
||||
(root / "extensions.json").write_text(
|
||||
"""
|
||||
{
|
||||
"extensions": [
|
||||
null,
|
||||
"not-an-extension",
|
||||
{"id": "templates", "scripts": ["templates/app.js"]},
|
||||
{"id": "off", "enabled": false, "scripts": ["off.js"]}
|
||||
]
|
||||
}
|
||||
""",
|
||||
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
|
||||
|
||||
status = get_extension_status()
|
||||
assert status["script_urls"] == ["/extensions/templates/app.js"]
|
||||
assert status["manifest"]["entry_count"] == 2
|
||||
assert status["manifest"]["script_count"] == 1
|
||||
assert status["warnings"] == []
|
||||
|
||||
|
||||
def test_extension_status_reports_missing_manifest_safely(tmp_path, monkeypatch):
|
||||
root = tmp_path / "extensions"
|
||||
root.mkdir()
|
||||
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_DIR", str(root))
|
||||
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_MANIFEST", "missing.json")
|
||||
|
||||
from api.extensions import get_extension_status
|
||||
|
||||
status = get_extension_status()
|
||||
assert status["manifest"]["status"] == "missing"
|
||||
assert status["manifest"]["loaded"] is False
|
||||
assert status["warnings"] == [{"code": "manifest_missing", "source": "manifest"}]
|
||||
assert str(root) not in repr(status)
|
||||
assert "missing.json" not in repr(status)
|
||||
|
||||
|
||||
def test_extension_status_reports_malformed_manifest_safely(tmp_path, monkeypatch):
|
||||
root = tmp_path / "extensions"
|
||||
root.mkdir()
|
||||
(root / "bad.json").write_text("{not json", encoding="utf-8")
|
||||
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_DIR", str(root))
|
||||
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_MANIFEST", "bad.json")
|
||||
|
||||
from api.extensions import get_extension_status
|
||||
|
||||
status = get_extension_status()
|
||||
assert status["manifest"]["status"] == "malformed"
|
||||
assert status["warnings"] == [{"code": "manifest_malformed", "source": "manifest"}]
|
||||
assert "bad.json" not in repr(status)
|
||||
|
||||
|
||||
def test_extension_status_reports_unreadable_manifest_safely(tmp_path, monkeypatch):
|
||||
root = tmp_path / "extensions"
|
||||
root.mkdir()
|
||||
(root / "bad-utf8.json").write_bytes(b"\xff\xfe")
|
||||
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_DIR", str(root))
|
||||
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_MANIFEST", "bad-utf8.json")
|
||||
|
||||
from api.extensions import get_extension_status
|
||||
|
||||
status = get_extension_status()
|
||||
assert status["manifest"]["status"] == "unreadable"
|
||||
assert status["warnings"] == [{"code": "manifest_unreadable", "source": "manifest"}]
|
||||
assert "bad-utf8.json" not in repr(status)
|
||||
|
||||
|
||||
def test_extension_status_reports_manifest_disabled_when_dir_unconfigured(monkeypatch):
|
||||
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_MANIFEST", "extensions.json")
|
||||
|
||||
from api.extensions import get_extension_status
|
||||
|
||||
status = get_extension_status()
|
||||
assert status["enabled"] is False
|
||||
assert status["extension_dir_configured"] is False
|
||||
assert status["extension_dir_valid"] is False
|
||||
assert status["manifest"]["status"] == "extension_disabled"
|
||||
assert status["manifest"]["configured"] is True
|
||||
assert status["warnings"] == []
|
||||
assert "extensions.json" not in repr(status)
|
||||
|
||||
|
||||
def test_extension_status_reports_oversized_manifest_safely(tmp_path, monkeypatch):
|
||||
root = tmp_path / "extensions"
|
||||
root.mkdir()
|
||||
(root / "huge.json").write_text(" " * (64 * 1024 + 1), encoding="utf-8")
|
||||
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_DIR", str(root))
|
||||
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_MANIFEST", "huge.json")
|
||||
|
||||
from api.extensions import get_extension_status
|
||||
|
||||
status = get_extension_status()
|
||||
assert status["manifest"]["status"] == "oversized"
|
||||
assert status["warnings"] == [{"code": "manifest_oversized", "source": "manifest"}]
|
||||
assert "huge.json" not in repr(status)
|
||||
|
||||
|
||||
def test_extension_status_reports_invalid_manifest_path_safely(tmp_path, monkeypatch):
|
||||
root = tmp_path / "extensions"
|
||||
root.mkdir()
|
||||
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_DIR", str(root))
|
||||
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_MANIFEST", "../outside.json")
|
||||
|
||||
from api.extensions import get_extension_status
|
||||
|
||||
status = get_extension_status()
|
||||
assert status["manifest"]["status"] == "invalid_path"
|
||||
assert status["warnings"] == [
|
||||
{"code": "manifest_invalid_path", "source": "manifest"}
|
||||
]
|
||||
assert "outside.json" not in repr(status)
|
||||
assert str(root) not in repr(status)
|
||||
|
||||
|
||||
def test_extension_status_reports_recursion_error_safely(tmp_path, monkeypatch):
|
||||
root = tmp_path / "extensions"
|
||||
root.mkdir()
|
||||
(root / "deep.json").write_text("[]", encoding="utf-8")
|
||||
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_DIR", str(root))
|
||||
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_MANIFEST", "deep.json")
|
||||
|
||||
import api.extensions as extensions
|
||||
|
||||
def raise_recursion_error(_manifest_file):
|
||||
raise RecursionError("manifest nesting exceeded")
|
||||
|
||||
monkeypatch.setattr(extensions, "_read_manifest_text", raise_recursion_error)
|
||||
|
||||
status = extensions.get_extension_status()
|
||||
assert status["manifest"]["status"] == "too_deeply_nested"
|
||||
assert status["warnings"] == [
|
||||
{"code": "manifest_too_deeply_nested", "source": "manifest"}
|
||||
]
|
||||
assert "deep.json" not in repr(status)
|
||||
|
||||
|
||||
def test_extension_status_reports_rejected_assets_without_rejected_values(tmp_path, monkeypatch):
|
||||
root = tmp_path / "extensions"
|
||||
root.mkdir()
|
||||
(root / "extensions.json").write_text(
|
||||
"""
|
||||
{
|
||||
"scripts": ["safe.js", "https://evil.example/app.js", "../escape.js"],
|
||||
"stylesheets": ["safe.css", "nested/../escape.css"]
|
||||
}
|
||||
""",
|
||||
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
|
||||
|
||||
status = get_extension_status()
|
||||
assert status["script_urls"] == ["/extensions/safe.js"]
|
||||
assert status["stylesheet_urls"] == ["/extensions/safe.css"]
|
||||
assert {tuple(sorted(item.items())) for item in status["warnings"]} == {
|
||||
tuple(sorted({"code": "asset_url_rejected", "source": "manifest:scripts"}.items())),
|
||||
tuple(sorted({"code": "asset_url_rejected", "source": "manifest:stylesheets"}.items())),
|
||||
}
|
||||
rendered = repr(status)
|
||||
assert "evil.example" not in rendered
|
||||
assert "escape.js" not in rendered
|
||||
assert "escape.css" not in rendered
|
||||
|
||||
|
||||
def test_extension_status_reports_rejected_env_assets_without_rejected_values(tmp_path, monkeypatch):
|
||||
root = tmp_path / "extensions"
|
||||
root.mkdir()
|
||||
monkeypatch.setenv("HERMES_WEBUI_EXTENSION_DIR", str(root))
|
||||
monkeypatch.setenv(
|
||||
"HERMES_WEBUI_EXTENSION_SCRIPT_URLS",
|
||||
"/extensions/safe.js, https://evil.example/env.js",
|
||||
)
|
||||
|
||||
from api.extensions import get_extension_status
|
||||
|
||||
status = get_extension_status()
|
||||
assert status["script_urls"] == ["/extensions/safe.js"]
|
||||
assert status["warnings"] == [
|
||||
{"code": "asset_url_rejected", "source": "HERMES_WEBUI_EXTENSION_SCRIPT_URLS"}
|
||||
]
|
||||
rendered = repr(status)
|
||||
assert "evil.example" not in rendered
|
||||
assert "env.js" not in rendered
|
||||
|
||||
|
||||
def test_extension_status_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, "j", fake_j)
|
||||
handler = FakeHandler()
|
||||
assert routes.handle_get(handler, SimpleNamespace(path="/api/extensions/status")) is True
|
||||
assert captured["status"] == 200
|
||||
assert captured["data"]["enabled"] is False
|
||||
|
||||
|
||||
def test_extension_status_route_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/status", query="")) is False
|
||||
assert handler.status == 401
|
||||
assert handler.header("Location") is None
|
||||
Reference in new issue
Block a user