GovOPlaN/govoplan
2
0
Fork 0
Code Issues 18 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

feat(release): build verified full-registry catalog candidates

Browse Source
This commit is contained in:
zemion
2026-09-08 02:26:19 +02:00
parent 32fe4b7238
commit 6bcb75f577
13 changed files with 1338 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/release/govoplan_release/artifact_identity.py
+23
View File
@@ -14,6 +14,8 @@ import re
import stat
import zipfile
from .registry_reference import registry_artifact_conflicts, registry_entry_source
WHEEL_PAYLOAD_ALGORITHM = "govoplan-wheel-declared-payload-v1"
INSTALLED_PAYLOAD_ALGORITHM = "govoplan-installed-record-payload-v1"
@@ -185,10 +187,23 @@ def selected_artifact_identity_issues(payload: object) -> tuple[str, ...]:
modules = payload.get("modules")
if isinstance(modules, list):
entries.extend(modules)
conflicts = registry_artifact_conflicts(entries)
if conflicts:
return conflicts
package_by_repo: dict[str, tuple[str, str]] = {}
registry_artifacts: dict[str, dict[str, object]] = {}
for entry in entries:
if not isinstance(entry, dict):
continue
try:
registry_source = registry_entry_source(entry)
except ValueError as exc:
return (str(exc),)
if registry_source is not None:
package = str(entry["python_package"])
package_by_repo[registry_source.repository] = (package, registry_source.version)
registry_artifacts[package] = entry["artifact_integrity"]["python"]
continue
python_ref = entry.get("python_ref")
match = _PYTHON_REF.search(python_ref) if isinstance(python_ref, str) else None
package_name = entry.get("python_package")
@@ -219,6 +234,14 @@ def selected_artifact_identity_issues(payload: object) -> tuple[str, ...]:
issues: list[str] = []
if malformed_artifacts:
issues.append("release.artifacts contains malformed or duplicate identities")
for package, registry_artifact in registry_artifacts.items():
artifact = artifacts_by_package.get(package)
if artifact is None or (
artifact.get("archive_sha256") != registry_artifact.get("sha256")
or artifact.get("archive_size") != registry_artifact.get("size")
or artifact.get("package_version") != registry_artifact["registry_identity"].rsplit("@", 1)[-1]
):
issues.append(f"registry artifact {package} has no matching inspected wheel byte identity")
seen_repos: set[str] = set()
for unit in selected_units:
if not isinstance(unit, dict):
tools/release/govoplan_release/catalog_entry_synthesis.py
+5 -5
View File
@@ -19,6 +19,7 @@ from typing import Iterator
from govoplan_core.core.modules import ModuleManifest
from govoplan_core.core.versioning import version_satisfies_range
from .git_state import sanitized_git_environment, scoped_git_command
from .workspace import load_repository_specs, resolve_repo_path
@@ -255,20 +256,19 @@ def materialized_source_tree(root: Path, *, source_ref: str | None) -> Iterator[
source_root = temporary / "source"
source_root.mkdir()
result = subprocess.run(
[
"git",
"-C",
str(root),
scoped_git_command(
root, "-C", str(root),
"archive",
"--format=tar",
f"--output={archive_path}",
source_ref,
],
),
check=False,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
text=True,
timeout=30,
env=sanitized_git_environment(),
)
if result.returncode != 0:
detail = result.stderr.strip() or "Git archive failed"
tools/release/govoplan_release/full_catalog.py
+389
View File
@@ -0,0 +1,389 @@
"""Build a private full-profile candidate from exact verified registry bytes."""
from __future__ import annotations
import base64
from datetime import UTC, datetime, timedelta
from functools import lru_cache
import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import re
import runpy
import stat
import tarfile
from typing import Any
from urllib.parse import quote
from .artifact_identity import inspect_python_wheel, selected_artifact_identity_issues
from .candidate_artifact import (
ensure_private_candidate_root, harden_private_candidate_tree,
validate_release_channel,
)
from .catalog import canonical_hash
from .module_directory import write_module_directory
from .selective_catalog import (
authenticate_base_catalog_signatures, authenticate_base_keyring,
configured_signer_public_keys, enforce_selected_version_alignment,
next_sequence, parse_signing_key, read_bounded_json_source, signature,
validate_catalog_object,
)
from .source_provenance import (
registered_source_origin_issues, selected_source_provenance,
source_tag_provenance_issues,
)
from .version_alignment import candidate_catalog_version_issues
from .workspace import META_ROOT, resolve_workspace_root, website_root
MAX_ARTIFACT_BYTES = 512 * 1024 * 1024
@lru_cache(maxsize=3)
def _tool(name: str) -> dict[str, Any]:
# These are fixed, operator-controlled release modules, not caller paths.
if name not in {
"generate-release-package-set", "generate-release-catalog",
"resolve-package-artifacts",
}:
raise ValueError("unknown registry release tool")
return runpy.run_path(str(META_ROOT / "tools" / "release" / f"{name}.py"))
def authenticate_full_rebuild_base(
*, web_root: Path, channel: str, signer_public_keys: dict[str, str],
) -> tuple[dict[str, Any], dict[str, Any]]:
"""Authenticate the fixed website pair without reusing legacy entry data.
Only this full rebuild may migrate a legacy catalog without a keyring hash.
In that case *every* active website key must exactly equal a configured
signer; injecting an additional website key cannot extend trust. Selective
candidates retain their stricter existing hash-pinned-base requirement.
"""
root = web_root / "public" / "catalogs" / "v1"
catalog = read_bounded_json_source(root / "channels" / f"{channel}.json", label="published base catalog")
keyring = read_bounded_json_source(root / "keyring.json", label="published website keyring")
if not isinstance(catalog, dict) or not isinstance(keyring, dict):
raise ValueError("published website catalog/keyring must be objects")
trusted_keys = authenticate_base_keyring(keyring)
release = catalog.get("release")
pinned = release.get("keyring_sha256") if isinstance(release, dict) else None
if pinned is None:
if trusted_keys != signer_public_keys or len(keyring["keys"]) != len(trusted_keys):
raise ValueError("legacy full rebuild requires exactly the configured known website signers")
elif pinned != canonical_hash(keyring):
raise ValueError("published base catalog does not pin its exact website keyring")
if any(trusted_keys.get(key) != value for key, value in signer_public_keys.items()):
raise ValueError("full rebuild cannot introduce or replace a website signer")
authenticate_base_catalog_signatures(
catalog, base_trusted_keys=trusted_keys, configured_signers=signer_public_keys,
)
validation = validate_catalog_object(
catalog, approved_channel=channel, signer_public_keys=signer_public_keys,
)
if validation.get("valid") is not True:
raise ValueError(f"published base catalog failed validation: {validation.get('error')}")
return catalog, keyring
def build_full_registry_candidate(
*, package_set_path: Path, package_lock_path: Path,
wheelhouse: Path, webui_packages: Path, output_dir: Path,
selected_repositories: tuple[str, ...], signing_keys: tuple[str, ...],
workspace_root: Path | str | None = None, channel: str = "stable",
source_remote: str = "origin", public_base_url: str = "https://govoplan.add-ideas.de",
expires_days: int = 90, sequence: int | None = None,
) -> dict[str, object]:
channel = validate_release_channel(channel)
if not isinstance(expires_days, int) or isinstance(expires_days, bool) or not 1 <= expires_days <= 365:
raise ValueError("catalog expiry must be between 1 and 365 days")
workspace = resolve_workspace_root(workspace_root)
ensure_private_candidate_root(workspace)
output = output_dir.expanduser().absolute()
ensure_private_candidate_root(output.parent, create=True)
if output.exists() or output.is_symlink():
raise ValueError("full candidate output must not already exist")
parsed_keys = tuple(parse_signing_key(value) for value in signing_keys)
if not parsed_keys:
raise ValueError("full candidate needs a configured signing key")
signer_keys = configured_signer_public_keys(parsed_keys)
base, keyring = authenticate_full_rebuild_base(
web_root=website_root(workspace), channel=channel, signer_public_keys=signer_keys,
)
package_set = _hashed_json(package_set_path, "package_set_sha256")
lock = _hashed_json(package_lock_path, "lock_sha256")
generator = _tool("generate-release-catalog")
version = package_set.get("release_version")
if package_set.get("profile") != "full" or not isinstance(version, str):
raise ValueError("full candidate requires the complete full-profile package set")
expected = _tool("generate-release-package-set")["generate_package_set"](
core_version=version, requirements=META_ROOT / "requirements-release.txt",
workspace=workspace, profile="full",
meta_package=META_ROOT / "packages/govoplan-meta/pyproject.toml",
)
if package_set != expected:
raise ValueError("package set differs from exact Meta full pins or immutable tag metadata")
generator["_validate_release_inputs"](package_set, lock, core_version=version)
if lock.get("registries") != package_set.get("registries"):
raise ValueError("artifact lock uses different package registries")
versions = {row["repository"]: row["version"] for row in package_set["python"]}
origin_failures = registered_source_origin_issues(
repo_versions=versions, workspace=workspace, remote=source_remote,
)
if origin_failures:
raise ValueError("Registered source origin gate failed: " + "; ".join(item.describe() for item in origin_failures))
selected = set(selected_repositories)
if not selected or len(selected) != len(selected_repositories) or not selected <= versions.keys():
raise ValueError("selected repositories must be unique members of the full package set")
selected_versions = {repo: versions[repo] for repo in sorted(selected)}
enforce_selected_version_alignment(repo_versions=selected_versions, workspace=workspace)
failures = source_tag_provenance_issues(
repo_versions=versions, workspace=workspace, remote=source_remote,
require_head_repos=selected,
)
if failures:
raise ValueError("Full source provenance gate failed: " + "; ".join(item.describe() for item in failures))
provenance = selected_source_provenance(repo_versions=versions, workspace=workspace)
wheel_identities = verify_registry_artifacts(
package_set=package_set, lock=lock, wheelhouse=wheelhouse,
webui_packages=webui_packages,
)
generated_at = datetime.now(tz=UTC)
resolved_sequence = sequence if sequence is not None else next_sequence(base, generated_at=generated_at)
if isinstance(resolved_sequence, bool) or not isinstance(resolved_sequence, int) or resolved_sequence <= int(base.get("sequence") or 0):
raise ValueError("full candidate sequence must advance the authenticated published channel")
# Fresh tagged manifests, including unchanged tagged ancestors; no legacy
# entry, registry hash, source URL or dependency contract is carried over.
candidate = generator["_catalog_payload"](
package_set=package_set, package_lock=lock, channel=channel,
sequence=resolved_sequence, generated_at=generated_at,
expires_at=generated_at + timedelta(days=expires_days), workspace=workspace,
public_base_url=public_base_url.rstrip("/"),
)
for entry in [candidate["core_release"], *candidate["modules"]]:
repo = entry["python_package"]
entry["source"] = {
"repository": repo, "tag": f"v{versions[repo]}",
"commit": provenance[repo]["commit_sha"],
"tag_object_sha": provenance[repo]["tag_object_sha"],
"repository_url": f"https://git.add-ideas.de/GovOPlaN/{repo}",
"revision_url": f"https://git.add-ideas.de/GovOPlaN/{repo}/commit/{provenance[repo]['commit_sha']}",
}
candidate["release"].update({
"selected_units": [
{"repo": repo, "version": versions[repo], "tag": f"v{versions[repo]}", **provenance[repo]}
for repo in sorted(selected)
],
"keyring_sha256": canonical_hash(keyring),
"artifacts": wheel_identities,
"base_catalog_sha256": canonical_hash(base),
})
# Recheck the exact objects used by synthesis, including unchanged source
# ancestors, before signing. No late tag movement can change this candidate.
origin_failures = registered_source_origin_issues(
repo_versions=versions, workspace=workspace, remote=source_remote,
)
if origin_failures:
raise ValueError("Registered source origin changed during synthesis: " + "; ".join(item.describe() for item in origin_failures))
failures = source_tag_provenance_issues(
repo_versions=versions, workspace=workspace, remote=source_remote,
require_head_repos=selected,
expected_commits={repo: row["commit_sha"] for repo, row in provenance.items()},
expected_tag_objects={repo: row["tag_object_sha"] for repo, row in provenance.items()},
)
if failures:
raise ValueError("Full source provenance changed during synthesis: " + "; ".join(item.describe() for item in failures))
failures = candidate_catalog_version_issues(candidate)
identity_failures = selected_artifact_identity_issues(candidate)
if failures or identity_failures:
raise ValueError("full candidate identity validation failed: " + "; ".join(
[item.message for item in failures] + list(identity_failures)
))
candidate["signatures"] = [signature(candidate, key_id=key, private_key=value) for key, value in parsed_keys]
validation = validate_catalog_object(candidate, approved_channel=channel, signer_public_keys=signer_keys)
if validation.get("valid") is not True:
raise ValueError(f"signed full candidate failed validation: {validation.get('error')}")
# Exclusive output creation preserves earlier reviewed candidates.
output.mkdir(mode=0o700)
(output / "channels").mkdir(mode=0o700)
catalog_path = output / "channels" / f"{channel}.json"
_write_private_json(catalog_path, candidate)
_write_private_json(output / "keyring.json", keyring)
write_module_directory(
catalog_payload=candidate, keyring_payload=keyring, output_root=output,
channel=channel, public_base_url=public_base_url,
)
result = {
"status": "ready", "candidate_dir": str(output), "catalog_path": str(catalog_path),
"channel": channel, "sequence": resolved_sequence,
"package_count": len(package_set["python"]), "webui_count": len(package_set["webui"]),
"selected_count": len(selected), "candidate_catalog_hash": canonical_hash(candidate),
"candidate_keyring_hash": canonical_hash(keyring), "validation_valid": True,
}
_write_private_json(output / "summary.json", result)
harden_private_candidate_tree(output)
return result
def _hashed_json(path: Path, field: str) -> dict[str, Any]:
payload = read_bounded_json_source(path, label=field)
if not isinstance(payload, dict):
raise ValueError(f"{field} input must be an object")
unsigned = dict(payload)
expected = unsigned.pop(field, None)
# Registry tools use their established ASCII-escaped canonical form.
encoded = json.dumps(unsigned, sort_keys=True, separators=(",", ":")).encode()
if expected != hashlib.sha256(encoded).hexdigest():
raise ValueError(f"{field} does not match its contents")
return payload
def verify_registry_artifacts(
*, package_set: dict[str, Any], lock: dict[str, Any],
wheelhouse: Path, webui_packages: Path,
) -> list[dict[str, object]]:
"""Compare exact registry bytes, metadata and source bindings without installs."""
identities = []
for group, root, suffix in (("python", wheelhouse, ".whl"), ("webui", webui_packages, ".tgz")):
ensure_private_candidate_root(root)
expected = {row["name"]: row for row in package_set[group]}
rows = lock[group]
if not isinstance(rows, list) or len(rows) != len(expected):
raise ValueError(f"artifact lock has duplicate/missing {group} rows")
filenames: set[str] = set()
seen: set[str] = set()
for row in rows:
selected = expected.get(row.get("name")) if isinstance(row, dict) else None
if selected is None or row["name"] in seen:
raise ValueError(f"artifact lock has unexpected/duplicate {group} identities")
seen.add(row["name"])
for key in ("name", "version", "repository", "tag", "commit"):
if row.get(key) != selected[key]:
raise ValueError("artifact lock differs from selected source identity")
filename = row.get("filename")
if not isinstance(filename, str) or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._+-]{0,254}", filename) is None or not filename.endswith(suffix) or filename in filenames:
raise ValueError("artifact filename is invalid or duplicated")
filenames.add(filename)
path = root / filename
sha256, sha512, size = _hash_artifact(path)
if (sha256, size) != (row.get("sha256"), row.get("size")):
raise ValueError(f"registry artifact bytes differ from locked identity: {filename}")
if group == "python":
expected_url = _tool("resolve-package-artifacts")["_python_artifact_url"](
package_set["registries"]["python"], package=selected, filename=filename,
)
if row.get("url") != expected_url:
raise ValueError("Python artifact URL differs from the selected registry")
identity = inspect_python_wheel(path)
if (identity.package_name, identity.package_version, identity.archive_sha256, identity.archive_size) != (row["name"], row["version"], sha256, size):
raise ValueError("wheel metadata or bytes differ from the registry lock")
identities.append(identity.catalog_payload())
else:
expected_url = (
package_set["registries"]["npm"].rstrip("/") + "/"
+ quote(row["name"], safe="") + "/-/"
+ quote(row["version"], safe="") + "/"
+ quote(row["name"].split("/", 1)[1] + "-" + row["version"] + ".tgz", safe="")
)
if row.get("url") != expected_url:
raise ValueError("WebUI artifact URL differs from the selected registry")
if row.get("integrity") != "sha512-" + base64.b64encode(sha512).decode("ascii"):
raise ValueError("WebUI registry integrity differs from downloaded bytes")
_inspect_npm_metadata(
path, name=row["name"], version=row["version"],
expected_identity=(sha256, sha512, size),
)
actual = set()
for index, path in enumerate(root.iterdir()):
if index >= 1000:
raise ValueError("registry artifact directory exceeds its inspection bound")
actual.add(path.name)
if actual != filenames:
raise ValueError(f"registry directory contains unexpected or missing {group} files")
return sorted(identities, key=lambda row: str(row["package_name"]))
def _hash_artifact(path: Path) -> tuple[str, bytes, int]:
descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0))
try:
initial = os.fstat(descriptor)
if not stat.S_ISREG(initial.st_mode) or not 0 < initial.st_size <= MAX_ARTIFACT_BYTES:
raise ValueError("registry artifact must be a bounded regular file")
digest, sri, total = hashlib.sha256(), hashlib.sha512(), 0
while chunk := os.read(descriptor, 1024 * 1024):
total += len(chunk)
if total > MAX_ARTIFACT_BYTES:
raise ValueError("registry artifact exceeds its byte bound")
digest.update(chunk)
sri.update(chunk)
final = os.fstat(descriptor)
if (initial.st_ino, initial.st_size, initial.st_mtime_ns, initial.st_ctime_ns) != (final.st_ino, total, final.st_mtime_ns, final.st_ctime_ns):
raise ValueError("registry artifact changed while being inspected")
return digest.hexdigest(), sri.digest(), total
finally:
os.close(descriptor)
def _inspect_npm_metadata(
path: Path, *, name: str, version: str,
expected_identity: tuple[str, bytes, int] | None = None,
) -> None:
descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0))
try:
initial = os.fstat(descriptor)
if not stat.S_ISREG(initial.st_mode) or not 0 < initial.st_size <= MAX_ARTIFACT_BYTES:
raise ValueError("WebUI archive must be a bounded regular file")
digest, sri, total = hashlib.sha256(), hashlib.sha512(), 0
while chunk := os.read(descriptor, 1024 * 1024):
total += len(chunk)
if total > MAX_ARTIFACT_BYTES:
raise ValueError("WebUI archive exceeds its byte bound")
digest.update(chunk)
sri.update(chunk)
if expected_identity is not None and (digest.hexdigest(), sri.digest(), total) != expected_identity:
raise ValueError("WebUI archive changed before metadata inspection")
os.lseek(descriptor, 0, os.SEEK_SET)
with os.fdopen(os.dup(descriptor), "rb") as stream:
_inspect_npm_stream(stream, name=name, version=version)
final = os.fstat(descriptor)
if (initial.st_ino, initial.st_size, initial.st_mtime_ns, initial.st_ctime_ns) != (final.st_ino, total, final.st_mtime_ns, final.st_ctime_ns):
raise ValueError("WebUI archive changed during metadata inspection")
finally:
os.close(descriptor)
def _inspect_npm_stream(stream: Any, *, name: str, version: str) -> None:
found = False
total = 0
with tarfile.open(fileobj=stream, mode="r|gz") as archive:
for index, member in enumerate(archive):
total += member.size
parts = PurePosixPath(member.name).parts
if index >= 10000 or total > 1024 * 1024 * 1024 or member.size > 64 * 1024 * 1024:
raise ValueError("WebUI archive exceeds its inspection bound")
if not parts or parts[0] != "package" or ".." in parts or not (member.isfile() or member.isdir()):
raise ValueError("WebUI archive contains an unsafe member")
if member.name == "package/package.json":
if found or not member.isfile() or member.size > 1024 * 1024:
raise ValueError("WebUI archive has duplicate or oversized metadata")
stream = archive.extractfile(member)
if stream is None:
raise ValueError("WebUI archive metadata cannot be read")
metadata = json.loads(stream.read(1024 * 1024 + 1))
if not isinstance(metadata, dict) or (metadata.get("name"), metadata.get("version")) != (name, version):
raise ValueError("WebUI metadata differs from the registry lock")
found = True
if not found:
raise ValueError("WebUI archive metadata is missing")
def _write_private_json(path: Path, payload: object) -> None:
encoded = (json.dumps(payload, indent=2, sort_keys=True) + "\n").encode()
if len(encoded) > 16 * 1024 * 1024:
raise ValueError("candidate JSON exceeds its byte bound")
descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), 0o600)
with os.fdopen(descriptor, "wb") as handle:
handle.write(encoded)
tools/release/govoplan_release/publisher.py
+20 -1
View File
@@ -27,7 +27,10 @@ from .candidate_artifact import validate_release_channel
from .model import CatalogPublishResult, CatalogPublishStep
from .module_directory import module_directory_payloads
from .selective_catalog import read_bounded_json_source, trusted_keys_from_keyring
from .source_provenance import catalog_source_selection, source_tag_provenance_issues
from .source_provenance import (
catalog_source_selection, registered_source_origin_issues,
source_tag_provenance_issues,
)
from .version_alignment import candidate_catalog_version_issues
from .workspace import (
DEFAULT_WORKSPACE_ROOT,
@@ -176,6 +179,22 @@ def publish_catalog_candidate(
for issue in version_issues
)
source_selection = catalog_source_selection(candidate_payload)
entries = [candidate_payload.get("core_release")]
if isinstance(candidate_payload.get("modules"), list):
entries.extend(candidate_payload["modules"])
if any(
isinstance(entry, dict)
and isinstance(entry.get("python_ref"), str)
and " @ https://" in entry["python_ref"]
for entry in entries
):
blockers.extend(
f"registered source origin: {issue.describe()}"
for issue in registered_source_origin_issues(
repo_versions=source_selection.all_versions,
workspace=workspace, remote=source_remote,
)
)
blockers.extend(
f"source provenance: {issue.describe()}"
for issue in source_selection.issues
tools/release/govoplan_release/registry_reference.py
+178
View File
@@ -0,0 +1,178 @@
"""Strict source identities for immutable, registry-backed catalog entries."""
from __future__ import annotations
import base64
from dataclasses import dataclass
import json
import re
from urllib.parse import unquote, urlsplit
_SHA256 = re.compile(r"[0-9a-f]{64}\Z")
_COMMIT = re.compile(r"[0-9a-f]{40}(?:[0-9a-f]{24})?\Z")
_REPO = re.compile(r"govoplan-[a-z0-9-]+\Z")
_VERSION = re.compile(r"\d+\.\d+\.\d+(?:[-+][0-9A-Za-z.-]+)?\Z")
_PYTHON = re.compile(
r"(?P<name>govoplan-[a-z0-9-]+)(?:\[[a-z0-9_,.-]+\])? @ "
r"(?P<url>https://\S+)#sha256=(?P<digest>[0-9a-f]{64})\Z"
)
@dataclass(frozen=True, slots=True)
class RegistrySource:
repository: str
version: str
commit: str
tag_object: str
def registry_artifact_conflicts(entries: list[object]) -> tuple[str, ...]:
"""Allow repeated module projections only when package artifacts agree."""
observed: dict[tuple[str, str], str] = {}
issues = []
for entry in entries:
if not isinstance(entry, dict):
continue
try:
source = registry_entry_source(entry)
except ValueError as exc:
issues.append(str(exc))
continue
if source is None:
continue
for kind, identity in entry["artifact_integrity"].items():
if kind not in {"python", "webui"}:
continue
key = (source.repository, kind)
encoded = json.dumps(identity, sort_keys=True, separators=(",", ":"))
if observed.setdefault(key, encoded) != encoded:
issues.append(f"conflicting {kind} registry artifacts for {source.repository}")
return tuple(issues)
def registry_entry_source(entry: dict[str, object]) -> RegistrySource | None:
"""Admit registry refs only with a complete matching artifact/source binding.
Git-backed catalogs keep their existing validation path. An HTTPS Python
requirement cannot masquerade as a source-only/non-Python entry when its
registry provenance is missing or inconsistent.
"""
ref = entry.get("python_ref")
if not isinstance(ref, str) or " @ https://" not in ref:
return None
match = _PYTHON.fullmatch(ref)
source = entry.get("source")
version = entry.get("version")
package = entry.get("python_package")
integrity = entry.get("artifact_integrity")
if (
match is None
or not isinstance(source, dict)
or not isinstance(integrity, dict)
or not isinstance(version, str)
or _VERSION.fullmatch(version) is None
or package != match.group("name")
):
raise ValueError("registry entry has no complete package/source identity")
repo = source.get("repository")
commit = source.get("commit")
tag_object = source.get("tag_object_sha")
if (
not isinstance(repo, str)
or _REPO.fullmatch(repo) is None
or repo != package
or source.get("tag") != f"v{version}"
or not isinstance(commit, str)
or _COMMIT.fullmatch(commit) is None
or not isinstance(tag_object, str)
or _COMMIT.fullmatch(tag_object) is None
):
raise ValueError("registry entry has invalid immutable tag provenance")
python = _artifact(
integrity.get("python"), ref=ref, package=package, version=version,
commit=commit,
)
if python["url"] != match.group("url") or python["sha256"] != match.group("digest"):
raise ValueError("registry Python ref differs from its artifact identity")
webui_package = entry.get("webui_package")
webui_ref = entry.get("webui_ref")
if bool(webui_package) != bool(webui_ref):
raise ValueError("registry WebUI package and ref must be declared together")
if webui_package:
if not isinstance(webui_package, str) or re.fullmatch(
r"@govoplan/[a-z0-9-]+-webui", webui_package
) is None or not isinstance(webui_ref, str):
raise ValueError("registry WebUI package identity is malformed")
if webui_package != f"@govoplan/{repo.removeprefix('govoplan-')}-webui":
raise ValueError("registry WebUI package belongs to another source repository")
webui = _artifact(
integrity.get("webui"), ref=webui_ref, package=webui_package,
version=version, commit=commit,
)
if webui_ref != webui["url"]:
raise ValueError("registry WebUI ref differs from its artifact identity")
sri = webui.get("integrity")
try:
valid_sri = isinstance(sri, str) and sri.startswith("sha512-") and len(
base64.b64decode(sri[7:], validate=True)
) == 64
except ValueError:
valid_sri = False
if not valid_sri:
raise ValueError("registry WebUI artifact needs a SHA-512 integrity identity")
elif "webui" in integrity:
raise ValueError("registry entry carries an unexpected WebUI artifact")
return RegistrySource(repo, version, commit, tag_object)
def _artifact(
value: object, *, ref: str, package: str, version: str, commit: str,
) -> dict[str, object]:
if not isinstance(value, dict):
raise ValueError("registry entry is missing artifact integrity")
url = value.get("url")
filename = value.get("filename")
digest = value.get("sha256")
size = value.get("size")
parsed = urlsplit(url) if isinstance(url, str) else None
url_filename = unquote(parsed.path.rsplit("/", 1)[-1]) if parsed else ""
decoded_parts = unquote(parsed.path).split("/") if parsed else []
expected_filenames = {url_filename}
expected_path = [package, version, url_filename]
if package.startswith("@govoplan/"):
# npm pack includes the scope in its local filename; the registry's
# immutable download URL uses the unscoped package basename.
expected_filenames = {
f"govoplan-{package.split('/', 1)[1]}-{version}.tgz",
} if url_filename == f"{package.split('/', 1)[1]}-{version}.tgz" else set()
expected_path = [*package.split("/"), "-", version, url_filename]
if (
parsed is None
or parsed.scheme != "https"
or not parsed.netloc
or parsed.username is not None
or parsed.password is not None
or parsed.fragment
or parsed.query
or any(part in {".", ".."} for part in unquote(parsed.path).split("/"))
or "%" in unquote(parsed.path)
or "\\" in unquote(parsed.path)
or any(ord(character) < 32 for character in url)
or decoded_parts[-len(expected_path):] != expected_path
or not isinstance(filename, str)
or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._+-]{0,254}", filename) is None
or filename not in expected_filenames
or not isinstance(digest, str)
or _SHA256.fullmatch(digest) is None
or not isinstance(size, int)
or isinstance(size, bool)
or not 0 < size <= 512 * 1024 * 1024
or value.get("ref") != ref
or value.get("registry_identity") != f"{package}@{version}"
or value.get("git_ref") != f"v{version}"
or value.get("source_commit") != commit
):
raise ValueError("registry artifact ref, version, bytes, or source binding is inconsistent")
return value
tools/release/govoplan_release/source_provenance.py
+48 -2
View File
@@ -19,10 +19,33 @@ from .git_state import (
scoped_git_command,
)
from .repository_tag import RemoteTagResult, ref_commit, remote_tag_commit
from .registry_reference import registry_entry_source
from .version_alignment import repository_version_issues
from .workspace import load_repository_specs, resolve_repo_path
def registered_source_origin_issues(
*, repo_versions: dict[str, str], workspace: Path, remote: str,
) -> tuple[SourceTagProvenanceIssue, ...]:
"""Bind registry candidate attestations to registered source endpoints."""
specs = {spec.name: spec for spec in load_repository_specs(include_website=False)}
issues = []
for repo, version in sorted(repo_versions.items()):
spec = specs.get(repo)
if spec is None:
issues.append(SourceTagProvenanceIssue(repo, f"v{version}", "source repository is not registered"))
continue
path = resolve_repo_path(spec, workspace)
if path.absolute() != path.resolve() or not path.resolve().is_relative_to(workspace.resolve()):
issues.append(SourceTagProvenanceIssue(repo, f"v{version}", "source checkout leaves the private workspace or traverses a symlink"))
continue
fetch = git_text(path, "remote", "get-url", "--all", remote).splitlines()
push = git_text(path, "remote", "get-url", "--push", "--all", remote).splitlines()
if fetch != [spec.remote] or push != [spec.remote]:
issues.append(SourceTagProvenanceIssue(repo, f"v{version}", "source remote does not exactly match the registered origin"))
return tuple(issues)
_CATALOG_PYTHON_REF = re.compile(
r"/(?P<repo>govoplan-[a-z0-9-]+)\.git@v(?P<version>[^\s;]+)$"
)
@@ -61,6 +84,8 @@ def catalog_source_selection(payload: object) -> CatalogSourceSelection:
)
versions: dict[str, str] = {}
registry_commits: dict[str, str] = {}
registry_tag_objects: dict[str, str] = {}
issues: list[SourceTagProvenanceIssue] = []
entries: list[tuple[str, object]] = [("core_release", payload.get("core_release"))]
modules = payload.get("modules")
@@ -70,6 +95,21 @@ def catalog_source_selection(payload: object) -> CatalogSourceSelection:
for source, raw_entry in entries:
if not isinstance(raw_entry, dict):
continue
try:
registry_source = registry_entry_source(raw_entry)
except ValueError as exc:
issues.append(SourceTagProvenanceIssue(source, "", str(exc)))
continue
if registry_source is not None:
repo, version = registry_source.repository, registry_source.version
previous = versions.setdefault(repo, version)
previous_commit = registry_commits.setdefault(repo, registry_source.commit)
previous_object = registry_tag_objects.setdefault(repo, registry_source.tag_object)
if (previous, previous_commit, previous_object) != (
version, registry_source.commit, registry_source.tag_object,
):
issues.append(SourceTagProvenanceIssue(repo, f"v{version}", "registry entries have conflicting immutable source identities"))
continue
python_ref = raw_entry.get("python_ref")
match = _CATALOG_PYTHON_REF.search(python_ref) if isinstance(python_ref, str) else None
if match is None:
@@ -89,8 +129,8 @@ def catalog_source_selection(payload: object) -> CatalogSourceSelection:
release = payload.get("release")
selected_units = release.get("selected_units") if isinstance(release, dict) else None
selected: dict[str, str] = {}
selected_commits: dict[str, str] = {}
selected_tag_objects: dict[str, str] = {}
selected_commits: dict[str, str] = dict(registry_commits)
selected_tag_objects: dict[str, str] = dict(registry_tag_objects)
if not isinstance(selected_units, list) or not selected_units:
issues.append(
SourceTagProvenanceIssue(
@@ -106,16 +146,22 @@ def catalog_source_selection(payload: object) -> CatalogSourceSelection:
repo = unit.get("repo")
version = unit.get("version")
if isinstance(repo, str) and isinstance(version, str):
if repo in selected:
issues.append(SourceTagProvenanceIssue(repo, f"v{version}", "duplicate selected repository"))
selected[repo] = version.removeprefix("v")
commit = unit.get("commit_sha")
tag_object = unit.get("tag_object_sha")
if isinstance(commit, str) and re.fullmatch(r"[0-9a-fA-F]{40}|[0-9a-fA-F]{64}", commit):
if repo in registry_commits and registry_commits[repo] != commit.lower():
issues.append(SourceTagProvenanceIssue(repo, f"v{version}", "selected commit differs from registry artifact source"))
selected_commits[repo] = commit.lower()
else:
issues.append(
SourceTagProvenanceIssue(repo, f"v{version.removeprefix('v')}", "selected unit has no valid commit_sha provenance")
)
if isinstance(tag_object, str) and re.fullmatch(r"[0-9a-fA-F]{40}|[0-9a-fA-F]{64}", tag_object):
if repo in registry_tag_objects and registry_tag_objects[repo] != tag_object.lower():
issues.append(SourceTagProvenanceIssue(repo, f"v{version}", "selected tag object differs from registry artifact source"))
selected_tag_objects[repo] = tag_object.lower()
else:
issues.append(
tools/release/govoplan_release/version_alignment.py
+15
View File
@@ -10,6 +10,7 @@ import subprocess
import tomllib
from .git_state import collect_versions, sanitized_git_environment
from .registry_reference import registry_artifact_conflicts, registry_entry_source
from .workspace import load_repository_specs, resolve_repo_path
@@ -407,6 +408,11 @@ def candidate_catalog_version_issues(payload: object) -> tuple[VersionAlignmentI
)
release = payload.get("release")
registry_entries = [core_release, *(modules if isinstance(modules, list) else [])]
issues.extend(
_catalog_shape_issue("artifact_integrity", issue)
for issue in registry_artifact_conflicts(registry_entries)
)
if isinstance(release, dict):
issues.extend(_catalog_release_issues(release, represented=represented))
return tuple(issues)
@@ -419,6 +425,15 @@ def _catalog_entry_issues(
represented: dict[str, str],
) -> list[VersionAlignmentIssue]:
issues: list[VersionAlignmentIssue] = []
try:
registry_source = registry_entry_source(entry)
except ValueError as exc:
return [_catalog_shape_issue(source, str(exc))]
if registry_source is not None:
previous = represented.setdefault(registry_source.repository, registry_source.version)
if previous != registry_source.version:
issues.append(_catalog_shape_issue(source, "repository appears with conflicting catalog versions"))
return issues
version = entry.get("version")
normalized_version = version.removeprefix("v") if isinstance(version, str) and version else None
if normalized_version is None: