GovOPlaN/govoplan-core
2
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

Add trusted public module catalog installs

Browse Source
This commit is contained in:
zemion
2026-08-06 21:12:54 +02:00
parent 32c234fbdb
commit 9ceb1b8c22
9 changed files with 754 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_core/core/install_config.py
+7 -4
View File
@@ -698,12 +698,15 @@ def _validate_module_catalog_trust(
"A module catalog source is configured without a trusted keyring file.",
"Pin the published GovOPlaN catalog keyring locally and set GOVOPLAN_MODULE_PACKAGE_CATALOG_TRUSTED_KEYS_FILE.",
)
if not _clean(env.get("GOVOPLAN_MODULE_PACKAGE_CATALOG_APPROVED_CHANNEL")):
if not (
_clean(env.get("GOVOPLAN_MODULE_PACKAGE_CATALOG_APPROVED_CHANNELS"))
or _clean(env.get("GOVOPLAN_MODULE_PACKAGE_CATALOG_APPROVED_CHANNEL"))
):
collector.add(
"error",
"GOVOPLAN_MODULE_PACKAGE_CATALOG_APPROVED_CHANNEL",
"GOVOPLAN_MODULE_PACKAGE_CATALOG_APPROVED_CHANNELS",
"A module catalog source is configured without an approved release channel.",
"Set GOVOPLAN_MODULE_PACKAGE_CATALOG_APPROVED_CHANNEL=stable or another approved deployment channel.",
"Set GOVOPLAN_MODULE_PACKAGE_CATALOG_APPROVED_CHANNELS=stable or another approved deployment channel.",
)
@@ -777,7 +780,7 @@ DEV_MAILBOX_API_ENABLED=false
GOVOPLAN_MODULE_PACKAGE_CATALOG_URL=https://govoplan.add-ideas.de/catalogs/v1/channels/stable.json
GOVOPLAN_MODULE_PACKAGE_CATALOG_TRUSTED_KEYS_FILE=/etc/govoplan/catalog-keyring.json
GOVOPLAN_MODULE_PACKAGE_CATALOG_APPROVED_CHANNEL=stable
GOVOPLAN_MODULE_PACKAGE_CATALOG_APPROVED_CHANNELS=stable
"""
src/govoplan_core/core/module_installer.py
+358 -17
View File
@@ -3,7 +3,7 @@ from __future__ import annotations
from collections import defaultdict
from collections.abc import Iterable, Mapping
from contextlib import AbstractContextManager, closing
from dataclasses import dataclass, field
from dataclasses import dataclass, field, replace
from datetime import UTC, datetime
from importlib import metadata
import hashlib
@@ -18,6 +18,7 @@ import sqlite3
import stat
import subprocess # nosec B404 - installer commands are structured and policy-validated before execution.
import sys
import tempfile
import tomllib
from typing import Any, Literal
import time
@@ -62,6 +63,7 @@ MIGRATION_TASK_PHASES = (
MIGRATION_TASK_MUTATING_PHASES = {"pre_migration_prepare", "post_migration_backfill"}
MIGRATION_TASK_REVIEW_SAFETY = {"requires_review", "forward_only", "destructive"}
MIGRATION_TASK_BLOCKING_SAFETY = {"forward_only", "destructive"}
MAX_PACKAGE_ARTIFACT_BYTES = 512 * 1024 * 1024
@dataclass(frozen=True, slots=True)
@@ -464,14 +466,20 @@ def _package_target_action_preflight_issues(
"Python installs must include the distribution package name so rollback can uninstall newly added packages.",
item.module_id,
))
if item.python_ref and not _looks_pinned_dependency_ref(item.python_ref):
if item.python_ref and not (
_looks_pinned_dependency_ref(item.python_ref)
or _artifact_ref_is_digest_pinned(item, "python", item.python_ref)
):
issues.append(ModuleInstallerIssue(
"blocker",
"unpinned_python_ref",
"Python install refs must be pinned to an exact version or tagged git ref.",
item.module_id,
))
if item.webui_ref and not _looks_pinned_dependency_ref(item.webui_ref):
if item.webui_ref and not (
_looks_pinned_dependency_ref(item.webui_ref)
or _artifact_ref_is_digest_pinned(item, "webui", item.webui_ref)
):
issues.append(ModuleInstallerIssue(
"blocker",
"unpinned_webui_ref",
@@ -481,6 +489,23 @@ def _package_target_action_preflight_issues(
return tuple(issues)
def _artifact_ref_is_digest_pinned(
item: ModuleInstallPlanItem,
kind: str,
package_ref: str,
) -> bool:
metadata = _artifact_metadata(item.artifact_integrity, kind)
if metadata is None:
return False
expected_ref = _artifact_text(metadata, "ref") or _artifact_text(metadata, "expected_ref")
sha256 = _artifact_text(metadata, "sha256")
return bool(
expected_ref == package_ref
and sha256
and re.fullmatch(r"[0-9a-f]{64}", sha256.lower())
)
def _frontend_rebuild_preflight_issues(
*,
frontend_rebuild_required: bool,
@@ -519,6 +544,7 @@ def run_module_install_plan(
) -> ModuleInstallerRunResult:
maintenance_mode = saved_maintenance_mode(session)
effective_runtime_dir = runtime_dir or default_installer_runtime_dir(database_url)
effective_plan = plan
preflight = module_install_preflight(
plan=plan,
available=available,
@@ -532,9 +558,30 @@ def run_module_install_plan(
if not preflight.allowed:
raise ModuleInstallerError("Install preflight is blocked: " + "; ".join(issue.message for issue in preflight.issues if issue.severity == "blocker"))
if not dry_run:
effective_plan = acquire_catalog_package_artifacts(
plan,
runtime_dir=effective_runtime_dir,
)
preflight = module_install_preflight(
plan=effective_plan,
available=available,
current_enabled=current_enabled,
desired_enabled=desired_enabled,
maintenance_mode=maintenance_mode.enabled,
session=session,
webui_root=webui_root,
runtime_dir=effective_runtime_dir,
)
if not preflight.allowed:
raise ModuleInstallerError(
"Install preflight is blocked after artifact acquisition: "
+ "; ".join(issue.message for issue in preflight.issues if issue.severity == "blocker")
)
state = _prepare_module_install_run(
session=session,
plan=plan,
plan=effective_plan,
preflight=preflight,
database_url=database_url,
effective_runtime_dir=effective_runtime_dir,
@@ -556,7 +603,7 @@ def run_module_install_plan(
executed, failed_error = _execute_module_install_run(
session=session,
plan=plan,
plan=effective_plan,
available=available,
effective_runtime_dir=effective_runtime_dir,
state=state,
@@ -566,7 +613,7 @@ def run_module_install_plan(
return _failed_module_install_run_result(
session=session,
state=state,
plan=plan,
plan=effective_plan,
executed=executed,
failed_error=failed_error,
effective_runtime_dir=effective_runtime_dir,
@@ -578,7 +625,7 @@ def run_module_install_plan(
return _applied_module_install_run_result(
session=session,
plan=plan,
plan=effective_plan,
desired_enabled=desired_enabled,
activate_installed_modules=activate_installed_modules,
remove_uninstalled_modules_from_desired=remove_uninstalled_modules_from_desired,
@@ -1568,9 +1615,11 @@ def _structured_item_commands(
webui_changed = False
if item.action in PACKAGE_TARGET_ACTIONS:
if item.python_ref:
commands.append(_structured_command([sys.executable, "-m", "pip", "install", item.python_ref], source="module-plan.python"))
python_source = _verified_artifact_install_ref(item, "python") or item.python_ref
commands.append(_structured_command([sys.executable, "-m", "pip", "install", python_source], source="module-plan.python"))
if item.webui_package and item.webui_ref and webui_root is not None:
commands.append(_structured_command([npm_bin, "pkg", "set", f"dependencies.{item.webui_package}={item.webui_ref}"], cwd=webui_root, source="module-plan.webui"))
webui_source = _verified_artifact_install_ref(item, "webui") or item.webui_ref
commands.append(_structured_command([npm_bin, "pkg", "set", f"dependencies.{item.webui_package}={webui_source}"], cwd=webui_root, source="module-plan.webui"))
webui_changed = True
elif item.action == "uninstall":
if item.python_package:
@@ -1581,6 +1630,14 @@ def _structured_item_commands(
return tuple(commands), webui_changed
def _verified_artifact_install_ref(item: ModuleInstallPlanItem, kind: str) -> str | None:
metadata = _artifact_metadata(item.artifact_integrity, kind)
path = _artifact_path(metadata) if metadata is not None else None
if path is None:
return None
return path.as_uri() if kind == "webui" else str(path)
def _structured_webui_followup_commands(
*,
webui_changed: bool,
@@ -1932,9 +1989,7 @@ def _package_catalog_preflight_issues(
return ()
catalog_items = tuple(item for item in package_items if item.source == "catalog")
try:
from govoplan_core.core.module_package_catalog import validate_module_package_catalog
result = validate_module_package_catalog()
result = _validate_catalog_for_plan(catalog_items)
except Exception as exc:
return _catalog_validation_exception_issues(exc, catalog_items=bool(catalog_items))
issues = list(_catalog_validation_result_issues(result, catalog_items=bool(catalog_items)))
@@ -1942,10 +1997,33 @@ def _package_catalog_preflight_issues(
return tuple(issues)
issues.extend(_catalog_warning_issues(result))
if catalog_items:
issues.extend(_catalog_plan_binding_issues(catalog_items, result))
issues.extend(_selected_catalog_interface_issues(catalog_items, result, available))
return tuple(issues)
def _validate_catalog_for_plan(
catalog_items: tuple[ModuleInstallPlanItem, ...],
) -> dict[str, object]:
from govoplan_core.core.module_package_catalog import (
OFFICIAL_MODULE_PACKAGE_CATALOG_URL,
validate_module_package_catalog,
validate_official_module_package_catalog,
)
configured = validate_module_package_catalog()
if configured.get("configured") or not catalog_items:
return configured
sources = {
str(item.catalog.get("source") or "")
for item in catalog_items
if isinstance(item.catalog, Mapping)
}
if sources == {OFFICIAL_MODULE_PACKAGE_CATALOG_URL}:
return validate_official_module_package_catalog()
return configured
def _catalog_validation_exception_issues(exc: Exception, *, catalog_items: bool) -> tuple[ModuleInstallerIssue, ...]:
severity: IssueSeverity = "blocker" if catalog_items else "warning"
return (ModuleInstallerIssue(
@@ -1985,6 +2063,97 @@ def _catalog_warning_issues(result: Mapping[str, object]) -> tuple[ModuleInstall
return tuple(ModuleInstallerIssue("warning", "catalog_warning", str(warning)) for warning in warnings)
def _catalog_plan_binding_issues(
items: tuple[ModuleInstallPlanItem, ...],
validation: Mapping[str, object],
) -> tuple[ModuleInstallerIssue, ...]:
"""Require every trusted plan row to match its signed catalog entry exactly."""
modules = _catalog_modules_by_id(validation)
issues: list[ModuleInstallerIssue] = []
for item in items:
entry = modules.get(item.module_id)
if entry is None or entry.get("action") not in PACKAGE_TARGET_ACTIONS:
issues.append(ModuleInstallerIssue(
"blocker",
"catalog_plan_entry_missing",
f"The validated catalog no longer contains an install or update entry for {item.module_id!r}.",
item.module_id,
))
continue
mismatches = _catalog_plan_entry_mismatches(item, entry, validation)
if mismatches:
issues.append(ModuleInstallerIssue(
"blocker",
"catalog_plan_binding_mismatch",
(
"The saved package plan differs from its validated signed catalog entry "
f"for: {', '.join(mismatches)}. Remove and add the catalog item again."
),
item.module_id,
))
return tuple(issues)
def _catalog_plan_entry_mismatches(
item: ModuleInstallPlanItem,
entry: Mapping[str, object],
validation: Mapping[str, object],
) -> tuple[str, ...]:
mismatches: list[str] = []
for attribute in ("python_package", "python_ref", "webui_package", "webui_ref"):
if getattr(item, attribute) != _catalog_optional_string(entry, attribute):
mismatches.append(attribute)
if _catalog_integrity_identity(item.artifact_integrity) != _catalog_integrity_identity(
entry.get("artifact_integrity")
):
mismatches.append("artifact_integrity")
catalog = item.catalog if isinstance(item.catalog, Mapping) else {}
expected_snapshot = {
"source": validation.get("source") or validation.get("path"),
"channel": validation.get("channel"),
"sequence": validation.get("sequence"),
"signed": bool(validation.get("signed")),
"trusted": bool(validation.get("trusted")),
"key_id": validation.get("key_id"),
}
for attribute, expected in expected_snapshot.items():
actual = catalog.get(attribute)
if actual != expected:
mismatches.append(f"catalog.{attribute}")
return tuple(mismatches)
def _catalog_integrity_identity(value: object) -> dict[str, dict[str, object]]:
if not isinstance(value, Mapping):
return {}
identity: dict[str, dict[str, object]] = {}
for kind in ("python", "webui"):
raw = value.get(kind)
if not isinstance(raw, Mapping):
continue
identity[kind] = {
field: raw.get(field)
for field in (
"ref",
"url",
"filename",
"sha256",
"size",
"integrity",
"sbom_url",
"provenance_url",
"registry_identity",
"git_ref",
"source_commit",
)
if raw.get(field) is not None
}
return identity
def _module_install_target_plan(
plan: ModuleInstallPlan,
available: Mapping[str, ModuleManifest],
@@ -2817,12 +2986,15 @@ def _topological_cycle_ids(incoming: Mapping[str, set[str]]) -> tuple[str, ...]:
def _catalog_modules_for_target_plan(
planned_items: tuple[ModuleInstallPlanItem, ...],
) -> dict[str, Mapping[str, object]]:
if not any(item.source == "catalog" and item.action in PACKAGE_TARGET_ACTIONS for item in planned_items):
catalog_items = tuple(
item
for item in planned_items
if item.source == "catalog" and item.action in PACKAGE_TARGET_ACTIONS
)
if not catalog_items:
return {}
try:
from govoplan_core.core.module_package_catalog import validate_module_package_catalog
result = validate_module_package_catalog()
result = _validate_catalog_for_plan(catalog_items)
except Exception:
return {}
if result.get("valid") is not True:
@@ -3772,6 +3944,156 @@ def _configured_require_artifact_integrity() -> bool:
return os.getenv("GOVOPLAN_MODULE_INSTALLER_REQUIRE_ARTIFACT_INTEGRITY", "").strip().lower() in {"1", "true", "yes", "on"}
def acquire_catalog_package_artifacts(
plan: ModuleInstallPlan,
*,
runtime_dir: Path,
) -> ModuleInstallPlan:
"""Materialize trusted catalog archives before package mutation."""
items: list[ModuleInstallPlanItem] = []
for item in plan.items:
if item.status != "planned" or item.action not in PACKAGE_TARGET_ACTIONS:
items.append(item)
continue
raw_integrity = item.artifact_integrity
if not isinstance(raw_integrity, Mapping):
items.append(item)
continue
integrity: dict[str, object] = dict(raw_integrity)
changed = False
for kind in ("python", "webui"):
metadata = _artifact_metadata(integrity, kind)
if metadata is None or _artifact_path(metadata) is not None:
continue
if not _catalog_artifact_acquisition_ready(item, metadata):
continue
updated = dict(metadata)
updated["artifact_path"] = str(
_acquire_package_artifact(
metadata,
runtime_dir=runtime_dir,
module_id=item.module_id,
kind=kind,
)
)
integrity[kind] = updated
changed = True
items.append(replace(item, artifact_integrity=integrity) if changed else item)
return replace(plan, items=tuple(items))
def _catalog_artifact_acquisition_ready(
item: ModuleInstallPlanItem,
metadata: Mapping[str, object],
) -> bool:
catalog = item.catalog
if (
item.source != "catalog"
or not isinstance(catalog, Mapping)
or catalog.get("signed") is not True
or catalog.get("trusted") is not True
):
return False
url = _artifact_text(metadata, "url")
filename = _artifact_text(metadata, "filename")
sha256 = _artifact_text(metadata, "sha256")
size = metadata.get("size")
return bool(
url
and url.startswith("https://")
and filename
and Path(filename).name == filename
and sha256
and re.fullmatch(r"[0-9a-f]{64}", sha256.lower())
and isinstance(size, int)
and not isinstance(size, bool)
and 0 < size <= MAX_PACKAGE_ARTIFACT_BYTES
)
def _acquire_package_artifact(
metadata: Mapping[str, object],
*,
runtime_dir: Path,
module_id: str,
kind: str,
) -> Path:
url = validate_http_url(_artifact_text(metadata, "url") or "", label=f"{kind.capitalize()} package URL")
if not url.startswith("https://"):
raise ModuleInstallerError(f"{kind.capitalize()} package URL must use HTTPS.")
filename = _artifact_text(metadata, "filename") or ""
expected_sha256 = (_artifact_text(metadata, "sha256") or "").lower()
expected_size = metadata.get("size")
if (
not filename
or Path(filename).name != filename
or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._+!-]{0,255}", filename) is None
or re.fullmatch(r"[0-9a-f]{64}", expected_sha256) is None
or not isinstance(expected_size, int)
or isinstance(expected_size, bool)
or not 0 < expected_size <= MAX_PACKAGE_ARTIFACT_BYTES
):
raise ModuleInstallerError(f"Catalog artifact metadata is incomplete for {module_id}/{kind}.")
cache_root = runtime_dir / "artifacts"
_ensure_private_artifact_directory(cache_root)
digest_root = cache_root / expected_sha256
_ensure_private_artifact_directory(digest_root)
target = digest_root / filename
if target.exists() or target.is_symlink():
if target.is_symlink() or not target.is_file():
raise ModuleInstallerError(f"Cached package artifact is not a regular file: {target}")
if target.stat().st_size != expected_size or _sha256_file(target) != expected_sha256:
raise ModuleInstallerError(f"Cached package artifact does not match its catalog identity: {target}")
return target
try:
response = fetch_http(
url,
timeout=float(os.getenv("GOVOPLAN_MODULE_INSTALLER_DOWNLOAD_TIMEOUT_SECONDS", "120")),
label=f"{module_id} {kind} package URL",
max_bytes=min(expected_size + 1, MAX_PACKAGE_ARTIFACT_BYTES),
)
except (OSError, ValueError) as exc:
raise ModuleInstallerError(f"Could not download {module_id} {kind} package: {exc}") from exc
if response.status < 200 or response.status >= 300:
raise ModuleInstallerError(f"Could not download {module_id} {kind} package: HTTP {response.status}.")
if len(response.body) != expected_size or hashlib.sha256(response.body).hexdigest() != expected_sha256:
raise ModuleInstallerError(f"Downloaded {module_id} {kind} package does not match its signed catalog identity.")
temporary_path: Path | None = None
try:
with tempfile.NamedTemporaryFile(
mode="wb",
prefix=f".{filename}.",
suffix=".tmp",
dir=digest_root,
delete=False,
) as handle:
temporary_path = Path(handle.name)
handle.write(response.body)
handle.flush()
os.fsync(handle.fileno())
temporary_path.chmod(0o600)
os.replace(temporary_path, target)
target.chmod(0o600)
except OSError as exc:
if temporary_path is not None:
temporary_path.unlink(missing_ok=True)
raise ModuleInstallerError(f"Could not cache {module_id} {kind} package.") from exc
return target
def _ensure_private_artifact_directory(path: Path) -> None:
if path.is_symlink():
raise ModuleInstallerError(f"Installer artifact cache must not be a symlink: {path}")
path.mkdir(parents=True, mode=0o700, exist_ok=True)
path.chmod(0o700)
if not path.is_dir() or stat.S_IMODE(path.stat().st_mode) != 0o700:
raise ModuleInstallerError(f"Installer artifact cache is not private: {path}")
def _verify_artifact_integrity(
planned_items: tuple[ModuleInstallPlanItem, ...],
*,
@@ -3839,7 +4161,17 @@ def _verify_artifact_metadata(
}
if package_name:
record["package"] = package_name
for key in ("sha256", "sbom_url", "provenance_url", "registry_identity", "git_ref"):
for key in (
"sha256",
"url",
"filename",
"integrity",
"sbom_url",
"provenance_url",
"registry_identity",
"git_ref",
"source_commit",
):
value = _artifact_text(metadata, key)
if value:
record[key] = value
@@ -3860,6 +4192,15 @@ def _verify_artifact_metadata(
item.module_id,
))
return record, tuple(issues)
if artifact_path is None and _catalog_artifact_acquisition_ready(item, metadata):
record["acquisition_pending"] = True
issues.append(ModuleInstallerIssue(
"info",
"artifact_acquisition_pending",
f"{kind.capitalize()} artifact will be downloaded and verified by the installer daemon before package mutation.",
item.module_id,
))
return record, tuple(issues)
if artifact_path is None:
issues.append(ModuleInstallerIssue(
"blocker" if require_verified else "warning",
src/govoplan_core/core/module_package_catalog.py
+42 -2
View File
@@ -6,6 +6,7 @@ from collections import defaultdict
from collections.abc import Mapping
from dataclasses import dataclass
from datetime import UTC, datetime
from importlib.resources import files
from pathlib import Path
import json
import os
@@ -29,6 +30,8 @@ from govoplan_core.core.provider_governance import (
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}$")
CATALOG_MIGRATION_SAFETY = ("automatic", "requires_review", "forward_only", "destructive")
CATALOG_MIGRATION_TASK_PHASES = (
"pre_migration_check",
@@ -36,6 +39,8 @@ CATALOG_MIGRATION_TASK_PHASES = (
"post_migration_backfill",
"post_migration_verify",
)
OFFICIAL_MODULE_PACKAGE_CATALOG_URL = "https://govoplan.add-ideas.de/catalogs/v1/channels/stable.json"
OFFICIAL_MODULE_PACKAGE_CATALOG_CHANNEL = "stable"
@dataclass(frozen=True, slots=True)
@@ -101,6 +106,18 @@ def validate_module_package_catalog(
return _valid_catalog_result(catalog_source, state)
def validate_official_module_package_catalog() -> dict[str, object]:
"""Read the public GovOPlaN directory against Core's pinned trust anchor."""
keyring = files("govoplan_core").joinpath("resources/catalog-keyring.json").read_text(encoding="utf-8")
return validate_module_package_catalog(
OFFICIAL_MODULE_PACKAGE_CATALOG_URL,
require_trusted=True,
approved_channels=(OFFICIAL_MODULE_PACKAGE_CATALOG_CHANNEL,),
trusted_keys=_parse_trusted_keys(keyring),
)
def _catalog_validation_state(
source: Path | str | None,
*,
@@ -323,7 +340,10 @@ def _configured_require_signature() -> bool:
def _configured_approved_channels() -> tuple[str, ...]:
value = os.getenv("GOVOPLAN_MODULE_PACKAGE_CATALOG_APPROVED_CHANNELS", "").strip()
value = (
os.getenv("GOVOPLAN_MODULE_PACKAGE_CATALOG_APPROVED_CHANNELS", "").strip()
or os.getenv("GOVOPLAN_MODULE_PACKAGE_CATALOG_APPROVED_CHANNEL", "").strip()
)
if not value:
return ()
return tuple(item.strip() for item in value.split(",") if item.strip())
@@ -969,20 +989,40 @@ def _normalize_artifact_integrity(value: Any) -> dict[str, object]:
continue
if not isinstance(raw, dict):
raise ValueError(f"Module package catalog artifact_integrity.{key} must be an object.")
clean = {
clean: dict[str, object] = {
field: text
for field in (
"ref",
"path",
"artifact_path",
"url",
"filename",
"sha256",
"integrity",
"sbom_url",
"provenance_url",
"registry_identity",
"git_ref",
"source_commit",
)
if (text := _optional_str(raw, field))
}
url = clean.get("url")
if isinstance(url, str) and (not is_http_url(url) or not url.startswith("https://")):
raise ValueError(f"Module package catalog artifact_integrity.{key}.url must use HTTPS.")
filename = clean.get("filename")
if isinstance(filename, str) and _ARTIFACT_FILENAME_RE.fullmatch(filename) is None:
raise ValueError(f"Module package catalog artifact_integrity.{key}.filename is invalid.")
sha256 = clean.get("sha256")
if isinstance(sha256, str) and _SHA256_RE.fullmatch(sha256.lower()) is None:
raise ValueError(f"Module package catalog artifact_integrity.{key}.sha256 is invalid.")
if isinstance(sha256, str):
clean["sha256"] = sha256.lower()
size = raw.get("size")
if size is not None:
if not isinstance(size, int) or isinstance(size, bool) or size <= 0 or size > 512 * 1024 * 1024:
raise ValueError(f"Module package catalog artifact_integrity.{key}.size is invalid.")
clean["size"] = size
if clean:
normalized[key] = clean
return normalized