From 35c346a1faf8aecf344382d6e82019521dc0d169 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Wed, 22 Jul 2026 20:41:36 +0200 Subject: [PATCH] Add private release candidate handles --- tests/test_release_candidate_artifact.py | 153 +++++++++++ .../govoplan_release/candidate_artifact.py | 254 ++++++++++++++++++ 2 files changed, 407 insertions(+) create mode 100644 tests/test_release_candidate_artifact.py create mode 100644 tools/release/govoplan_release/candidate_artifact.py diff --git a/tests/test_release_candidate_artifact.py b/tests/test_release_candidate_artifact.py new file mode 100644 index 0000000..336c08c --- /dev/null +++ b/tests/test_release_candidate_artifact.py @@ -0,0 +1,153 @@ +from __future__ import annotations + +import json +from pathlib import Path +import sys +import tempfile +import unittest + + +META_ROOT = Path(__file__).resolve().parents[1] +RELEASE_TOOLS_ROOT = META_ROOT / "tools" / "release" +if str(RELEASE_TOOLS_ROOT) not in sys.path: + sys.path.insert(0, str(RELEASE_TOOLS_ROOT)) + +from govoplan_release.candidate_artifact import ( # noqa: E402 + CandidateArtifactError, + candidate_output_path, + issue_candidate_id, + issue_candidate_receipt, + verify_candidate_receipt, + validate_release_channel, +) + + +class CandidateArtifactTests(unittest.TestCase): + def test_channel_is_one_canonical_basename(self) -> None: + self.assertEqual("stable_1", validate_release_channel("stable_1")) + for value in ("../stable", "/stable", ".", "..", "Stable", "stable/name"): + with self.subTest(value=value), self.assertRaises(CandidateArtifactError): + validate_release_channel(value) + + def test_handle_is_deterministic_and_can_precede_output(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) / "not-created" / "release-candidates" + candidate_id = issue_candidate_id("rr-123", "attempt-456") + + first = candidate_output_path(root, candidate_id) + second = candidate_output_path(root, candidate_id) + + self.assertEqual(first, second) + self.assertEqual(candidate_id, first.name) + self.assertRegex(candidate_id, r"^candidate-[0-9a-f]{32}$") + + def test_receipt_rejects_catalog_drift(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) / "release-candidates" + candidate_id = issue_candidate_id("run", "attempt") + catalog = self._write_candidate(root / candidate_id) + receipt = issue_candidate_receipt( + root=root, candidate_id=candidate_id, channel="stable" + ) + + resolved = verify_candidate_receipt( + root=root, + candidate_id=receipt.candidate_id, + catalog_sha256=receipt.catalog_sha256, + channel="stable", + ) + catalog.write_text( + json.dumps({"channel": "stable", "sequence": 2, "signatures": [{}]}), + encoding="utf-8", + ) + + self.assertEqual(root / candidate_id, resolved) + with self.assertRaisesRegex(CandidateArtifactError, "no longer matches"): + verify_candidate_receipt( + root=root, + candidate_id=receipt.candidate_id, + catalog_sha256=receipt.catalog_sha256, + channel="stable", + ) + + def test_unissued_ids_traversal_and_symlinks_fail_closed(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) / "release-candidates" + root.mkdir(mode=0o700) + candidate_id = issue_candidate_id("run", "attempt") + outside = Path(temp_dir) / "outside" + outside.mkdir() + (root / candidate_id).symlink_to(outside, target_is_directory=True) + + with self.assertRaises(CandidateArtifactError): + candidate_output_path(root, "../candidate-" + "0" * 32) + with self.assertRaisesRegex(CandidateArtifactError, "real directory"): + candidate_output_path(root, candidate_id) + + def test_catalog_and_channel_components_must_not_be_symlinks(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) / "release-candidates" + candidate_id = issue_candidate_id("run", "attempt") + candidate = root / candidate_id + target = Path(temp_dir) / "target" + target.mkdir() + (target / "stable.json").write_text( + json.dumps({"signatures": [{}]}), encoding="utf-8" + ) + root.mkdir(mode=0o700) + candidate.mkdir(mode=0o700) + (candidate / "channels").symlink_to(target, target_is_directory=True) + + with self.assertRaisesRegex(CandidateArtifactError, "real directory"): + issue_candidate_receipt( + root=root, candidate_id=candidate_id, channel="stable" + ) + + def test_shared_candidate_roots_and_catalog_files_fail_closed(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) / "release-candidates" + root.mkdir(mode=0o755) + candidate_id = issue_candidate_id("run", "attempt") + with self.assertRaisesRegex(CandidateArtifactError, "another user"): + candidate_output_path(root, candidate_id) + + root.chmod(0o700) + catalog = self._write_candidate(root / candidate_id) + catalog.chmod(0o640) + with self.assertRaisesRegex(CandidateArtifactError, "another user"): + issue_candidate_receipt( + root=root, candidate_id=candidate_id, channel="stable" + ) + + def test_receipt_rejects_catalog_with_inconsistent_channel(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) / "release-candidates" + candidate_id = issue_candidate_id("run", "attempt") + catalog = self._write_candidate(root / candidate_id) + catalog.write_text( + json.dumps({"channel": "next", "signatures": [{}]}), + encoding="utf-8", + ) + + with self.assertRaisesRegex(CandidateArtifactError, "does not match"): + issue_candidate_receipt( + root=root, candidate_id=candidate_id, channel="stable" + ) + + @staticmethod + def _write_candidate(candidate: Path) -> Path: + candidate.parent.mkdir(mode=0o700, exist_ok=True) + candidate.mkdir(mode=0o700) + channels = candidate / "channels" + channels.mkdir(mode=0o700) + catalog = channels / "stable.json" + catalog.write_text( + json.dumps({"channel": "stable", "sequence": 1, "signatures": [{}]}), + encoding="utf-8", + ) + catalog.chmod(0o600) + return catalog + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/release/govoplan_release/candidate_artifact.py b/tools/release/govoplan_release/candidate_artifact.py new file mode 100644 index 0000000..8eb3f3b --- /dev/null +++ b/tools/release/govoplan_release/candidate_artifact.py @@ -0,0 +1,254 @@ +"""Opaque, workspace-scoped handles for generated catalog candidates.""" + +from __future__ import annotations + +from dataclasses import dataclass +import hashlib +import hmac +import json +import os +from pathlib import Path +import re +import stat +from typing import Iterable + +from .catalog import canonical_hash + + +MAX_CATALOG_BYTES = 16 * 1024 * 1024 +_CANDIDATE_ID = re.compile(r"^candidate-[0-9a-f]{32}$") +_CHANNEL = re.compile(r"^[a-z][a-z0-9_-]{0,63}$") + + +class CandidateArtifactError(ValueError): + """A candidate handle or the artifact behind it is unsafe or inconsistent.""" + + +@dataclass(frozen=True, slots=True) +class CandidateArtifactReceipt: + candidate_id: str + catalog_sha256: str + + +def issue_candidate_id(*seed_parts: str) -> str: + """Derive an opaque, deterministic candidate basename before output exists.""" + + if not seed_parts or any( + not isinstance(part, str) or not part or len(part) > 512 + for part in seed_parts + ): + raise CandidateArtifactError("candidate ID seeds must be non-empty bounded strings") + encoded = json.dumps( + list(seed_parts), sort_keys=False, separators=(",", ":"), ensure_ascii=True + ).encode("utf-8") + return f"candidate-{hashlib.sha256(encoded).hexdigest()[:32]}" + + +def validate_release_channel(value: str) -> str: + """Return one bounded channel basename or reject path-capable input.""" + + if not isinstance(value, str) or _CHANNEL.fullmatch(value) is None: + raise CandidateArtifactError( + "release channel must be a bounded lowercase identifier" + ) + return value + + +def candidate_output_path(root: Path | str, candidate_id: str) -> Path: + """Resolve a not-yet-created candidate path below a trusted configured root.""" + + checked_id = _checked_candidate_id(candidate_id) + root_path = Path(root).expanduser() + ensure_private_candidate_root(root_path, create=True) + candidate = root_path / checked_id + try: + mode = candidate.lstat().st_mode + except FileNotFoundError: + return candidate + except OSError as exc: + raise CandidateArtifactError("candidate output path cannot be inspected") from exc + if stat.S_ISLNK(mode) or not stat.S_ISDIR(mode): + raise CandidateArtifactError("candidate output path must be a real directory") + _require_private_owner_mode(candidate, directory=True, label="candidate output path") + return candidate + + +def ensure_private_candidate_root( + root: Path | str, *, create: bool = False +) -> Path: + """Admit only an operator-owned candidate root inaccessible to other users.""" + + root_path = Path(root).expanduser() + _reject_symlink_components(root_path, allow_missing=create) + if create: + try: + root_path.mkdir(mode=0o700, parents=True, exist_ok=True) + except OSError as exc: + raise CandidateArtifactError("candidate root cannot be created") from exc + _reject_symlink_components(root_path, allow_missing=False) + _require_private_owner_mode(root_path, directory=True, label="candidate root") + return root_path + + +def harden_private_candidate_tree(path: Path | str) -> Path: + """Seal a newly generated, operator-owned candidate tree before a receipt.""" + + root = Path(path).expanduser() + _reject_symlink_components(root, allow_missing=False) + _require_current_owner(root, label="candidate directory") + try: + os.chmod(root, 0o700, follow_symlinks=False) + for current, directory_names, file_names in os.walk(root, followlinks=False): + current_path = Path(current) + _require_current_owner(current_path, label="candidate directory") + os.chmod(current_path, 0o700, follow_symlinks=False) + for name in (*directory_names, *file_names): + child = current_path / name + mode = child.lstat().st_mode + if stat.S_ISLNK(mode): + raise CandidateArtifactError("candidate tree contains a symlink") + _require_current_owner(child, label="candidate tree member") + if stat.S_ISDIR(mode): + os.chmod(child, 0o700, follow_symlinks=False) + elif stat.S_ISREG(mode): + os.chmod(child, 0o600, follow_symlinks=False) + else: + raise CandidateArtifactError( + "candidate tree contains a non-file member" + ) + except OSError as exc: + raise CandidateArtifactError("candidate tree cannot be sealed") from exc + return root + + +def issue_candidate_receipt( + *, root: Path | str, candidate_id: str, channel: str +) -> CandidateArtifactReceipt: + """Hash the signed catalog at a generated candidate handle.""" + + candidate = _existing_candidate_path(root, candidate_id) + payload = _read_signed_catalog(candidate, channel=channel) + return CandidateArtifactReceipt( + candidate_id=candidate_id, + catalog_sha256=canonical_hash(payload), + ) + + +def verify_candidate_receipt( + *, + root: Path | str, + candidate_id: str, + catalog_sha256: str, + channel: str, +) -> Path: + """Re-resolve and re-hash a persisted receipt before candidate consumption.""" + + if re.fullmatch(r"[0-9a-f]{64}", catalog_sha256) is None: + raise CandidateArtifactError("candidate catalog digest is malformed") + candidate = _existing_candidate_path(root, candidate_id) + payload = _read_signed_catalog(candidate, channel=channel) + if not hmac.compare_digest(canonical_hash(payload), catalog_sha256): + raise CandidateArtifactError("candidate catalog no longer matches its receipt") + return candidate + + +def _checked_candidate_id(candidate_id: str) -> str: + if not isinstance(candidate_id, str) or _CANDIDATE_ID.fullmatch(candidate_id) is None: + raise CandidateArtifactError("candidate ID is not an issued opaque basename") + return candidate_id + + +def _existing_candidate_path(root: Path | str, candidate_id: str) -> Path: + candidate = candidate_output_path(root, candidate_id) + try: + mode = candidate.lstat().st_mode + except OSError as exc: + raise CandidateArtifactError("candidate directory is unavailable") from exc + if stat.S_ISLNK(mode) or not stat.S_ISDIR(mode): + raise CandidateArtifactError("candidate directory must be a real directory") + _reject_symlink_components(candidate, allow_missing=False) + return candidate + + +def _read_signed_catalog(candidate: Path, *, channel: str) -> dict[str, object]: + channel = validate_release_channel(channel) + channels = candidate / "channels" + catalog_path = channels / f"{channel}.json" + _require_real_directory(channels, label="candidate channels directory") + _require_regular_file(catalog_path, label="candidate catalog") + try: + with catalog_path.open("rb") as handle: + encoded = handle.read(MAX_CATALOG_BYTES + 1) + except OSError as exc: + raise CandidateArtifactError("candidate catalog cannot be read") from exc + if len(encoded) > MAX_CATALOG_BYTES: + raise CandidateArtifactError("candidate catalog exceeds its size limit") + try: + payload = json.loads(encoded.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise CandidateArtifactError("candidate catalog is not valid JSON") from exc + if not isinstance(payload, dict): + raise CandidateArtifactError("candidate catalog must be a JSON object") + if payload.get("channel") != channel: + raise CandidateArtifactError( + "candidate catalog channel does not match its requested handle" + ) + signatures = payload.get("signatures") + if not isinstance(signatures, list) or not signatures: + raise CandidateArtifactError("candidate catalog has no signature envelope") + return payload + + +def _reject_symlink_components(path: Path, *, allow_missing: bool) -> None: + absolute = path.absolute() + parts: Iterable[Path] = reversed((absolute, *absolute.parents)) + for component in parts: + try: + mode = component.lstat().st_mode + except FileNotFoundError: + if allow_missing: + continue + raise CandidateArtifactError("candidate path has a missing component") + except OSError as exc: + raise CandidateArtifactError("candidate path cannot be inspected") from exc + if stat.S_ISLNK(mode): + raise CandidateArtifactError("candidate path must not traverse symlinks") + + +def _require_real_directory(path: Path, *, label: str) -> None: + try: + mode = path.lstat().st_mode + except OSError as exc: + raise CandidateArtifactError(f"{label} is unavailable") from exc + if stat.S_ISLNK(mode) or not stat.S_ISDIR(mode): + raise CandidateArtifactError(f"{label} must be a real directory") + _require_private_owner_mode(path, directory=True, label=label) + + +def _require_regular_file(path: Path, *, label: str) -> None: + try: + mode = path.lstat().st_mode + except OSError as exc: + raise CandidateArtifactError(f"{label} is unavailable") from exc + if stat.S_ISLNK(mode) or not stat.S_ISREG(mode): + raise CandidateArtifactError(f"{label} must be a regular file") + _require_private_owner_mode(path, directory=False, label=label) + + +def _require_current_owner(path: Path, *, label: str) -> os.stat_result: + try: + observed = path.lstat() + except OSError as exc: + raise CandidateArtifactError(f"{label} cannot be inspected") from exc + if observed.st_uid != os.geteuid(): + raise CandidateArtifactError(f"{label} is not owned by the current operator") + return observed + + +def _require_private_owner_mode( + path: Path, *, directory: bool, label: str +) -> None: + observed = _require_current_owner(path, label=label) + expected_type = stat.S_ISDIR if directory else stat.S_ISREG + if not expected_type(observed.st_mode) or stat.S_IMODE(observed.st_mode) & 0o077: + raise CandidateArtifactError(f"{label} is accessible to another user")