From ee7ff6ed33c2033b579b5373636d6752f7cfaa89 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Wed, 24 Jun 2026 23:27:34 -0400 Subject: [PATCH] fix(extensions): harden gallery against live registry and security review findings --- api/extensions.py | 141 +++++++++++++---- static/panels.js | 2 +- tests/test_issue4746_extension_gallery.py | 178 ++++++++++++++++++++-- 3 files changed, 271 insertions(+), 50 deletions(-) diff --git a/api/extensions.py b/api/extensions.py index 5da2cd2a..575b4546 100644 --- a/api/extensions.py +++ b/api/extensions.py @@ -18,7 +18,7 @@ import hashlib import io import time import zipfile -from urllib.request import urlopen +from urllib.request import HTTPRedirectHandler, Request, build_opener, urlopen from api.helpers import _security_headers, j @@ -80,11 +80,32 @@ _MAX_INSTALL_MANIFEST_BYTES = 128 * 1024 _MAX_GALLERY_INSTALLED_IDS = 256 _MAX_ZIP_DOWNLOAD_BYTES = 32 * 1024 * 1024 _REGISTRY_URL = "https://hermes-webui.github.io/hermes-webui-extensions/registry.json" -_REGISTRY_ALLOWED_DOWNLOAD_HOST = "hermes-webui.github.io" +_REGISTRY_ALLOWED_DOWNLOAD_HOSTS = frozenset({"hermes-webui.github.io"}) _REGISTRY_CACHE: dict = {} _REGISTRY_LOCK = threading.Lock() _REGISTRY_TTL_SECONDS = 300 + +class _AllowlistRedirectHandler(HTTPRedirectHandler): + """Reject redirects to hosts not in the download allowlist.""" + + def redirect_request(self, req, fp, code, msg, headers, newurl): + parsed = urlsplit(newurl) + if parsed.scheme != "https" or parsed.hostname not in _REGISTRY_ALLOWED_DOWNLOAD_HOSTS: + raise ExtensionInstallError("Download redirected to disallowed host") + return super().redirect_request(req, fp, code, msg, headers, newurl) + + +def _safe_download(url: str, max_bytes: int, timeout: int = 30) -> bytes: + """Download from an allowlisted host, rejecting cross-host redirects.""" + opener = build_opener(_AllowlistRedirectHandler) + resp = opener.open(url, timeout=timeout) + try: + return resp.read(max_bytes + 1) + finally: + resp.close() + + _EXTENSION_MIME = { "css": "text/css", "js": "application/javascript", @@ -979,7 +1000,7 @@ def install_extension(id: object, download_url: object, sha256: object) -> Dict[ if not isinstance(download_url, str) or not download_url.startswith("https://"): raise ExtensionInstallError("Invalid download URL") parsed_url = urlsplit(download_url) - if parsed_url.hostname != _REGISTRY_ALLOWED_DOWNLOAD_HOST: + if parsed_url.hostname not in _REGISTRY_ALLOWED_DOWNLOAD_HOSTS: raise ExtensionInstallError("Invalid download URL") if not isinstance(sha256, str) or not re.fullmatch(r"[0-9a-f]{64}", sha256): raise ExtensionInstallError("Invalid sha256") @@ -987,7 +1008,9 @@ def install_extension(id: object, download_url: object, sha256: object) -> Dict[ if root is None: raise ExtensionInstallError("Extensions not configured", 404) try: - raw_data = urlopen(download_url, timeout=30).read(_MAX_ZIP_DOWNLOAD_BYTES + 1) + raw_data = _safe_download(download_url, _MAX_ZIP_DOWNLOAD_BYTES) + except ExtensionInstallError: + raise except Exception as exc: raise ExtensionInstallError("Download failed", 502) from exc if len(raw_data) > _MAX_ZIP_DOWNLOAD_BYTES: @@ -1003,36 +1026,54 @@ def install_extension(id: object, download_url: object, sha256: object) -> Dict[ total_uncompressed = sum(info.file_size for info in zf.infolist() if not info.is_dir()) if total_uncompressed > _MAX_ZIP_DOWNLOAD_BYTES * 10: raise ExtensionInstallError("Archive uncompressed size exceeds limit") - if len([n for n in member_names if n and not n.endswith("/")]) > 1024: + file_members = [n for n in member_names if n and not n.endswith("/")] + if len(file_members) > 1024: raise ExtensionInstallError("Archive contains too many files") - for member_name in member_names: - if not member_name or member_name.endswith("/"): - continue - decoded = _fully_unquote_path(member_name) - if not _is_safe_relative_path(decoded): + # Detect and strip a single top-level directory prefix matching the extension id. + # Registry artifacts root files under / (e.g. desktop-companion/manifest.json). + strip_prefix = "" + candidate = ext_id + "/" + if all(n.startswith(candidate) for n in file_members): + strip_prefix = candidate + def _stripped(name: str) -> str: + if strip_prefix and name.startswith(strip_prefix): + return name[len(strip_prefix):] + return name + root_resolved = root.resolve() + ext_dir_resolved = ext_dir.resolve() + for member_name in file_members: + decoded = _fully_unquote_path(_stripped(member_name)) + if not decoded or not _is_safe_relative_path(decoded): raise ExtensionInstallError("Unsafe archive member") resolved = (ext_dir / decoded).resolve() try: - resolved.relative_to(ext_dir.resolve()) + resolved.relative_to(root_resolved) except ValueError as exc: raise ExtensionInstallError("Zip-slip detected") from exc - # Determine version from manifest.json in zip if present - version = "unknown" - if "manifest.json" in member_names: try: - mdata = json.loads(zf.read("manifest.json").decode("utf-8")) - if isinstance(mdata, dict) and isinstance(mdata.get("version"), str): - version = mdata["version"] - except Exception: - pass + resolved.relative_to(ext_dir_resolved) + except ValueError as exc: + raise ExtensionInstallError("Zip-slip detected") from exc + # Determine version from extension.json or manifest.json in zip + version = "unknown" + for vfile in ("extension.json", "manifest.json"): + candidate_name = strip_prefix + vfile if strip_prefix else vfile + if candidate_name in member_names: + try: + mdata = json.loads(zf.read(candidate_name).decode("utf-8")) + if isinstance(mdata, dict) and isinstance(mdata.get("version"), str): + version = mdata["version"] + break + except Exception: + pass with _EXTENSION_STATE_LOCK: ext_dir.mkdir(parents=True, exist_ok=True) + if ext_dir.is_symlink(): + raise ExtensionInstallError("Extension directory is a symlink", 400) rollback: List[Path] = [] try: - for member_name in member_names: - if not member_name or member_name.endswith("/"): - continue - decoded = _fully_unquote_path(member_name) + for member_name in file_members: + decoded = _fully_unquote_path(_stripped(member_name)) dest = (ext_dir / decoded).resolve() dest.parent.mkdir(parents=True, exist_ok=True) dest.write_bytes(zf.read(member_name)) @@ -1052,12 +1093,28 @@ def install_extension(id: object, download_url: object, sha256: object) -> Dict[ try: manifest = _load_install_manifest() from datetime import datetime, timezone + rel_files = [p.relative_to(ext_dir_resolved).as_posix() for p in rollback] manifest["installed"][ext_id] = { "version": version, - "files": [str(p.relative_to(ext_dir)) for p in rollback], + "files": rel_files, "installed_at": datetime.now(timezone.utc).isoformat(), } + encoded = json.dumps(manifest, ensure_ascii=False, indent=2).encode("utf-8") + if len(encoded) > _MAX_INSTALL_MANIFEST_BYTES: + raise ExtensionInstallError("Install manifest would exceed size limit") _write_install_manifest(manifest) + except ExtensionInstallError: + for path in rollback: + try: + path.unlink(missing_ok=True) + except OSError: + pass + try: + if ext_dir.exists() and not any(ext_dir.iterdir()): + ext_dir.rmdir() + except OSError: + pass + raise except Exception as exc: for path in rollback: try: @@ -1099,11 +1156,23 @@ def uninstall_extension(id: object) -> Dict[str, Any]: target.unlink(missing_ok=True) except OSError: pass - try: - if ext_dir.exists() and not any(ext_dir.iterdir()): - ext_dir.rmdir() - except OSError: - pass + # Remove empty directories bottom-up + if ext_dir.exists(): + for dirpath in sorted( + (d for d in ext_dir.rglob("*") if d.is_dir()), + key=lambda p: len(p.parts), + reverse=True, + ): + try: + if not any(dirpath.iterdir()): + dirpath.rmdir() + except OSError: + pass + try: + if not any(ext_dir.iterdir()): + ext_dir.rmdir() + except OSError: + pass del manifest["installed"][ext_id] _write_install_manifest(manifest) return {"uninstalled": True, "id": ext_id} @@ -1120,11 +1189,17 @@ def get_extension_registry() -> Dict[str, Any]: try: raw = urlopen(_REGISTRY_URL, timeout=10).read(2 * 1024 * 1024) data = json.loads(raw.decode("utf-8")) - if not isinstance(data, list): - data = data.get("entries", []) if isinstance(data, dict) else [] - _REGISTRY_CACHE["data"] = data + if isinstance(data, list): + entries = data + elif isinstance(data, dict): + entries = data.get("extensions") or data.get("entries") or [] + else: + entries = [] + if not isinstance(entries, list): + entries = [] + _REGISTRY_CACHE["data"] = entries _REGISTRY_CACHE["fetched_at"] = now - return {"entries": data} + return {"entries": entries} except Exception: return {"entries": [], "error": "registry_unavailable"} diff --git a/static/panels.js b/static/panels.js index e168cfd9..1b016971 100644 --- a/static/panels.js +++ b/static/panels.js @@ -8357,7 +8357,7 @@ async function handleExtensionInstall(btn,entry){ try{ const result=await api('/api/extensions/install',{method:'POST',body:JSON.stringify({ id:entry.id, - download_url:entry.download_url, + download_url:entry.download_url||entry.download, sha256:entry.sha256, })}); const restart=!!(entry.lifecycle&&entry.lifecycle.restart_required); diff --git a/tests/test_issue4746_extension_gallery.py b/tests/test_issue4746_extension_gallery.py index bc4c99be..5a5c0b50 100644 --- a/tests/test_issue4746_extension_gallery.py +++ b/tests/test_issue4746_extension_gallery.py @@ -45,7 +45,8 @@ def test_install_valid(monkeypatch, tmp_path): mock_resp = MagicMock() mock_resp.read.return_value = zip_bytes - monkeypatch.setattr(ext_mod, "urlopen", lambda *a, **kw: mock_resp) + mock_resp.close = MagicMock() + monkeypatch.setattr(ext_mod, "_safe_download", lambda *a, **kw: zip_bytes) result = ext_mod.install_extension( "my-ext", @@ -61,6 +62,37 @@ def test_install_valid(monkeypatch, tmp_path): assert "index.js" in manifest["installed"]["my-ext"]["files"] +def test_install_prefixed_zip(monkeypatch, tmp_path): + """Zip members rooted under / are stripped so files land at ext_dir//.""" + ext_dir, state_dir = _setup_ext_env(monkeypatch, tmp_path) + import api.extensions as ext_mod + + files = { + "my-ext/manifest.json": json.dumps({"version": "2.0.0"}), + "my-ext/index.js": "console.log('prefixed');", + "my-ext/sub/style.css": "body{}", + } + zip_bytes = _make_zip(files) + sha = hashlib.sha256(zip_bytes).hexdigest() + + monkeypatch.setattr(ext_mod, "_safe_download", lambda *a, **kw: zip_bytes) + + result = ext_mod.install_extension( + "my-ext", + "https://hermes-webui.github.io/exts/my-ext.zip", + sha, + ) + assert result["installed"] is True + assert result["version"] == "2.0.0" + assert (ext_dir / "my-ext" / "manifest.json").exists() + assert (ext_dir / "my-ext" / "index.js").exists() + assert (ext_dir / "my-ext" / "sub" / "style.css").exists() + # Verify NO double-nested directory + assert not (ext_dir / "my-ext" / "my-ext").exists() + manifest = ext_mod._load_install_manifest() + assert "manifest.json" in manifest["installed"]["my-ext"]["files"] + + def test_install_bad_hash(monkeypatch, tmp_path): ext_dir, state_dir = _setup_ext_env(monkeypatch, tmp_path) import api.extensions as ext_mod @@ -68,9 +100,7 @@ def test_install_bad_hash(monkeypatch, tmp_path): zip_bytes = _make_zip({"index.js": "code"}) wrong_sha = "a" * 64 - mock_resp = MagicMock() - mock_resp.read.return_value = zip_bytes - monkeypatch.setattr(ext_mod, "urlopen", lambda *a, **kw: mock_resp) + monkeypatch.setattr(ext_mod, "_safe_download", lambda *a, **kw: zip_bytes) with pytest.raises(ext_mod.ExtensionInstallError, match="SHA-256"): ext_mod.install_extension( @@ -85,16 +115,13 @@ def test_install_zipslip(monkeypatch, tmp_path): ext_dir, state_dir = _setup_ext_env(monkeypatch, tmp_path) import api.extensions as ext_mod - # Build a zip with a path-traversal member buf = io.BytesIO() with zipfile.ZipFile(buf, "w") as zf: zf.writestr("../../evil.txt", "pwned") zip_bytes = buf.getvalue() sha = hashlib.sha256(zip_bytes).hexdigest() - mock_resp = MagicMock() - mock_resp.read.return_value = zip_bytes - monkeypatch.setattr(ext_mod, "urlopen", lambda *a, **kw: mock_resp) + monkeypatch.setattr(ext_mod, "_safe_download", lambda *a, **kw: zip_bytes) with pytest.raises(ext_mod.ExtensionInstallError): ext_mod.install_extension( @@ -113,9 +140,7 @@ def test_uninstall(monkeypatch, tmp_path): zip_bytes = _make_zip(files) sha = hashlib.sha256(zip_bytes).hexdigest() - mock_resp = MagicMock() - mock_resp.read.return_value = zip_bytes - monkeypatch.setattr(ext_mod, "urlopen", lambda *a, **kw: mock_resp) + monkeypatch.setattr(ext_mod, "_safe_download", lambda *a, **kw: zip_bytes) ext_mod.install_extension( "rm-ext", @@ -128,10 +153,37 @@ def test_uninstall(monkeypatch, tmp_path): assert result["uninstalled"] is True assert not (ext_dir / "rm-ext" / "index.js").exists() assert not (ext_dir / "rm-ext" / "style.css").exists() + assert not (ext_dir / "rm-ext").exists() manifest = ext_mod._load_install_manifest() assert "rm-ext" not in manifest["installed"] +def test_uninstall_cleans_nested_dirs(monkeypatch, tmp_path): + """Uninstall removes empty subdirectories left by nested files.""" + ext_dir, state_dir = _setup_ext_env(monkeypatch, tmp_path) + import api.extensions as ext_mod + + files = { + "deep-ext/manifest.json": "{}", + "deep-ext/sub/a/b.js": "code", + "deep-ext/sub/c.css": "body{}", + } + zip_bytes = _make_zip(files) + sha = hashlib.sha256(zip_bytes).hexdigest() + + monkeypatch.setattr(ext_mod, "_safe_download", lambda *a, **kw: zip_bytes) + + ext_mod.install_extension( + "deep-ext", + "https://hermes-webui.github.io/exts/deep-ext.zip", + sha, + ) + assert (ext_dir / "deep-ext" / "sub" / "a" / "b.js").exists() + + ext_mod.uninstall_extension("deep-ext") + assert not (ext_dir / "deep-ext").exists() + + def test_install_rollback(monkeypatch, tmp_path): ext_dir, state_dir = _setup_ext_env(monkeypatch, tmp_path) import api.extensions as ext_mod @@ -140,9 +192,7 @@ def test_install_rollback(monkeypatch, tmp_path): zip_bytes = _make_zip(files) sha = hashlib.sha256(zip_bytes).hexdigest() - mock_resp = MagicMock() - mock_resp.read.return_value = zip_bytes - monkeypatch.setattr(ext_mod, "urlopen", lambda *a, **kw: mock_resp) + monkeypatch.setattr(ext_mod, "_safe_download", lambda *a, **kw: zip_bytes) write_count = [0] original_write_bytes = Path.write_bytes @@ -165,7 +215,8 @@ def test_install_rollback(monkeypatch, tmp_path): assert remaining == [] -def test_gallery_render(monkeypatch, tmp_path): +def test_gallery_registry_list_format(monkeypatch, tmp_path): + """Registry response as a top-level list (original format).""" import api.extensions as ext_mod registry_data = [ @@ -176,7 +227,6 @@ def test_gallery_render(monkeypatch, tmp_path): mock_resp = MagicMock() mock_resp.read.return_value = json.dumps(registry_data).encode("utf-8") - # Clear cache so we always do a fresh fetch ext_mod._REGISTRY_CACHE.clear() monkeypatch.setattr(ext_mod, "urlopen", lambda *a, **kw: mock_resp) @@ -185,3 +235,99 @@ def test_gallery_render(monkeypatch, tmp_path): assert len(result["entries"]) == 2 assert result["entries"][0]["id"] == "ext-one" assert result["entries"][1]["id"] == "ext-two" + + +def test_gallery_registry_extensions_format(monkeypatch, tmp_path): + """Registry response wrapping extensions under {"extensions": [...]} (live format).""" + import api.extensions as ext_mod + + registry_data = { + "version": 1, + "generated_at": "2026-06-24T18:30:40.265Z", + "extensions": [ + {"id": "desktop-companion", "name": "Desktop Companion", "version": "0.1.0"}, + ], + } + + mock_resp = MagicMock() + mock_resp.read.return_value = json.dumps(registry_data).encode("utf-8") + + ext_mod._REGISTRY_CACHE.clear() + monkeypatch.setattr(ext_mod, "urlopen", lambda *a, **kw: mock_resp) + + result = ext_mod.get_extension_registry() + assert len(result["entries"]) == 1 + assert result["entries"][0]["id"] == "desktop-companion" + + +def test_install_rejects_symlinked_ext_dir_outside_root(monkeypatch, tmp_path): + """Installation rejects a symlinked extension directory pointing outside root.""" + ext_dir, state_dir = _setup_ext_env(monkeypatch, tmp_path) + import api.extensions as ext_mod + + target_dir = tmp_path / "symlink_target" + target_dir.mkdir() + link = ext_dir / "linked-ext" + try: + link.symlink_to(target_dir) + except (OSError, NotImplementedError): + pytest.skip("symlinks not supported") + + files = {"index.js": "code"} + zip_bytes = _make_zip(files) + sha = hashlib.sha256(zip_bytes).hexdigest() + + monkeypatch.setattr(ext_mod, "_safe_download", lambda *a, **kw: zip_bytes) + + with pytest.raises(ext_mod.ExtensionInstallError): + ext_mod.install_extension( + "linked-ext", + "https://hermes-webui.github.io/exts/linked-ext.zip", + sha, + ) + assert not (target_dir / "index.js").exists() + + +def test_install_rejects_symlinked_ext_dir_inside_root(monkeypatch, tmp_path): + """Installation rejects a pre-existing symlink even when target is within root.""" + ext_dir, state_dir = _setup_ext_env(monkeypatch, tmp_path) + import api.extensions as ext_mod + + target_dir = ext_dir / "real-target" + target_dir.mkdir() + link = ext_dir / "linked-ext" + try: + link.symlink_to(target_dir) + except (OSError, NotImplementedError): + pytest.skip("symlinks not supported") + + files = {"index.js": "code"} + zip_bytes = _make_zip(files) + sha = hashlib.sha256(zip_bytes).hexdigest() + + monkeypatch.setattr(ext_mod, "_safe_download", lambda *a, **kw: zip_bytes) + + with pytest.raises(ext_mod.ExtensionInstallError, match="symlink"): + ext_mod.install_extension( + "linked-ext", + "https://hermes-webui.github.io/exts/linked-ext.zip", + sha, + ) + + +def test_install_rejects_redirect_to_disallowed_host(monkeypatch, tmp_path): + """Download rejects redirects to disallowed hosts.""" + import api.extensions as ext_mod + + def fake_download(url, max_bytes, timeout=30): + raise ext_mod.ExtensionInstallError("Download redirected to disallowed host") + + ext_dir, state_dir = _setup_ext_env(monkeypatch, tmp_path) + monkeypatch.setattr(ext_mod, "_safe_download", fake_download) + + with pytest.raises(ext_mod.ExtensionInstallError, match="disallowed host"): + ext_mod.install_extension( + "redir-ext", + "https://hermes-webui.github.io/exts/redir-ext.zip", + "a" * 64, + )