#!/usr/bin/env python3 """Seal, verify, and package-deploy the bounded DualCoach provider-auth candidate. The candidate is intentionally limited to the new production command, its human-driven controller, their tests, the package wheel, and immutable historical evidence. It never contacts a provider, Telegram, or systemd. """ from __future__ import annotations import argparse import ctypes import hashlib import json import os import shutil import stat import subprocess import sys import tempfile import zipfile from pathlib import Path from typing import Any, Iterable import yaml SCHEMA = "dualcoach-provider-auth-candidate-v1" CHECKPOINT_SCHEMA = "dualcoach-provider-auth-candidate-checkpoint-v1" DEPLOYMENT_SCHEMA = "dualcoach-provider-auth-profile-package-deployment-v1" COMMAND = "dualcoach_admin provider-auth check" COMMAND_VERSION = "v2" EVENT_ID = "f4cd2ea29c98b7038bc9490b87ee0db2" CORE_SOURCE_FILES = ( "gateway/platforms/dualcoach_admin.py", "gateway/platforms/dualcoach_tasks21_25_controller.py", "agent/auxiliary_client.py", "agent/codex_runtime.py", "tests/gateway/test_dualcoach_admin_provider_auth.py", "tests/gateway/test_dualcoach_tasks21_25_controller.py", "tests/agent/test_auxiliary_client.py", "pyproject.toml", ) PYTEST_STARTUP_FILES = ( "tests/__init__.py", "tests/conftest.py", "tests/gateway/__init__.py", "tests/gateway/conftest.py", ) GATE18_TEST_NODE = "tests/gateway/test_nutrition_coaching.py" ADMIN_MEMBER = "gateway/platforms/dualcoach_admin.py" CONTROLLER_MEMBER = "gateway/platforms/dualcoach_tasks21_25_controller.py" CODEX_ADAPTER_MEMBER = "agent/auxiliary_client.py" CODEX_RUNTIME_MEMBER = "agent/codex_runtime.py" WHEEL_SOURCE_MEMBERS = ( ADMIN_MEMBER, CONTROLLER_MEMBER, CODEX_ADAPTER_MEMBER, CODEX_RUNTIME_MEMBER, ) ENTRY_POINTS = ( "dualcoach_admin = gateway.platforms.dualcoach_admin:main", "dualcoach_tasks21_25_controller = gateway.platforms.dualcoach_tasks21_25_controller:main", ) PYPROJECT_ENTRY_POINTS = ( 'dualcoach_admin = "gateway.platforms.dualcoach_admin:main"', 'dualcoach_tasks21_25_controller = "gateway.platforms.dualcoach_tasks21_25_controller:main"', ) PROFILE_BASELINE_FILES = ( "config.yaml", "auth.json", "customers/registry.json", "gateway_state.json", "data/owner-actions/draft-deliveries.json", "data/scheduled-deliveries.jsonl", "data/onboarding/telegram-publication-outbox-v1/ledger.json", ) TERMINAL_EVIDENCE = ( ".omo/evidence/dualcoach-task-24-evidence-index.json", ".omo/evidence/dualcoach-task-25-evidence-index.json", ".omo/evidence/task24-terminal-noop-matrix.json", ".omo/evidence/task26/task24-terminal-replay-report.json", ".omo/evidence/task26/task24-task25-final-candidate-manifest.json", ".omo/evidence/task26/task26-owner-customer-v1-readiness-v3.redacted.json", ) def canonical(value: object) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8") def sha256(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def sha256_file(path: Path) -> str: return sha256(path.read_bytes()) def mode(path: Path) -> int: return stat.S_IMODE(path.lstat().st_mode) def regular(path: Path, label: str, allowed_modes: set[int] | None = None) -> None: info = path.lstat() if path.is_symlink() or not path.is_file() or info.st_nlink != 1: raise AssertionError(f"{label} is not a regular single-link file") if allowed_modes is not None and mode(path) not in allowed_modes: raise AssertionError(f"{label} has an unsafe mode") def private_directory(path: Path, label: str, allowed_modes: set[int]) -> None: info = path.lstat() if path.is_symlink() or not path.is_dir() or mode(path) not in allowed_modes: raise AssertionError(f"{label} is not a private directory") def source_entry(root: Path, relative: str) -> dict[str, object]: path = root / relative regular(path, relative, {0o644, 0o664, 0o755, 0o775}) return { "path": relative, "mode": f"{mode(path):04o}", "bytes": path.stat().st_size, "sha256": sha256_file(path), } def _ordered_unique(paths: Iterable[str]) -> tuple[str, ...]: return tuple(dict.fromkeys(paths)) def gate18_source_files(repository: Path) -> tuple[str, ...]: gateway_tests = tuple( sorted( path.relative_to(repository).as_posix() for path in (repository / "tests/gateway").rglob("test_*.py") if path.is_file() ) ) if GATE18_TEST_NODE not in gateway_tests: raise AssertionError("Gate 18 executable test is absent from source closure") return _ordered_unique((*PYTEST_STARTUP_FILES, *gateway_tests)) def candidate_source_files(repository: Path) -> tuple[str, ...]: return _ordered_unique((*CORE_SOURCE_FILES, *gate18_source_files(repository))) def sealed_source_entries(repository: Path) -> list[dict[str, object]]: return [source_entry(repository, relative) for relative in candidate_source_files(repository)] def package_tree(root: Path) -> list[dict[str, object]]: private_directory(root, "profile package", {0o500}) entries: list[dict[str, object]] = [] for path in sorted(root.rglob("*"), key=lambda item: item.relative_to(root).as_posix()): relative = path.relative_to(root).as_posix() info = path.lstat() if path.is_symlink() or (path.is_file() and info.st_nlink != 1): raise AssertionError("profile package has link or alias") if path.is_dir(): entries.append({"path": relative, "kind": "directory", "mode": f"{mode(path):04o}"}) elif path.is_file(): entries.append( { "path": relative, "kind": "file", "mode": f"{mode(path):04o}", "bytes": info.st_size, "sha256": sha256_file(path), } ) else: raise AssertionError("profile package contains an unsupported entry") return entries def package_digest(entries: list[dict[str, object]]) -> str: return sha256(canonical(entries)) def status_bytes(repository: Path) -> bytes: result = subprocess.run( ["git", "status", "--porcelain=v1", "-z", "--untracked-files=all"], cwd=repository, capture_output=True, check=False, ) if result.returncode != 0: raise AssertionError("repository status is unavailable") return result.stdout def status_count(value: bytes) -> int: return len([entry for entry in value.split(b"\0") if entry]) def provider_candidate_digest(admin_source: bytes) -> str: return sha256( canonical( { "command": COMMAND, "command_version": COMMAND_VERSION, "source_sha256": sha256(admin_source), } ) ) def wheel_members(wheel: Path) -> tuple[list[dict[str, object]], list[str]]: with zipfile.ZipFile(wheel) as archive: entries: list[dict[str, object]] = [] for name in sorted(name for name in archive.namelist() if not name.endswith("/")): raw = archive.read(name) entries.append({"path": name, "bytes": len(raw), "sha256": sha256(raw)}) entry_points = archive.read("hermes_agent-0.17.0.dist-info/entry_points.txt").decode("utf-8").splitlines() if not all(entry in entry_points for entry in ENTRY_POINTS): raise AssertionError("wheel entry points are incomplete") return entries, entry_points def _file_record(path: Path, display: str, allowed_modes: set[int]) -> dict[str, object]: regular(path, display, allowed_modes) return { "path": display, "mode": f"{mode(path):04o}", "bytes": path.stat().st_size, "sha256": sha256_file(path), } def profile_process_absent(profile: Path) -> bool: expected = b"HERMES_HOME=" + os.fsencode(profile) for candidate in Path("/proc").iterdir(): if not candidate.name.isdecimal(): continue try: environment = (candidate / "environ").read_bytes().split(b"\0") except (FileNotFoundError, PermissionError, ProcessLookupError, OSError): continue if expected in environment: return False return True def profile_baseline(profile: Path) -> dict[str, object]: records: list[dict[str, object]] = [] for relative in PROFILE_BASELINE_FILES: path = profile / relative if path.exists(): records.append(_file_record(path, relative, {0o600})) else: records.append({"path": relative, "absent": True}) config = yaml.safe_load((profile / "config.yaml").read_text(encoding="utf-8")) registry = json.loads((profile / "customers/registry.json").read_text(encoding="utf-8")) state = json.loads((profile / "gateway_state.json").read_text(encoding="utf-8")) if not isinstance(config, dict) or not isinstance(registry, dict) or not isinstance(state, dict): raise AssertionError("profile baseline is malformed") adaptive = config.get("platforms", {}).get("telegram", {}).get("extra", {}).get("adaptive_nutrition", {}) telegram = state.get("platforms", {}).get("telegram", {}) if ( not isinstance(adaptive, dict) or adaptive.get("delivery_enabled") is not False or registry.get("customers") != [] or state.get("gateway_state") != "stopped" or not isinstance(telegram, dict) or telegram.get("state") != "disconnected" ): raise AssertionError("profile baseline is not clean") ledger_path = profile / "data/onboarding/telegram-publication-outbox-v1/ledger.json" ledger = json.loads(ledger_path.read_text(encoding="utf-8")) if not isinstance(ledger, dict) or ledger.get("records") != []: raise AssertionError("profile delivery ledger is not empty") service = subprocess.run( ["systemctl", "--user", "show", "hermes-gateway-dualcoachtest.service", "--property=ActiveState,SubState", "--value"], capture_output=True, text=True, check=False, ) if service.returncode != 0 or tuple(service.stdout.splitlines()) != ("inactive", "dead"): raise AssertionError("rehearsal service is not inactive") if not profile_process_absent(profile): raise AssertionError("rehearsal profile process is present") return { "files": records, "delivery_enabled": False, "registry_customer_count": 0, "delivery_count": 0, "gateway_state": "stopped", "telegram_state": "disconnected", "service_state": "inactive/dead", "profile_process_count": 0, } def historical_records(evidence_root: Path, workspace_receipt: Path, workspace_index: Path) -> tuple[dict[str, object], list[dict[str, object]]]: receipt = _file_record(workspace_receipt, "historical_workspace_quarantine_receipt", {0o400}) index = _file_record(workspace_index, "historical_workspace_quarantine_index", {0o400}) receipt_payload = json.loads(workspace_receipt.read_text(encoding="utf-8")) index_payload = json.loads(workspace_index.read_text(encoding="utf-8")) if ( receipt_payload.get("event_id") != EVENT_ID or index_payload.get("event_id") != EVENT_ID or receipt_payload.get("candidate_digest") != index_payload.get("candidate_digest") ): raise AssertionError("historical workspace quarantine binding is invalid") prior_identity = evidence_root / "task26/task26-task21-preflight-blocked-v1.redacted.json" identity = _file_record(prior_identity, ".omo/evidence/task26/task26-task21-preflight-blocked-v1.redacted.json", {0o600}) identity_payload = json.loads(prior_identity.read_text(encoding="utf-8")) customer = identity_payload.get("preflight_customer") if not isinstance(customer, dict) or ( customer.get("archive_record_historically_enabled") is not True or customer.get("exact_archive_customer_cardinality") != 1 or customer.get("current_registry_customer_count") != 0 or customer.get("current_registry_enabled_customer_count") != 0 or customer.get("private_dm_shape") is not True or customer.get("distinct_from_owner_and_staff_authorities") is not True ): raise AssertionError("synthetic identity evidence is invalid") terminal = [_file_record(evidence_root.parent.parent / relative, relative, {0o600}) for relative in TERMINAL_EVIDENCE] return {"receipt": receipt, "index": index, "event_id": EVENT_ID}, [identity, *terminal] def write_private(path: Path, value: bytes, mode_value: int = 0o400) -> None: path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) descriptor, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) temporary = Path(temporary_name) try: os.fchmod(descriptor, mode_value) with os.fdopen(descriptor, "wb") as stream: descriptor = -1 stream.write(value) stream.flush() os.fsync(stream.fileno()) os.replace(temporary, path) path.chmod(mode_value) finally: if descriptor >= 0: os.close(descriptor) if temporary.exists(): temporary.unlink() def candidate_core( *, source: list[dict[str, object]], wheel: Path, wheel_entries: list[dict[str, object]], package_entries: list[dict[str, object]], historical_workspace: dict[str, object], historical_terminal: list[dict[str, object]], gate_receipt: dict[str, object], gate_receipt_sha256: str, baseline: dict[str, object], baseline_sha256: str, status: bytes, repository: Path, ) -> dict[str, object]: admin = next(entry for entry in source if entry["path"] == "gateway/platforms/dualcoach_admin.py") controller = next(entry for entry in source if entry["path"] == "gateway/platforms/dualcoach_tasks21_25_controller.py") codex_adapter = next(entry for entry in source if entry["path"] == CODEX_ADAPTER_MEMBER) candidate_digest = provider_candidate_digest((repository / str(admin["path"])).read_bytes()) return { "schema": SCHEMA, "candidate_digest": candidate_digest, "canonicalization": "SHA-256 over UTF-8 canonical JSON with sorted keys and compact separators", "source_entries": source, "source_entries_sha256": sha256(canonical(source)), "provider_auth_module_sha256": admin["sha256"], "controller_module_sha256": controller["sha256"], "codex_adapter_module_sha256": codex_adapter["sha256"], "provider_auth_contract": { "command": COMMAND, "version": COMMAND_VERSION, "adapter_version": "v1", "receipt_schema": "dualcoach-provider-auth-receipt-v2", "codex_active_probe": "explicit_one_shot_nonpersistent_generation_v1", }, "controller_contract": {"schema": "dualcoach-tasks21-25-controller-arm-v1", "tasks": [21, 22, 23, 24, 25], "version": "v2-codex-active-receipt"}, "wheel_entry_points": list(ENTRY_POINTS), "pyproject_entry_points": list(PYPROJECT_ENTRY_POINTS), "wheel": { "path": "artifacts/hermes_agent-0.17.0-py3-none-any.whl", "sha256": sha256_file(wheel), "bytes": wheel.stat().st_size, "members": wheel_entries, "members_sha256": sha256(canonical(wheel_entries)), }, "profile_package": { "source_tree_digest": package_digest(package_entries), "entries": package_entries, "entry_count": len(package_entries), }, "quality_gates": { "path": "artifacts/gate-receipt.json", "sha256": gate_receipt_sha256, "status": gate_receipt.get("status"), }, "historical_workspace_quarantine": historical_workspace, "historical_terminal_and_adverse_indexes": historical_terminal, "synthetic_identity_evidence_sha256": historical_terminal[0]["sha256"], "historical_workspace_receipt_sha256": historical_workspace["receipt"]["sha256"], "historical_workspace_index_sha256": historical_workspace["index"]["sha256"], "repository_status": { "path": "artifacts/repository-status.nul", "sha256": sha256(status), "entry_count": status_count(status), }, "repository_status_sha256": sha256(status), "active_profile_baseline": { "path": "artifacts/profile-baseline.json", "sha256": baseline_sha256, "value": baseline, }, } def root_files(root: Path) -> set[Path]: return {path for path in root.rglob("*") if path.is_file()} def seal(args: argparse.Namespace) -> Path: repository = args.repository.resolve() evidence_root = args.evidence_root.resolve() profile = args.profile.resolve() wheel = args.wheel.resolve() gate_path = args.gate_receipt.resolve() workspace_receipt = args.workspace_receipt.resolve() workspace_index = args.workspace_index.resolve() regular(wheel, "wheel") regular(gate_path, "gate receipt", {0o600}) gates = json.loads(gate_path.read_text(encoding="utf-8")) if not isinstance(gates, dict) or gates.get("status") != "PASS": raise AssertionError("gate receipt is not PASS") source = sealed_source_entries(repository) wheel_entries, _entry_points = wheel_members(wheel) package_entries = package_tree(profile / "workspace/checkin_cli") workspace, terminal = historical_records(evidence_root, workspace_receipt, workspace_index) baseline = profile_baseline(profile) status = status_bytes(repository) baseline_bytes = canonical(baseline) + b"\n" core = candidate_core( source=source, wheel=wheel, wheel_entries=wheel_entries, package_entries=package_entries, historical_workspace=workspace, historical_terminal=terminal, gate_receipt=gates, gate_receipt_sha256=sha256_file(gate_path), baseline=baseline, baseline_sha256=sha256(baseline_bytes), status=status, repository=repository, ) full_digest = sha256(canonical(core)) manifest = {**core, "full_candidate_digest": full_digest} output_root = args.output_parent.resolve() / f"dualcoach-provider-auth-candidate-{full_digest}" if output_root.exists() or output_root.is_symlink(): raise FileExistsError("candidate root already exists") output_root.mkdir(mode=0o700, parents=True) artifacts = output_root / "artifacts" artifacts.mkdir(mode=0o700) shutil.copyfile(wheel, artifacts / "hermes_agent-0.17.0-py3-none-any.whl") (artifacts / "hermes_agent-0.17.0-py3-none-any.whl").chmod(0o400) write_private(artifacts / "gate-receipt.json", gate_path.read_bytes()) write_private(artifacts / "profile-baseline.json", baseline_bytes) write_private(artifacts / "repository-status.nul", status) write_private(output_root / "candidate-manifest.json", canonical(manifest) + b"\n") checkpoint = { "schema": CHECKPOINT_SCHEMA, "full_candidate_digest": full_digest, "candidate_digest": manifest["candidate_digest"], "candidate_manifest_sha256": sha256_file(output_root / "candidate-manifest.json"), "wheel_sha256": manifest["wheel"]["sha256"], "repository_status_sha256": manifest["repository_status_sha256"], "status": "SEALED", } write_private(output_root / "candidate-checkpoint.json", canonical(checkpoint) + b"\n") artifacts.chmod(0o500) output_root.chmod(0o500) verify(output_root, repository=repository, evidence_root=evidence_root, profile=profile, workspace_receipt=workspace_receipt, workspace_index=workspace_index) return output_root def verify( root: Path, *, repository: Path, evidence_root: Path, profile: Path, workspace_receipt: Path, workspace_index: Path, ) -> dict[str, object]: private_directory(root, "candidate root", {0o500}) artifacts = root / "artifacts" private_directory(artifacts, "candidate artifacts", {0o500}) expected = { root / "candidate-manifest.json", root / "candidate-checkpoint.json", artifacts / "hermes_agent-0.17.0-py3-none-any.whl", artifacts / "gate-receipt.json", artifacts / "profile-baseline.json", artifacts / "repository-status.nul", } if root_files(root) != expected: raise AssertionError("candidate inventory differs") for path in expected: regular(path, "candidate artifact", {0o400}) manifest_path = root / "candidate-manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) if not isinstance(manifest, dict) or manifest.get("schema") != SCHEMA: raise AssertionError("candidate manifest is invalid") full_digest = manifest.pop("full_candidate_digest", None) if not isinstance(full_digest, str) or sha256(canonical(manifest)) != full_digest or root.name != f"dualcoach-provider-auth-candidate-{full_digest}": raise AssertionError("candidate digest differs") manifest["full_candidate_digest"] = full_digest checkpoint = json.loads((root / "candidate-checkpoint.json").read_text(encoding="utf-8")) if checkpoint != { "schema": CHECKPOINT_SCHEMA, "full_candidate_digest": full_digest, "candidate_digest": manifest["candidate_digest"], "candidate_manifest_sha256": sha256_file(manifest_path), "wheel_sha256": manifest["wheel"]["sha256"], "repository_status_sha256": manifest["repository_status_sha256"], "status": "SEALED", }: raise AssertionError("candidate checkpoint differs") actual_source = [source_entry(repository, str(entry["path"])) for entry in manifest["source_entries"]] if actual_source != manifest["source_entries"]: raise AssertionError("candidate source differs") admin_path = repository / ADMIN_MEMBER if manifest["candidate_digest"] != provider_candidate_digest(admin_path.read_bytes()): raise AssertionError("provider candidate digest differs") pyproject = (repository / "pyproject.toml").read_text(encoding="utf-8") if not all(entry in pyproject for entry in PYPROJECT_ENTRY_POINTS): raise AssertionError("source entry points differ") wheel_path = root / str(manifest["wheel"]["path"]) if sha256_file(wheel_path) != manifest["wheel"]["sha256"]: raise AssertionError("candidate wheel differs") members, entry_points = wheel_members(wheel_path) if members != manifest["wheel"]["members"] or sha256(canonical(members)) != manifest["wheel"]["members_sha256"] or not all(item in entry_points for item in ENTRY_POINTS): raise AssertionError("candidate wheel members differ") wheel_member_hashes = {str(entry["path"]): entry["sha256"] for entry in members} source_hashes = {str(entry["path"]): entry["sha256"] for entry in manifest["source_entries"]} if any(wheel_member_hashes.get(path) != source_hashes.get(path) for path in WHEEL_SOURCE_MEMBERS): raise AssertionError("candidate wheel source differs") if json.loads((root / "artifacts/gate-receipt.json").read_text(encoding="utf-8")).get("status") != "PASS" or sha256_file(root / "artifacts/gate-receipt.json") != manifest["quality_gates"]["sha256"]: raise AssertionError("candidate gate receipt differs") if (root / "artifacts/repository-status.nul").read_bytes() != status_bytes(repository): raise AssertionError("repository status drift") baseline = profile_baseline(profile) baseline_bytes = canonical(baseline) + b"\n" if baseline != manifest["active_profile_baseline"]["value"] or sha256(baseline_bytes) != manifest["active_profile_baseline"]["sha256"] or (root / "artifacts/profile-baseline.json").read_bytes() != baseline_bytes: raise AssertionError("profile baseline drift") workspace, terminal = historical_records(evidence_root, workspace_receipt, workspace_index) if workspace != manifest["historical_workspace_quarantine"] or terminal != manifest["historical_terminal_and_adverse_indexes"]: raise AssertionError("historical evidence drift") return { "schema": "dualcoach-provider-auth-candidate-verification-v1", "status": "PASS", "full_candidate_digest": full_digest, "candidate_digest": manifest["candidate_digest"], "candidate_manifest_sha256": sha256_file(manifest_path), "wheel_sha256": sha256_file(wheel_path), } def _copy_tree_exact(source: Path, destination: Path) -> None: shutil.copytree(source, destination, copy_function=shutil.copy2) for path in [destination, *destination.rglob("*")]: source_path = source / path.relative_to(destination) path.chmod(mode(source_path)) def _rename_exchange(left: Path, right: Path) -> None: result = ctypes.CDLL(None, use_errno=True).syscall( 316, -100, os.fsencode(left), -100, os.fsencode(right), 2, ) if result != 0: error = ctypes.get_errno() raise OSError(error, os.strerror(error)) def _remove_tree(path: Path) -> None: for item in sorted(path.rglob("*"), key=lambda value: len(value.parts), reverse=True): if item.is_dir() and not item.is_symlink(): item.chmod(0o700) elif item.is_file() and not item.is_symlink(): item.chmod(0o600) path.chmod(0o700) shutil.rmtree(path) def deployment_receipt_path() -> Path: return Path.home() / ".local/state/dualcoach-provider-auth-deployment/latest-deployment-receipt.redacted.json" def deploy(args: argparse.Namespace) -> dict[str, object]: root = args.candidate.resolve() verified = verify(root, repository=args.repository.resolve(), evidence_root=args.evidence_root.resolve(), profile=args.profile.resolve(), workspace_receipt=args.workspace_receipt.resolve(), workspace_index=args.workspace_index.resolve()) manifest = json.loads((root / "candidate-manifest.json").read_text(encoding="utf-8")) profile = args.profile.resolve() workspace = profile / "workspace" active = workspace / "checkin_cli" source_entries = manifest["profile_package"]["entries"] if package_tree(active) != source_entries: raise AssertionError("profile package drift before deployment") state = Path.home() / ".local/state/dualcoach-provider-auth-deployment" backup = state / "backups" / str(manifest["full_candidate_digest"]) / "checkin_cli" backup.parent.mkdir(mode=0o700, parents=True, exist_ok=True) if backup.exists() or backup.is_symlink(): raise FileExistsError("deployment backup already exists") _copy_tree_exact(active, backup) backup.chmod(0o500) staging = workspace / f".checkin_cli.deploy-{manifest['full_candidate_digest']}" if staging.exists() or staging.is_symlink(): raise FileExistsError("deployment staging already exists") _copy_tree_exact(active, staging) if package_tree(staging) != source_entries: raise AssertionError("deployment staging differs") _rename_exchange(active, staging) try: deployed_entries = package_tree(active) if deployed_entries != source_entries: raise AssertionError("deployed profile package differs") _remove_tree(staging) except BaseException: if staging.exists() and active.exists(): _rename_exchange(active, staging) raise receipt = { "schema": DEPLOYMENT_SCHEMA, "status": "PASS", "full_candidate_digest": manifest["full_candidate_digest"], "candidate_digest": manifest["candidate_digest"], "candidate_manifest_sha256": sha256_file(root / "candidate-manifest.json"), "wheel_sha256": manifest["wheel"]["sha256"], "service_state": "inactive/dead", "delivery_enabled": False, "registry_customer_count": 0, "delivery_count": 0, "source_package_tree_digest": manifest["profile_package"]["source_tree_digest"], "deployed_package_tree_digest": package_digest(deployed_entries), "source_package_entry_count": len(source_entries), "deployed_package_entry_count": len(deployed_entries), "backup_path": str(backup), "wheel_source_equality": { "provider_auth_module_sha256": manifest["provider_auth_module_sha256"], "controller_module_sha256": manifest["controller_module_sha256"], "codex_adapter_module_sha256": manifest["codex_adapter_module_sha256"], "wheel_sha256": manifest["wheel"]["sha256"], }, } receipt_path = deployment_receipt_path() write_private(receipt_path, canonical(receipt) + b"\n", 0o600) return {**verified, "deployment_receipt": str(receipt_path), "deployed_package_tree_digest": receipt["deployed_package_tree_digest"]} def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() commands = parser.add_subparsers(dest="command", required=True) common = argparse.ArgumentParser(add_help=False) common.add_argument("--repository", type=Path, required=True) common.add_argument("--evidence-root", type=Path, required=True) common.add_argument("--profile", type=Path, required=True) common.add_argument("--workspace-receipt", type=Path, required=True) common.add_argument("--workspace-index", type=Path, required=True) seal_parser = commands.add_parser("seal", parents=[common]) seal_parser.add_argument("--wheel", type=Path, required=True) seal_parser.add_argument("--gate-receipt", type=Path, required=True) seal_parser.add_argument("--output-parent", type=Path, required=True) verify_parser = commands.add_parser("verify", parents=[common]) verify_parser.add_argument("--candidate", type=Path, required=True) deploy_parser = commands.add_parser("deploy", parents=[common]) deploy_parser.add_argument("--candidate", type=Path, required=True) return parser.parse_args() def main() -> int: global args args = parse_args() try: if args.command == "seal": root = seal(args) result = {**verify(root, repository=args.repository.resolve(), evidence_root=args.evidence_root.resolve(), profile=args.profile.resolve(), workspace_receipt=args.workspace_receipt.resolve(), workspace_index=args.workspace_index.resolve()), "candidate_root": str(root)} elif args.command == "verify": result = verify(args.candidate.resolve(), repository=args.repository.resolve(), evidence_root=args.evidence_root.resolve(), profile=args.profile.resolve(), workspace_receipt=args.workspace_receipt.resolve(), workspace_index=args.workspace_index.resolve()) else: result = deploy(args) except (AssertionError, FileExistsError, OSError, KeyError, TypeError, ValueError, zipfile.BadZipFile) as error: print(json.dumps({"status": "FAIL", "reason": type(error).__name__}, sort_keys=True)) return 1 print(json.dumps(result, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())