Validate module catalog provenance and availability

src/govoplan_core/core/module_package_catalog.py

@@ -32,6 +32,9 @@ from govoplan_core.security.http_fetch import fetch_http_text, is_http_url
 _INTERFACE_NAME_RE = re.compile(r"^[a-z][a-z0-9_]*(?:\.[a-z][a-z0-9_]*)+$")
 _SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
 _ARTIFACT_FILENAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._+!-]{0,255}$")
+_SOURCE_REPOSITORY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,254}$")
+_SOURCE_REF_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/+!-]{0,127}$")
+_SOURCE_COMMIT_RE = re.compile(r"^(?:[0-9a-f]{40}|[0-9a-f]{64})$")
 CATALOG_MIGRATION_SAFETY = ("automatic", "requires_review", "forward_only", "destructive")
 CATALOG_MIGRATION_TASK_PHASES = (
     "pre_migration_check",
@@ -652,7 +655,23 @@ def _normalize_catalog_item(value: Any) -> dict[str, object]:
         "requires_interfaces": _normalize_catalog_interface_requirements(value.get("requires_interfaces"), module_id=module_id),
         "notes": _optional_str(value, "notes"),
         "tags": _string_list(value.get("tags")),
+        "availability": _catalog_availability(value, module_id=module_id),
+        "availability_reason": _optional_str(value, "availability_reason"),
+        "configuration_requirements": _string_list(value.get("configuration_requirements")),
     }
+    if item["availability"] == "withdrawn" and not item["availability_reason"]:
+        raise ValueError(
+            f"Withdrawn module package catalog entry {module_id!r} requires availability_reason."
+        )
+    release_notes_url = _optional_str(value, "release_notes_url")
+    if release_notes_url is not None:
+        item["release_notes_url"] = _catalog_https_url(
+            release_notes_url,
+            label=f"Module package catalog release_notes_url for {module_id!r}",
+        )
+    source = _normalize_catalog_source(value.get("source"), module_id=module_id)
+    if source:
+        item["source"] = source
     raw_architecture = value.get("architecture")
     architecture_maturity: str | None = None
     if raw_architecture is not None:
@@ -756,6 +775,67 @@ def _normalize_catalog_item(value: Any) -> dict[str, object]:
     return item
 
 
+def _catalog_availability(value: Mapping[str, object], *, module_id: str) -> str:
+    availability = str(value.get("availability") or "available").strip().lower()
+    if availability not in {"available", "withdrawn"}:
+        raise ValueError(
+            f"Unsupported catalog availability for {module_id!r}: {availability!r}."
+        )
+    return availability
+
+
+def _normalize_catalog_source(
+    value: object,
+    *,
+    module_id: str,
+) -> dict[str, object]:
+    if value is None:
+        return {}
+    if not isinstance(value, Mapping):
+        raise ValueError(
+            f"Module package catalog source for {module_id!r} must be an object."
+        )
+    repository = _required_str(value, "repository")
+    tag = _required_str(value, "tag")
+    commit = _required_str(value, "commit").lower()
+    if (
+        _SOURCE_REPOSITORY_RE.fullmatch(repository) is None
+        or repository.startswith("/")
+        or repository.endswith("/")
+        or ".." in repository.split("/")
+    ):
+        raise ValueError(
+            f"Module package catalog source repository for {module_id!r} is invalid."
+        )
+    if _SOURCE_REF_RE.fullmatch(tag) is None or ".." in tag.split("/"):
+        raise ValueError(
+            f"Module package catalog source tag for {module_id!r} is invalid."
+        )
+    if _SOURCE_COMMIT_RE.fullmatch(commit) is None:
+        raise ValueError(
+            f"Module package catalog source commit for {module_id!r} is invalid."
+        )
+    source: dict[str, object] = {
+        "repository": repository,
+        "tag": tag,
+        "commit": commit,
+    }
+    for field in ("repository_url", "revision_url"):
+        url = _optional_str(value, field)
+        if url is not None:
+            source[field] = _catalog_https_url(
+                url,
+                label=f"Module package catalog source {field} for {module_id!r}",
+            )
+    return source
+
+
+def _catalog_https_url(value: str, *, label: str) -> str:
+    if not is_http_url(value) or not value.startswith("https://"):
+        raise ValueError(f"{label} must use HTTPS.")
+    return value
+
+
 def _catalog_migration_safety(value: Any, *, module_id: str) -> str:
     if value is None:
         return "automatic"
@@ -826,14 +906,14 @@ def _catalog_optional_positive_int(value: dict[str, Any], key: str, *, module_id
     return integer
 
 
-def _required_str(value: dict[str, Any], key: str) -> str:
+def _required_str(value: Mapping[str, Any], key: str) -> str:
     item = _optional_str(value, key)
     if not item:
         raise ValueError(f"Module package catalog entry is missing {key!r}.")
     return item
 
 
-def _optional_str(value: dict[str, Any], key: str) -> str | None:
+def _optional_str(value: Mapping[str, Any], key: str) -> str | None:
     item = value.get(key)
     if item is None:
         return None