fix(assessment): establish independent release trust

tools/release/govoplan_release/catalog.py

@@ -7,11 +7,12 @@ import json
 from pathlib import Path
 from urllib.error import HTTPError, URLError
 
-from .http_fetch import fetch_http
+from .http_fetch import DEFAULT_MAX_RESPONSE_BYTES, fetch_http
 from .model import CatalogSnapshot
 from .workspace import website_root
 
 DEFAULT_PUBLIC_BASE_URL = "https://govoplan.add-ideas.de"
+MAX_PUBLIC_JSON_RESPONSE_BYTES = DEFAULT_MAX_RESPONSE_BYTES
 
 
 def collect_catalog_snapshot(
@@ -22,9 +23,13 @@ def collect_catalog_snapshot(
     check_public: bool = False,
 ) -> CatalogSnapshot:
     web_root = website_root(workspace_root)
-    catalog_path = web_root / "public" / "catalogs" / "v1" / "channels" / f"{channel}.json"
+    catalog_path = (
+        web_root / "public" / "catalogs" / "v1" / "channels" / f"{channel}.json"
+    )
     keyring_path = web_root / "public" / "catalogs" / "v1" / "keyring.json"
-    public_catalog_url = f"{public_base_url.rstrip('/')}/catalogs/v1/channels/{channel}.json"
+    public_catalog_url = (
+        f"{public_base_url.rstrip('/')}/catalogs/v1/channels/{channel}.json"
+    )
     public_keyring_url = f"{public_base_url.rstrip('/')}/catalogs/v1/keyring.json"
 
     catalog_version: str | None = None
@@ -129,8 +134,12 @@ def collect_catalog_snapshot(
         public_keyring_hash=public_keyring_hash,
         public_catalog_error=public_catalog_error,
         public_keyring_error=public_keyring_error,
-        catalog_matches_public=compare_hashes(local_catalog_hash, public_catalog_hash) if check_public else None,
-        keyring_matches_public=compare_hashes(local_keyring_hash, public_keyring_hash) if check_public else None,
+        catalog_matches_public=compare_hashes(local_catalog_hash, public_catalog_hash)
+        if check_public
+        else None,
+        keyring_matches_public=compare_hashes(local_keyring_hash, public_keyring_hash)
+        if check_public
+        else None,
     )
 
 
@@ -149,7 +158,9 @@ def parse_catalog_payload(payload: object) -> dict[str, str | int | None]:
     signatures = payload.get("signatures")
     return {
         "catalog_version": string_or_none(payload.get("catalog_version")),
-        "core_release_version": string_or_none(core_release.get("version")) if isinstance(core_release, dict) else None,
+        "core_release_version": string_or_none(core_release.get("version"))
+        if isinstance(core_release, dict)
+        else None,
         "module_count": len(modules) if isinstance(modules, list) else 0,
         "signature_count": len(signatures) if isinstance(signatures, list) else 0,
         "generated_at": string_or_none(payload.get("generated_at")),
@@ -164,6 +175,7 @@ def fetch_json(url: str) -> dict[str, object]:
             timeout=10,
             label="Public release catalog URL",
             headers={"User-Agent": "GovOPlaN release-console"},
+            max_response_bytes=MAX_PUBLIC_JSON_RESPONSE_BYTES,
         )
         return {"ok": True, "payload": json.loads(response.text())}
     except (HTTPError, URLError, TimeoutError, ValueError, json.JSONDecodeError) as exc:
@@ -171,7 +183,9 @@ def fetch_json(url: str) -> dict[str, object]:
 
 
 def canonical_hash(payload: object) -> str:
-    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8")
+    canonical = json.dumps(
+        payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True
+    ).encode("utf-8")
     return hashlib.sha256(canonical).hexdigest()
 
 

tools/release/govoplan_release/http_fetch.py

@@ -8,6 +8,9 @@ from dataclasses import dataclass
 from typing import Mapping
 
 
+DEFAULT_MAX_RESPONSE_BYTES = 16 * 1024 * 1024
+
+
 @dataclass(frozen=True, slots=True)
 class HttpFetchResponse:
     status: int
@@ -34,11 +37,29 @@ def fetch_http(
     label: str = "URL",
     method: str = "GET",
     headers: Mapping[str, str] | None = None,
+    max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES,
 ) -> HttpFetchResponse:
-    request = urllib.request.Request(validate_http_url(url, label=label), headers=dict(headers or {}), method=method)
+    if max_response_bytes <= 0:
+        raise ValueError("max_response_bytes must be positive.")
+    request = urllib.request.Request(
+        validate_http_url(url, label=label),
+        headers=dict(headers or {}),
+        method=method,
+    )
     with urllib.request.urlopen(request, timeout=timeout) as response:  # noqa: S310 - URL is validated by validate_http_url.  # nosec B310  # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
+        raw_length = response.headers.get("Content-Length")
+        if raw_length is not None:
+            try:
+                declared_length = int(raw_length)
+            except (TypeError, ValueError):
+                declared_length = None
+            if declared_length is not None and declared_length > max_response_bytes:
+                raise ValueError(f"{label} response exceeds the configured size limit.")
+        body = response.read(max_response_bytes + 1)
+        if len(body) > max_response_bytes:
+            raise ValueError(f"{label} response exceeds the configured size limit.")
         return HttpFetchResponse(
             status=int(getattr(response, "status", 0)),
             headers=dict(response.headers.items()),
-            body=response.read(),
+            body=body,
         )