#!/usr/bin/env python3 """Build and validate the immutable GCP leoclean no-send service package.""" from __future__ import annotations import argparse import hashlib import json import re import shutil import stat import subprocess import sys import tomllib from collections.abc import Mapping from pathlib import Path from typing import Any CONTRACT_ROOT = ( Path(__file__).resolve().parents[1] if Path(__file__).resolve().parent.name == "ops" else Path(__file__).resolve().parent ) if str(CONTRACT_ROOT) not in sys.path: sys.path.insert(0, str(CONTRACT_ROOT)) from scripts import leo_identity_manifest as strict_identity_manifest # noqa: E402 IMAGE_INPUT_SCHEMA = "livingip.leocleanNoSendImageInput.v2" RELEASE_SCHEMA = "livingip.leocleanNoSendRelease.v1" ARTIFACT_MANIFEST_SCHEMA = "livingip.leocleanNoSendArtifactManifest.v1" ARTIFACT_RECEIPT_SCHEMA = "livingip.leocleanNoSendArtifactVerification.v1" IDENTITY_MANIFEST_SCHEMA = "livingip.leoIdentityManifest.v1" IDENTITY_LOCK_SCHEMA = "livingip.leoIdentityProfileLock.v1" IDENTITY_RUNTIME_CONTRACT_SCHEMA = "livingip.leocleanNoSendIdentityRuntimeContract.v1" COMPILED_IDENTITY_SCHEMA = "livingip.leocleanNoSendCompiledIdentity.v1" PERMISSION_RECEIPT_SCHEMA = "livingip.leocleanNoSendPermissionReceipt.v1" PROJECT = "teleo-501523" ZONE = "europe-west6-a" INSTANCE = "teleo-staging-1" SERVICE = "leoclean-gcp-nosend.service" SERVICE_ACCOUNT = "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com" PLATFORM = "linux/amd64" IMAGE_REPOSITORY = "europe-west6-docker.pkg.dev/teleo-501523/teleo/leoclean-nosend-staging" CONTAINER_NAME = "livingip-leoclean-nosend" HERMES_COMMIT = "b2f477a30b3c05d0f383c543af98496ae8a96070" PYTHON_VERSION = "3.11.9" PYYAML_VERSION = "6.0.3" UV_VERSION = "0.9.30" POSTGRES_VERSION = "16.14" MODEL_PROVIDER = "openrouter" DEFAULT_MODEL = "anthropic/claude-sonnet-4-6" PYTHON_IMAGE = ( "docker.io/library/python:3.11.9-slim-bookworm@" "sha256:8fb099199b9f2d70342674bd9dbccd3ed03a258f26bbd1d556822c6dfc60c317" ) UV_IMAGE = "ghcr.io/astral-sh/uv:0.9.30@sha256:538e0b39736e7feae937a65983e49d2ab75e1559d35041f9878b7b7e51de91e4" POSTGRES_IMAGE = ( "docker.io/library/postgres:16.14-bookworm@sha256:92620daddcd947f8d5ab5ba66e848702fe443d87fed30c4cea8e389fd78dfc55" ) CA_SHA256 = "80701e768f0e1f6b9d621aa0b53f6e851daaa276c6d9a8e51a300fbc015539cb" CA_DESTINATION = "/usr/local/libexec/livingip/leoclean-kb/cloudsql-server-ca.pem" RUNTIME_ROOT = "/opt/livingip/leoclean-nosend" PROFILE_ROOT = "/run/leoclean-profile" RUNTIME_UID = 65532 RUNTIME_GID = 65532 HEX_40 = re.compile(r"^[0-9a-f]{40}$") HEX_64 = re.compile(r"^[0-9a-f]{64}$") IMAGE_REFERENCE = re.compile(rf"^{re.escape(IMAGE_REPOSITORY)}@sha256:([0-9a-f]{{64}})$") FORBIDDEN_DESCRIPTOR_MARKERS = ( "77.42.65.182", "gcp-teleo-pgvector-standby-postgres-password", "leoclean-gateway.service", "sa-teleo-prod-vm", "telegram_bot_token", "teleo-prod-1", ) PROPOSAL_FUNCTION = "kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)" IDENTITY_FILES = ("SOUL.md", "identity.json", "identity-lock.json", "identity-manifest.json") RUNTIME_IDENTITY_FILES = ("SOUL.md", "identity-lock.json", "identity-manifest.json") IDENTITY_CONTRACT_SOURCES = ( "scripts/leo_behavior_manifest.py", "scripts/leo_identity_manifest.py", ) PACKAGE_SOURCE_PATHS = ( "docker/leoclean-nosend/Dockerfile", "docker/leoclean-nosend/entrypoint.py", "ops/gcp-teleo-pgvector-standby-server-ca.pem", "ops/gcp_leoclean_nosend_package.py", *IDENTITY_CONTRACT_SOURCES, ) NO_SEND_TOOL_SURFACE = strict_identity_manifest.LEOCLEAN_NO_SEND_TOOL_SURFACE SYNTHETIC_IDENTITY_SOURCE_LABELS = ( "database fingerprint", "constitution source", "database identity source", ) class PackageError(RuntimeError): """The image or service package does not satisfy the staging contract.""" def _require(condition: bool, message: str) -> None: if not condition: raise PackageError(message) def pid_one_command_line_is_exact(cmdline: list[bytes], expected_native: list[bytes]) -> bool: """Accept the pinned native argv and exact x86_64 binfmt wrappers only.""" if not expected_native or expected_native[-1] != b"": return False expected_binfmt = [expected_native[0], *expected_native] accepted = ( expected_native, expected_binfmt, [b"/usr/bin/qemu-x86_64", *expected_binfmt], [b"/usr/bin/qemu-x86_64-static", *expected_binfmt], ) return cmdline in accepted def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def canonical_sha256(value: Any) -> str: encoded = json.dumps(value, ensure_ascii=True, separators=(",", ":"), sort_keys=True).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def load_json(path: Path, label: str) -> dict[str, Any]: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise PackageError(f"cannot read {label}: {type(exc).__name__}") from exc _require(isinstance(value, dict), f"{label} must be a JSON object") return value def write_json(path: Path, value: Mapping[str, Any], mode: int = 0o644) -> None: path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") path.chmod(mode) def _is_sha256(value: Any) -> bool: return isinstance(value, str) and bool(HEX_64.fullmatch(value)) def _is_revision(value: Any) -> bool: return isinstance(value, str) and bool(HEX_40.fullmatch(value)) def _safe_tree(root: Path, label: str) -> None: _require(root.is_dir() and not root.is_symlink(), f"{label} root is missing or unsafe") for path in root.rglob("*"): _require(not path.is_symlink(), f"{label} contains a symlink") _require(path.is_dir() or path.is_file(), f"{label} contains an unsupported entry") def tree_manifest(root: Path, *, excluded: frozenset[str] = frozenset()) -> dict[str, Any]: _safe_tree(root, "tree") files: list[dict[str, Any]] = [] for path in sorted(root.rglob("*"), key=lambda item: item.relative_to(root).as_posix()): relative = path.relative_to(root).as_posix() if relative in excluded or path.is_dir(): continue mode = stat.S_IMODE(path.stat().st_mode) files.append( { "path": relative, "bytes": path.stat().st_size, "mode": f"0o{mode:03o}", "sha256": sha256_file(path), } ) stable = {"files": files, "file_count": len(files), "total_bytes": sum(item["bytes"] for item in files)} return {**stable, "sha256": canonical_sha256(stable)} def identity_bundle_manifest(root: Path) -> dict[str, Any]: _safe_tree(root, "identity bundle") actual = sorted(path.name for path in root.iterdir()) _require(actual == sorted(IDENTITY_FILES), "identity bundle entry set is not exact") files = [] for name in IDENTITY_FILES: path = root / name _require(path.is_file() and not path.is_symlink(), f"identity file is unsafe: {name}") _require(path.stat().st_size <= 1024 * 1024, f"identity file is too large: {name}") _require(not stat.S_IMODE(path.stat().st_mode) & 0o022, f"identity file is group/world writable: {name}") files.append({"path": name, "bytes": path.stat().st_size, "sha256": sha256_file(path)}) stable = {"files": files, "file_count": len(files), "total_bytes": sum(item["bytes"] for item in files)} return {**stable, "sha256": canonical_sha256(stable)} def no_send_permission_policy() -> dict[str, Any]: return { "schema": PERMISSION_RECEIPT_SCHEMA, "messaging_send": { "capability": "absent", "transport_authority": "absent", "gateway_adapters": "none", "send_message_tool": "absent", }, "general_network_egress": { "container_mode": "bridge", "denial_enforced": False, "denial_tested": False, }, "database": { "direct_canonical_writes": "denied", "direct_stage_table_writes": "denied", "proposal_staging": "function_only", "proposal_function": PROPOSAL_FUNCTION, "live_effective_permissions_proven": False, }, "allowed_tools": list(NO_SEND_TOOL_SURFACE["tools"]), } def _runtime_surface(value: Any, label: str) -> dict[str, Any]: _require(isinstance(value, dict), f"{label} is missing") _require(value == NO_SEND_TOOL_SURFACE, f"{label} is not the exact sealed no-send tool surface") return dict(value) def _validate_runtime_probe( receipt: dict[str, Any], *, expected_config_sha256: str | None = None, ) -> tuple[dict[str, Any], dict[str, Any]]: probe = receipt.get("runtime_probe") _require( isinstance(probe, dict) and probe.get("schema") == "livingip.leocleanNoSendRuntimeProbe.v1" and probe.get("status") == "pass", "runtime probe is missing or did not pass", ) probe_config_sha256 = probe.get("config_sha256") _require(_is_sha256(probe_config_sha256), "runtime probe config hash is invalid") _require( probe_config_sha256 == receipt.get("config_sha256"), "runtime probe config differs from the runtime receipt", ) if expected_config_sha256 is not None: _require(_is_sha256(expected_config_sha256), "expected runtime config hash is invalid") _require( probe_config_sha256 == expected_config_sha256, "runtime probe config differs from the packaged config", ) _require(probe.get("gateway_adapter_count") == 0, "runtime probe found a gateway adapter") _require(probe.get("connected_platforms") == [], "runtime probe found a connected platform") _require(probe.get("python_version") == PYTHON_VERSION, "runtime probe Python version drifted") _require(probe.get("model") == DEFAULT_MODEL, "runtime probe model drifted") _require( probe.get("provider_credential_names_present") == ["OPENROUTER_API_KEY"] and probe.get("provider_credential_sentinel_redaction") == "pass", "runtime probe provider-credential boundary drifted", ) effective_policy = probe.get("effective_policy") _require( effective_policy == { "cheap_model_route": None, "memory_enabled": False, "memory_nudge_interval": 0, "skill_creation_nudge_interval": 0, "smart_model_routing": False, "user_profile_enabled": False, }, "runtime probe learning/routing policy drifted", ) surface = _runtime_surface(probe.get("tool_surface"), "runtime probe tool surface") lifecycle = probe.get("lifecycle") _require( isinstance(lifecycle, dict) and lifecycle.get("started") is True and lifecycle.get("stopped") is True, "runtime probe lifecycle was not exercised", ) for phase in ("post_start_registry", "post_discovery_registry"): _require(_runtime_surface(lifecycle.get(phase), f"runtime probe {phase}") == surface, f"{phase} drifted") return effective_policy, surface def _locked_python_runtime(artifact: Path, *, expected_uv_lock_sha256: str) -> dict[str, Any]: uv_lock_path = artifact / "hermes-agent" / "uv.lock" _require(uv_lock_path.is_file() and not uv_lock_path.is_symlink(), "artifact uv.lock is missing or unsafe") _require(sha256_file(uv_lock_path) == expected_uv_lock_sha256, "artifact uv.lock differs from the runtime lock") try: lock = tomllib.loads(uv_lock_path.read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, tomllib.TOMLDecodeError) as exc: raise PackageError(f"cannot read artifact uv.lock: {type(exc).__name__}") from exc packages = lock.get("package") _require(isinstance(packages, list), "artifact uv.lock package set is missing") matches = [ item for item in packages if isinstance(item, dict) and str(item.get("name", "")).casefold() == "pyyaml" ] _require(len(matches) == 1 and matches[0].get("version") == PYYAML_VERSION, "artifact PyYAML lock drifted") distributions = {"PyYAML": PYYAML_VERSION} return { "implementation": "CPython", "python_version": PYTHON_VERSION, "abi_tag": "cpython-311", "runtime_distributions": distributions, "runtime_distributions_sha256": canonical_sha256(distributions), } def _identity_contract_source_bindings(source_root: Path) -> dict[str, str]: bindings: dict[str, str] = {} for relative in IDENTITY_CONTRACT_SOURCES: path = _safe_bound_file(source_root, relative, label=f"identity contract source: {relative}") bindings[relative] = sha256_file(path) return bindings def _identity_runtime_payload(contract: Mapping[str, Any]) -> dict[str, Any]: runtime = contract.get("runtime") or {} return { "schema": contract.get("schema"), "runtime": {key: value for key, value in runtime.items() if key != "identity_runtime_sha256"}, "model": contract.get("model"), "skills": contract.get("skills"), "permissions": contract.get("permissions"), "behavior_inputs": contract.get("behavior_inputs"), } def validate_expected_identity_contract(value: dict[str, Any], *, expected_revision: str) -> None: _require(_is_revision(expected_revision), "expected identity revision is invalid") _require( set(value) == {"schema", "runtime", "model", "skills", "permissions", "behavior_inputs", "contract_sha256"}, "expected identity contract fields are not exact", ) _require(value.get("schema") == IDENTITY_RUNTIME_CONTRACT_SCHEMA, "expected identity contract schema is invalid") stable = {key: item for key, item in value.items() if key != "contract_sha256"} _require(value.get("contract_sha256") == canonical_sha256(stable), "expected identity contract hash drifted") runtime = value.get("runtime") _require( isinstance(runtime, dict) and set(runtime) == { "identity_runtime_sha256", "hermes_git_head", "hermes_source_sha256", "teleo_git_head", "teleo_source_sha256", "python", }, "expected identity runtime fields are not exact", ) _require( runtime.get("identity_runtime_sha256") == canonical_sha256(_identity_runtime_payload(value)), "expected identity runtime hash drifted", ) _require(runtime.get("hermes_git_head") == HERMES_COMMIT, "expected identity Hermes revision drifted") _require(runtime.get("teleo_git_head") == expected_revision, "expected identity Teleo revision drifted") for field in ("identity_runtime_sha256", "hermes_source_sha256", "teleo_source_sha256"): _require(_is_sha256(runtime.get(field)), f"expected identity runtime {field} is invalid") distributions = {"PyYAML": PYYAML_VERSION} _require( runtime.get("python") == { "implementation": "CPython", "python_version": PYTHON_VERSION, "abi_tag": "cpython-311", "runtime_distributions": distributions, "runtime_distributions_sha256": canonical_sha256(distributions), }, "expected identity Python runtime drifted", ) model = value.get("model") _require( isinstance(model, dict) and set(model) == { "provider", "default_model", "smart_routing", "smart_routing_model", "gateway_smart_model_routing", "model_section_sha256", "hosted_weights", "actual_model_required_in_turn_receipt", }, "expected identity model fields are not exact", ) _require( model.get("provider") == MODEL_PROVIDER and model.get("default_model") == DEFAULT_MODEL and model.get("smart_routing") is False and model.get("smart_routing_model") is None and model.get("gateway_smart_model_routing") is False and _is_sha256(model.get("model_section_sha256")) and model.get("hosted_weights") == "external_provider_managed_not_locally_hashable" and model.get("actual_model_required_in_turn_receipt") is True, "expected identity model binding drifted", ) skills = value.get("skills") _require( isinstance(skills, dict) and set(skills) == {"content_sha256", "file_count"} and _is_sha256(skills.get("content_sha256")) and isinstance(skills.get("file_count"), int) and not isinstance(skills.get("file_count"), bool) and skills["file_count"] > 0, "expected identity skills binding is invalid", ) permissions = value.get("permissions") _require( isinstance(permissions, dict) and set(permissions) == { "identity_config_sha256", "gateway_permissions_sha256", "tool_permissions_sha256", "runtime_policy", "authorization_decision_required_in_runtime_receipt", }, "expected identity permission fields are not exact", ) for field in ("identity_config_sha256", "gateway_permissions_sha256", "tool_permissions_sha256"): _require(_is_sha256(permissions.get(field)), f"expected identity permission {field} is invalid") _require( permissions.get("runtime_policy") == no_send_permission_policy() and permissions.get("authorization_decision_required_in_runtime_receipt") is True, "expected identity permission policy drifted", ) behavior_inputs = value.get("behavior_inputs") _require( isinstance(behavior_inputs, dict) and set(behavior_inputs) == { "artifact_sha256", "artifact_receipt_sha256", "config_sha256", "runtime_contract_sha256", "uv_lock_sha256", "identity_contract_sources", "effective_policy", "tool_surface", }, "expected identity behavior inputs are not exact", ) for field in ("artifact_sha256", "artifact_receipt_sha256", "config_sha256", "runtime_contract_sha256", "uv_lock_sha256"): _require(_is_sha256(behavior_inputs.get(field)), f"expected identity behavior input {field} is invalid") source_bindings = behavior_inputs.get("identity_contract_sources") _require( isinstance(source_bindings, dict) and set(source_bindings) == set(IDENTITY_CONTRACT_SOURCES) and all(_is_sha256(item) for item in source_bindings.values()) and runtime.get("teleo_source_sha256") == canonical_sha256(source_bindings), "expected identity contract source binding drifted", ) _require( behavior_inputs.get("tool_surface") == NO_SEND_TOOL_SURFACE, "expected identity tool surface drifted", ) _require( behavior_inputs.get("effective_policy") == { "cheap_model_route": None, "memory_enabled": False, "memory_nudge_interval": 0, "skill_creation_nudge_interval": 0, "smart_model_routing": False, "user_profile_enabled": False, }, "expected identity effective policy drifted", ) def build_expected_identity_contract( *, artifact: Path, receipt_path: Path, runtime: Mapping[str, Any], source_revision: str, source_root: Path, ) -> dict[str, Any]: receipt = load_json(receipt_path, "artifact verification receipt") effective_policy, tool_surface = _validate_runtime_probe( receipt, expected_config_sha256=sha256_file(artifact / "profile-template" / "config.yaml"), ) profile = artifact / "profile-template" config_path = profile / "config.yaml" config = strict_identity_manifest.behavior.safe_model_config(config_path) _require(config.get("status") == "observed", "packaged identity model config was not observed") _require(config.get("config_sha256") == receipt.get("config_sha256"), "packaged config differs from runtime receipt") _require( config.get("provider") == MODEL_PROVIDER and config.get("default_model") == DEFAULT_MODEL, "packaged provider/default model drifted", ) _require( config.get("gateway_smart_model_routing") is False and config.get("smart_routing") in (None, False) and config.get("smart_routing_model") is None, "packaged smart routing is not disabled", ) runtime_contract_path = artifact / "runtime-contract.json" runtime_contract = load_json(runtime_contract_path, "artifact runtime contract") _require(runtime_contract.get("transport_authority") == "absent", "runtime transport authority is not absent") declared_tools = (runtime_contract.get("toolset") or {}).get("allowed_tools") _require( isinstance(declared_tools, list) and sorted(declared_tools) == tool_surface["tools"], "runtime contract and probed tool surface differ", ) database = runtime_contract.get("database") or {} _require( database.get("direct_canonical_writes") == "denied" and database.get("direct_stage_table_writes") == "denied" and database.get("proposal_staging") == "function_only" and database.get("proposal_function") == PROPOSAL_FUNCTION, "runtime database proposal/write boundary drifted", ) skills = tree_manifest(profile / "skills") _require(skills["file_count"] > 0, "packaged identity skills are empty") source_bindings = _identity_contract_source_bindings(source_root) permissions = { "identity_config_sha256": config["identity_config_sha256"], "gateway_permissions_sha256": config["gateway_permissions_sha256"], "tool_permissions_sha256": config["tool_permissions_sha256"], "runtime_policy": no_send_permission_policy(), "authorization_decision_required_in_runtime_receipt": True, } behavior_inputs = { "artifact_sha256": runtime["artifact_sha256"], "artifact_receipt_sha256": runtime["artifact_receipt_sha256"], "config_sha256": config["config_sha256"], "runtime_contract_sha256": sha256_file(runtime_contract_path), "uv_lock_sha256": runtime["uv_lock_sha256"], "identity_contract_sources": source_bindings, "effective_policy": effective_policy, "tool_surface": tool_surface, } runtime_binding: dict[str, Any] = { "hermes_git_head": runtime["hermes_commit"], "hermes_source_sha256": runtime["hermes_source_sha256"], "teleo_git_head": source_revision, "teleo_source_sha256": canonical_sha256(source_bindings), "python": _locked_python_runtime(artifact, expected_uv_lock_sha256=str(runtime["uv_lock_sha256"])), } core: dict[str, Any] = { "schema": IDENTITY_RUNTIME_CONTRACT_SCHEMA, "runtime": runtime_binding, "model": { "provider": MODEL_PROVIDER, "default_model": DEFAULT_MODEL, "smart_routing": False, "smart_routing_model": None, "gateway_smart_model_routing": False, "model_section_sha256": config["model_section_sha256"], "hosted_weights": "external_provider_managed_not_locally_hashable", "actual_model_required_in_turn_receipt": True, }, "skills": {"content_sha256": skills["sha256"], "file_count": skills["file_count"]}, "permissions": permissions, "behavior_inputs": behavior_inputs, } runtime_binding["identity_runtime_sha256"] = canonical_sha256(_identity_runtime_payload(core)) value = {**core, "contract_sha256": canonical_sha256(core)} validate_expected_identity_contract(value, expected_revision=source_revision) return value def _identity_source_paths(manifest: dict[str, Any]) -> tuple[str, str, str]: inputs = manifest.get("identity_inputs") database = inputs.get("canonical_database") if isinstance(inputs, dict) else None sources = inputs.get("identity_sources") if isinstance(inputs, dict) else None raw_paths = ( (database or {}).get("source", {}).get("repo_path") if isinstance(database, dict) else None, (sources or {}).get("static_constitution", {}).get("repo_path") if isinstance(sources, dict) else None, (sources or {}).get("database_derived_identity", {}).get("repo_path") if isinstance(sources, dict) else None, ) paths: list[str] = [] for label, raw in zip(SYNTHETIC_IDENTITY_SOURCE_LABELS, raw_paths, strict=True): _require(isinstance(raw, str) and bool(raw), f"identity {label} path is missing") relative = Path(raw) _require(not relative.is_absolute() and ".." not in relative.parts, f"identity {label} path is unsafe") paths.append(relative.as_posix()) _require(len(set(paths)) == len(paths), "identity source paths are not distinct") return tuple(paths) # type: ignore[return-value] def _safe_bound_file(source_root: Path, relative: str, *, label: str) -> Path: """Resolve one repository-relative file without following any nested symlink.""" _require(source_root.is_dir() and not source_root.is_symlink(), "identity source root is missing or unsafe") relative_path = Path(relative) _require( not relative_path.is_absolute() and relative_path.parts and ".." not in relative_path.parts, f"{label} path is unsafe", ) try: resolved_root = source_root.resolve(strict=True) except OSError as exc: raise PackageError("identity source root cannot be resolved") from exc candidate = source_root for part in relative_path.parts: candidate = candidate / part _require(not candidate.is_symlink(), f"{label} has a symlink component") try: resolved = candidate.resolve(strict=True) resolved.relative_to(resolved_root) except (OSError, ValueError) as exc: raise PackageError(f"{label} escapes the identity source root") from exc _require(resolved.is_file(), f"{label} is missing or unsafe") return resolved def identity_source_manifest(manifest: dict[str, Any], source_root: Path) -> dict[str, Any]: _require(source_root.is_dir() and not source_root.is_symlink(), "identity source root is missing or unsafe") files = [] for relative in sorted(_identity_source_paths(manifest)): path = _safe_bound_file(source_root, relative, label=f"identity source: {relative}") _require(path.stat().st_size <= 1024 * 1024, f"identity source is too large: {relative}") files.append({"path": relative, "bytes": path.stat().st_size, "sha256": sha256_file(path)}) stable = {"files": files, "file_count": len(files), "total_bytes": sum(item["bytes"] for item in files)} return {**stable, "sha256": canonical_sha256(stable)} def compile_identity_bundle( *, manifest: dict[str, Any], expected_identity_contract: dict[str, Any], source_root: Path, output: Path, ) -> dict[str, Any]: expected_revision = expected_identity_contract.get("runtime", {}).get("teleo_git_head") _require(_is_revision(expected_revision), "compiled identity revision is invalid") validate_expected_identity_contract(expected_identity_contract, expected_revision=expected_revision) try: views = strict_identity_manifest.verify_bound_sources( manifest, source_root, expected_runtime_policy_value=expected_identity_contract["permissions"]["runtime_policy"], ) except strict_identity_manifest.IdentityManifestError as exc: raise PackageError("compiled identity manifest failed strict semantic validation") from exc _require(set(views) == {"SOUL.md", "identity.json"}, "compiled identity views are incomplete") _require(not output.exists() and not output.is_symlink(), "compiled identity output already exists") output.mkdir(parents=True, mode=0o700) try: for name, content in views.items(): path = output / name path.write_text(content, encoding="utf-8") path.chmod(0o600) write_json(output / "identity-manifest.json", manifest, mode=0o600) write_json( output / "identity-lock.json", { "schema": IDENTITY_LOCK_SCHEMA, "manifest_sha256": manifest["manifest_sha256"], "identity_inputs_sha256": manifest["identity_inputs_sha256"], "compiled_views": {name: {"sha256": sha256_file(output / name)} for name in sorted(views)}, "session_boundary": manifest["identity_inputs"]["session_boundary"], }, mode=0o600, ) return validate_identity_bundle( output, expected_revision=expected_revision, expected_identity_contract=expected_identity_contract, source_root=source_root, ) except BaseException: shutil.rmtree(output, ignore_errors=True) raise def build_synthetic_identity_manifest( *, artifact: Path, receipt: Path, source_revision: str, contract_source_root: Path, identity_source_root: Path, database_fingerprint: Path, constitution: Path, database_identity: Path, ) -> dict[str, Any]: """Generate one synthetic identity only through the package's strict runtime contract.""" runtime = validate_artifact(artifact, receipt, source_revision) expected_contract = build_expected_identity_contract( artifact=artifact, receipt_path=receipt, runtime=runtime, source_revision=source_revision, source_root=contract_source_root, ) try: manifest = strict_identity_manifest.build_identity_manifest( identity_runtime_contract=expected_contract, database_fingerprint_path=database_fingerprint, constitution_path=constitution, database_identity_path=database_identity, source_root=identity_source_root, ) strict_identity_manifest.verify_bound_sources( manifest, identity_source_root, expected_runtime_policy_value=expected_contract["permissions"]["runtime_policy"], ) except strict_identity_manifest.IdentityManifestError as exc: raise PackageError("synthetic identity manifest failed strict package validation") from exc return manifest def validate_identity_bundle( root: Path, *, expected_revision: str, expected_identity_contract: dict[str, Any], source_root: Path, ) -> dict[str, Any]: validate_expected_identity_contract(expected_identity_contract, expected_revision=expected_revision) bundle = identity_bundle_manifest(root) manifest = load_json(root / "identity-manifest.json", "identity manifest") lock = load_json(root / "identity-lock.json", "identity lock") _require( set(manifest) == {"schema", "identity_inputs", "identity_inputs_sha256", "compiled_views", "manifest_sha256"}, "identity manifest fields are not exact", ) _require(manifest.get("schema") == IDENTITY_MANIFEST_SCHEMA, "identity manifest schema is invalid") try: rendered = strict_identity_manifest.verify_bound_sources( manifest, source_root, expected_runtime_policy_value=expected_identity_contract["permissions"]["runtime_policy"], ) except strict_identity_manifest.IdentityManifestError as exc: raise PackageError("identity manifest failed strict semantic/source validation") from exc inputs = manifest["identity_inputs"] runtime = inputs.get("runtime") _require( inputs.get("source_commit") == (runtime or {}).get("teleo_git_head") == expected_revision, "identity source/runtime revision differs from package revision", ) for section in ("runtime", "model", "skills", "permissions"): _require( inputs.get(section) == expected_identity_contract[section], f"identity {section} differs from the expected package runtime contract", ) views = manifest.get("compiled_views") _require(isinstance(views, dict) and set(views) == set(rendered), "compiled identity views are incomplete") for name, content in rendered.items(): _require((root / name).read_text(encoding="utf-8") == content, f"compiled identity view derivation drifted: {name}") _require(views[name].get("sha256") == sha256_file(root / name), f"compiled identity view hash drifted: {name}") _require( set(lock) == {"schema", "manifest_sha256", "identity_inputs_sha256", "compiled_views", "session_boundary"}, "identity lock fields are not exact", ) _require(lock.get("schema") == IDENTITY_LOCK_SCHEMA, "identity lock schema is invalid") _require(lock.get("manifest_sha256") == manifest["manifest_sha256"], "identity lock manifest binding drifted") _require(lock.get("identity_inputs_sha256") == manifest["identity_inputs_sha256"], "identity lock input binding drifted") expected_views = {name: {"sha256": sha256_file(root / name)} for name in sorted(rendered)} _require(lock.get("compiled_views") == expected_views, "identity lock compiled-view binding drifted") _require(lock.get("session_boundary") == inputs.get("session_boundary"), "identity lock session binding drifted") database = inputs.get("canonical_database") _require(isinstance(database, dict), "identity database authority is missing") _require( database.get("authority") == "synthetic_noncanonical_fixture" and database.get("canonical_authority_granted") is False, "this package accepts only synthetic noncanonical identity", ) sources = inputs.get("identity_sources") provenance = (sources or {}).get("database_derived_identity", {}).get("provenance") if isinstance(sources, dict) else None _require( provenance == { "mode": "synthetic_noncanonical_fixture", "canonical_authority_granted": False, "same_transaction_as_fingerprint": False, "export_receipt_sha256": None, "authority": "synthetic_noncanonical_fixture", }, "synthetic identity provenance binding drifted", ) for field in ("fingerprint_sha256", "structure_sha256", "table_rows_sha256"): _require(_is_sha256(database.get(field)), f"identity database {field} is invalid") for field in ("database", "database_user", "system_identifier"): _require(isinstance(database.get(field), str) and bool(database[field]), f"identity database {field} is missing") _require( isinstance(database.get("table_count"), int) and not isinstance(database["table_count"], bool) and database["table_count"] > 0, "identity database table_count is invalid", ) _require( isinstance(database.get("total_rows"), int) and not isinstance(database["total_rows"], bool) and database["total_rows"] >= 0, "identity database total_rows is invalid", ) source_manifest = identity_source_manifest(manifest, source_root) return { "bundle_sha256": bundle["sha256"], "manifest_sha256": manifest["manifest_sha256"], "identity_inputs_sha256": manifest["identity_inputs_sha256"], "identity_lock_sha256": sha256_file(root / "identity-lock.json"), "soul_sha256": sha256_file(root / "SOUL.md"), "identity_view_sha256": sha256_file(root / "identity.json"), "identity_sources_sha256": source_manifest["sha256"], **{ field: database[field] for field in ( "database", "database_user", "system_identifier", "fingerprint_sha256", "structure_sha256", "table_rows_sha256", "table_count", "total_rows", ) }, "runtime_contract": expected_identity_contract, "authority": "synthetic_noncanonical_fixture", "canonical_authority_granted": False, } def validate_artifact(artifact: Path, receipt_path: Path, expected_revision: str) -> dict[str, Any]: _require(_is_revision(expected_revision), "expected Teleo revision is invalid") _safe_tree(artifact, "runtime artifact") manifest = load_json(artifact / "artifact-manifest.json", "artifact manifest") receipt = load_json(receipt_path, "artifact verification receipt") _require(manifest.get("schema") == ARTIFACT_MANIFEST_SCHEMA, "artifact manifest schema is invalid") _require(receipt.get("schema") == ARTIFACT_RECEIPT_SCHEMA, "artifact receipt schema is invalid") _require(receipt.get("status") == "pass", "artifact receipt did not pass") _require(receipt.get("verification_mode") == "release", "artifact receipt is not release evidence") _require(manifest.get("teleo_git_head") == expected_revision, "artifact revision differs from source revision") _require(receipt.get("teleo_git_head") == expected_revision, "receipt revision differs from source revision") content = tree_manifest(artifact, excluded=frozenset({"artifact-manifest.json"})) _require(manifest.get("content") == content, "artifact content manifest drifted") _require(receipt.get("artifact_sha256") == content["sha256"], "artifact receipt content hash drifted") _require(manifest.get("hermes_commit") == HERMES_COMMIT, "artifact Hermes commit drifted") _require(receipt.get("hermes_commit") == HERMES_COMMIT, "receipt Hermes commit drifted") _require(_is_sha256(receipt.get("hermes_source_sha256")), "receipt Hermes source hash is invalid") config_path = artifact / "profile-template" / "config.yaml" _require( config_path.is_file() and not config_path.is_symlink() and receipt.get("config_sha256") == sha256_file(config_path), "artifact config differs from the runtime receipt", ) _validate_runtime_probe(receipt, expected_config_sha256=sha256_file(config_path)) runtime_lock = load_json(artifact / "runtime-lock.json", "artifact runtime lock") _require( runtime_lock.get("base_images") == {"python": PYTHON_IMAGE.removeprefix("docker.io/library/"), "uv": UV_IMAGE}, "artifact base-image lock drifted", ) hermes = runtime_lock.get("hermes") _require(isinstance(hermes, dict) and hermes.get("commit") == HERMES_COMMIT, "artifact Hermes lock drifted") _require( hermes.get("patched_source_tree_sha256") == receipt["hermes_source_sha256"], "artifact Hermes source differs from the runtime lock", ) _require(_is_sha256(hermes.get("uv_lock_sha256")), "artifact uv lock hash is invalid") claim_ceiling = manifest.get("claim_ceiling") _require(isinstance(claim_ceiling, str) and claim_ceiling == receipt.get("claim_ceiling"), "claim ceiling drifted") return { "teleo_git_head": expected_revision, "artifact_sha256": content["sha256"], "artifact_manifest_sha256": sha256_file(artifact / "artifact-manifest.json"), "artifact_receipt_sha256": sha256_file(receipt_path), "hermes_commit": HERMES_COMMIT, "hermes_source_sha256": receipt.get("hermes_source_sha256"), "uv_lock_sha256": hermes.get("uv_lock_sha256"), "config_sha256": receipt.get("config_sha256"), "runtime_contract_sha256": sha256_file(artifact / "runtime-contract.json"), "claim_ceiling": claim_ceiling, } def _target() -> dict[str, str]: return { "project": PROJECT, "zone": ZONE, "instance": INSTANCE, "service": SERVICE, "service_account": SERVICE_ACCOUNT, "platform": PLATFORM, } def _base_images() -> dict[str, str]: return {"python": PYTHON_IMAGE, "uv": UV_IMAGE, "postgres": POSTGRES_IMAGE} def _assert_no_forbidden_descriptor_markers(value: Mapping[str, Any]) -> None: encoded = json.dumps(value, ensure_ascii=True, separators=(",", ":"), sort_keys=True).casefold() for marker in FORBIDDEN_DESCRIPTOR_MARKERS: _require( marker.casefold() not in encoded, "release descriptor contains a forbidden production/transport marker" ) def validate_image_input(value: dict[str, Any]) -> None: _require( set(value) == { "schema", "target", "runtime", "identity", "base_images", "source_bindings", "claim_ceiling", "input_sha256", }, "image input fields are not exact", ) _require(value.get("schema") == IMAGE_INPUT_SCHEMA, "image input schema is invalid") _require(value.get("target") == _target(), "image input target is not exact GCP staging") _require(value.get("base_images") == _base_images(), "image input base-image closure drifted") runtime = value.get("runtime") _require(isinstance(runtime, dict), "image runtime binding is missing") _require( set(runtime) == { "teleo_git_head", "artifact_sha256", "artifact_manifest_sha256", "artifact_receipt_sha256", "hermes_commit", "hermes_source_sha256", "uv_lock_sha256", "config_sha256", "runtime_contract_sha256", "python_version", "uv_version", "postgres_version", "cloudsql_ca_sha256", }, "image runtime binding fields are not exact", ) _require(_is_revision(runtime.get("teleo_git_head")), "image Teleo revision is invalid") for field in ( "artifact_sha256", "artifact_manifest_sha256", "artifact_receipt_sha256", "hermes_source_sha256", "uv_lock_sha256", "config_sha256", "runtime_contract_sha256", "cloudsql_ca_sha256", ): _require(_is_sha256(runtime.get(field)), f"image runtime {field} is invalid") _require(runtime.get("hermes_commit") == HERMES_COMMIT, "image Hermes commit drifted") _require(runtime.get("python_version") == PYTHON_VERSION, "image Python version drifted") _require(runtime.get("uv_version") == UV_VERSION, "image uv version drifted") _require(runtime.get("postgres_version") == POSTGRES_VERSION, "image PostgreSQL version drifted") _require(runtime.get("cloudsql_ca_sha256") == CA_SHA256, "image Cloud SQL CA drifted") identity = value.get("identity") _require( isinstance(identity, dict) and set(identity) == { "bundle_sha256", "manifest_sha256", "identity_inputs_sha256", "identity_lock_sha256", "soul_sha256", "identity_view_sha256", "identity_sources_sha256", "database", "database_user", "system_identifier", "fingerprint_sha256", "structure_sha256", "table_rows_sha256", "table_count", "total_rows", "runtime_contract", "authority", "canonical_authority_granted", }, "image identity binding fields are not exact", ) for field in ( "bundle_sha256", "manifest_sha256", "identity_inputs_sha256", "identity_lock_sha256", "soul_sha256", "identity_view_sha256", "identity_sources_sha256", "fingerprint_sha256", "structure_sha256", "table_rows_sha256", ): _require(_is_sha256(identity.get(field)), f"image identity {field} is invalid") _require( identity.get("authority") == "synthetic_noncanonical_fixture" and identity.get("canonical_authority_granted") is False, "image identity must remain synthetic and noncanonical", ) for field in ("database", "database_user", "system_identifier"): _require(isinstance(identity.get(field), str) and bool(identity[field]), f"image identity {field} is missing") _require( isinstance(identity.get("table_count"), int) and not isinstance(identity["table_count"], bool) and identity["table_count"] > 0 and isinstance(identity.get("total_rows"), int) and not isinstance(identity["total_rows"], bool) and identity["total_rows"] >= 0, "image identity database counts are invalid", ) identity_contract = identity.get("runtime_contract") _require(isinstance(identity_contract, dict), "image expected identity runtime contract is missing") validate_expected_identity_contract(identity_contract, expected_revision=runtime["teleo_git_head"]) _require( identity_contract["runtime"]["hermes_git_head"] == runtime["hermes_commit"] and identity_contract["runtime"]["hermes_source_sha256"] == runtime["hermes_source_sha256"] and identity_contract["runtime"]["teleo_git_head"] == runtime["teleo_git_head"] and identity_contract["runtime"]["python"]["python_version"] == runtime["python_version"], "image identity contract differs from the packaged runtime", ) behavior_inputs = identity_contract["behavior_inputs"] _require( behavior_inputs["artifact_sha256"] == runtime["artifact_sha256"] and behavior_inputs["artifact_receipt_sha256"] == runtime["artifact_receipt_sha256"] and behavior_inputs["config_sha256"] == runtime["config_sha256"] and behavior_inputs["runtime_contract_sha256"] == runtime["runtime_contract_sha256"] and behavior_inputs["uv_lock_sha256"] == runtime["uv_lock_sha256"], "image identity behavior inputs differ from the packaged runtime", ) bindings = value.get("source_bindings") _require( isinstance(bindings, dict) and set(bindings) == { "Dockerfile", "entrypoint.py", "package_contract.py", "scripts/leo_behavior_manifest.py", "scripts/leo_identity_manifest.py", } and all(_is_sha256(item) for item in bindings.values()), "image packaging source bindings are invalid", ) _require( identity_contract["behavior_inputs"]["identity_contract_sources"] == { "scripts/leo_behavior_manifest.py": bindings["scripts/leo_behavior_manifest.py"], "scripts/leo_identity_manifest.py": bindings["scripts/leo_identity_manifest.py"], }, "image identity contract differs from the packaged identity validators", ) _require( isinstance(value.get("claim_ceiling"), str) and bool(value["claim_ceiling"]), "image claim ceiling is missing" ) stable = {key: item for key, item in value.items() if key != "input_sha256"} _require(value.get("input_sha256") == canonical_sha256(stable), "image input self-hash drifted") _assert_no_forbidden_descriptor_markers(value) def build_image_input( *, artifact: Path, receipt: Path, identity: Path, ca: Path, source_revision: str, dockerfile: Path, entrypoint: Path, package_contract: Path, contract_source_root: Path, identity_source_root: Path, ) -> dict[str, Any]: _require(sha256_file(ca) == CA_SHA256, "Cloud SQL CA does not match the reviewed certificate") runtime = validate_artifact(artifact, receipt, source_revision) runtime.update( { "python_version": PYTHON_VERSION, "uv_version": UV_VERSION, "postgres_version": POSTGRES_VERSION, "cloudsql_ca_sha256": CA_SHA256, } ) expected_identity_contract = build_expected_identity_contract( artifact=artifact, receipt_path=receipt, runtime=runtime, source_revision=source_revision, source_root=contract_source_root, ) identity_binding = validate_identity_bundle( identity, expected_revision=source_revision, expected_identity_contract=expected_identity_contract, source_root=identity_source_root, ) identity_contract_sources = _identity_contract_source_bindings(contract_source_root) identity_manifest = load_json(identity / "identity-manifest.json", "identity manifest") identity_sources = identity_source_manifest(identity_manifest, identity_source_root) _require( contract_source_root.resolve() == identity_source_root.resolve(), "synthetic identity sources must use the reviewed package repository", ) fixed_source_hashes = { "docker/leoclean-nosend/Dockerfile": sha256_file(dockerfile), "docker/leoclean-nosend/entrypoint.py": sha256_file(entrypoint), "ops/gcp-teleo-pgvector-standby-server-ca.pem": sha256_file(ca), "ops/gcp_leoclean_nosend_package.py": sha256_file(package_contract), **identity_contract_sources, } reviewed_source_hashes = { **fixed_source_hashes, **{str(entry["path"]): str(entry["sha256"]) for entry in identity_sources["files"]}, } _verify_git_source_closure(contract_source_root, source_revision, reviewed_source_hashes) stable = { "schema": IMAGE_INPUT_SCHEMA, "target": _target(), "runtime": {key: runtime[key] for key in runtime if key != "claim_ceiling"}, "identity": identity_binding, "base_images": _base_images(), "source_bindings": { "Dockerfile": fixed_source_hashes["docker/leoclean-nosend/Dockerfile"], "entrypoint.py": fixed_source_hashes["docker/leoclean-nosend/entrypoint.py"], "package_contract.py": fixed_source_hashes["ops/gcp_leoclean_nosend_package.py"], **identity_contract_sources, }, "claim_ceiling": runtime["claim_ceiling"], } value = {**stable, "input_sha256": canonical_sha256(stable)} validate_image_input(value) return value def validate_installed_runtime_identity( image_input: dict[str, Any], *, artifact: Path, receipt: Path, identity: Path, contract_source_root: Path, identity_source_root: Path, ) -> tuple[dict[str, Any], dict[str, Any]]: """Re-derive installed runtime and identity authority instead of trusting the descriptor.""" validate_image_input(image_input) revision = image_input["runtime"]["teleo_git_head"] runtime = validate_artifact(artifact, receipt, revision) for field in ( "teleo_git_head", "artifact_sha256", "artifact_manifest_sha256", "artifact_receipt_sha256", "hermes_commit", "hermes_source_sha256", "uv_lock_sha256", "config_sha256", "runtime_contract_sha256", ): _require(runtime.get(field) == image_input["runtime"].get(field), f"installed runtime binding drifted: {field}") expected_identity_contract = build_expected_identity_contract( artifact=artifact, receipt_path=receipt, runtime=runtime, source_revision=revision, source_root=contract_source_root, ) _require( expected_identity_contract == image_input["identity"]["runtime_contract"], "installed identity runtime contract differs from its installed sources", ) identity_binding = validate_identity_bundle( identity, expected_revision=revision, expected_identity_contract=expected_identity_contract, source_root=identity_source_root, ) _require(identity_binding == image_input["identity"], "installed identity bundle drifted") return runtime, identity_binding def _copy_identity_sources(manifest: dict[str, Any], source_root: Path, output: Path) -> None: expected = identity_source_manifest(manifest, source_root) output.mkdir(parents=True, mode=0o755) for entry in expected["files"]: relative = entry["path"] destination = output / relative destination.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(_safe_bound_file(source_root, relative, label=f"identity source: {relative}"), destination) destination.chmod(0o644) copied = identity_source_manifest(manifest, output) _require(copied == expected, "copied identity sources differ from their reviewed inputs") def prepare_context( *, repo_root: Path, artifact: Path, receipt: Path, identity: Path, identity_source_root: Path, output: Path, source_revision: str, ) -> dict[str, Any]: _require(not output.exists() and not output.is_symlink(), "output context already exists") dockerfile = repo_root / "docker" / "leoclean-nosend" / "Dockerfile" entrypoint = repo_root / "docker" / "leoclean-nosend" / "entrypoint.py" package_contract = repo_root / "ops" / "gcp_leoclean_nosend_package.py" ca = repo_root / "ops" / "gcp-teleo-pgvector-standby-server-ca.pem" for path, label in ( (dockerfile, "Dockerfile"), (entrypoint, "entrypoint"), (package_contract, "package contract"), (ca, "Cloud SQL CA"), ): _require(path.is_file() and not path.is_symlink(), f"{label} source is missing or unsafe") image_input = build_image_input( artifact=artifact, receipt=receipt, identity=identity, ca=ca, source_revision=source_revision, dockerfile=dockerfile, entrypoint=entrypoint, package_contract=package_contract, contract_source_root=repo_root, identity_source_root=identity_source_root, ) identity_manifest = load_json(identity / "identity-manifest.json", "identity manifest") output.mkdir(parents=True, mode=0o700) try: shutil.copytree(artifact, output / "artifact", symlinks=False) shutil.copytree(identity, output / "identity", symlinks=False) for name in IDENTITY_FILES: (output / "identity" / name).chmod(0o600) _copy_identity_sources(identity_manifest, identity_source_root, output / "identity-sources") shutil.copyfile(receipt, output / "artifact-receipt.json") shutil.copyfile(ca, output / "cloudsql-server-ca.pem") shutil.copyfile(dockerfile, output / "Dockerfile") shutil.copyfile(entrypoint, output / "entrypoint.py") shutil.copyfile(package_contract, output / "package_contract.py") scripts = output / "scripts" scripts.mkdir(mode=0o755) for relative in IDENTITY_CONTRACT_SOURCES: destination = output / relative shutil.copyfile(repo_root / relative, destination) destination.chmod(0o644) for name in ( "artifact-receipt.json", "cloudsql-server-ca.pem", "Dockerfile", "entrypoint.py", "package_contract.py", ): (output / name).chmod(0o644) write_json(output / "image-input.json", image_input) write_json( output / "build-args.json", { "ARTIFACT_SHA256": image_input["runtime"]["artifact_sha256"], "IDENTITY_SHA256": image_input["identity"]["bundle_sha256"], "INPUT_SHA256": image_input["input_sha256"], "TELEO_REVISION": source_revision, }, ) _safe_tree(output, "prepared image context") return image_input except BaseException: shutil.rmtree(output, ignore_errors=True) raise def build_release_descriptor(image_input: dict[str, Any], image_reference: str) -> dict[str, Any]: validate_image_input(image_input) match = IMAGE_REFERENCE.fullmatch(image_reference) _require(match is not None, "release image must be the exact staging repository at a sha256 digest") digest = f"sha256:{match.group(1)}" labels = expected_image_labels(image_input) stable = { "schema": RELEASE_SCHEMA, "target": _target(), "image": { "reference": image_reference, "digest": digest, "platform": PLATFORM, "labels": labels, "input_sha256": image_input["input_sha256"], }, "image_input": image_input, } value = {**stable, "release_sha256": canonical_sha256(stable)} validate_release_descriptor(value) return value def expected_image_labels(image_input: dict[str, Any]) -> dict[str, str]: validate_image_input(image_input) return { "livingip.artifact.sha256": image_input["runtime"]["artifact_sha256"], "livingip.identity.sha256": image_input["identity"]["bundle_sha256"], "livingip.input.sha256": image_input["input_sha256"], "livingip.runtime": "leoclean-nosend", "livingip.target": "gcp-staging", "org.opencontainers.image.revision": image_input["runtime"]["teleo_git_head"], "org.opencontainers.image.source": "https://github.com/living-ip/teleo-infrastructure", } def validate_release_descriptor(value: dict[str, Any]) -> None: _require( set(value) == {"schema", "target", "image", "image_input", "release_sha256"}, "release descriptor fields are not exact", ) _require(value.get("schema") == RELEASE_SCHEMA, "release descriptor schema is invalid") _require(value.get("target") == _target(), "release target is not exact GCP staging") image_input = value.get("image_input") _require(isinstance(image_input, dict), "release image input is missing") validate_image_input(image_input) image = value.get("image") _require( isinstance(image, dict) and set(image) == {"reference", "digest", "platform", "labels", "input_sha256"}, "release image fields are not exact", ) reference = image.get("reference") match = IMAGE_REFERENCE.fullmatch(reference) if isinstance(reference, str) else None _require(match is not None, "release image is not digest-pinned in the staging repository") _require(image.get("digest") == f"sha256:{match.group(1)}", "release image digest differs from its reference") _require(image.get("platform") == PLATFORM, "release platform is not exact") _require(image.get("input_sha256") == image_input["input_sha256"], "release image input binding drifted") _require(image.get("labels") == expected_image_labels(image_input), "release image labels are not exact") stable = {key: item for key, item in value.items() if key != "release_sha256"} _require(value.get("release_sha256") == canonical_sha256(stable), "release descriptor self-hash drifted") _assert_no_forbidden_descriptor_markers(value) def render_systemd_unit(release: dict[str, Any]) -> str: validate_release_descriptor(release) image = release["image"]["reference"] release_sha = release["release_sha256"] profile_tmpfs = f"{PROFILE_ROOT}:rw,noexec,nosuid,nodev,size=64m,uid=0,gid=0,mode=0700" temp_tmpfs = f"/tmp:rw,noexec,nosuid,nodev,size=64m,uid={RUNTIME_UID},gid={RUNTIME_GID},mode=0700" postgres_data_tmpfs = "/var/lib/postgresql/data:ro,noexec,nosuid,nodev,size=64k,uid=0,gid=0,mode=0000" exec_start = " ".join( [ "/usr/bin/docker run", "--rm", f"--name={CONTAINER_NAME}", "--pull=never", "--read-only", "--cap-drop=ALL", "--cap-add=CHOWN", "--cap-add=SETGID", "--cap-add=SETUID", "--cap-add=SETPCAP", "--security-opt=no-new-privileges:true", "--pids-limit=512", "--network=bridge", "--stop-timeout=30", f"--tmpfs={profile_tmpfs}", f"--tmpfs={temp_tmpfs}", f"--tmpfs={postgres_data_tmpfs}", f"--label=livingip.release.sha256={release_sha}", image, "service", ] ) unit = f"""[Unit] Description=LivingIP leoclean-gcp-nosend staging Documentation=https://github.com/living-ip/teleo-infrastructure After=docker.service network-online.target Requires=docker.service Wants=network-online.target [Service] Type=simple ExecStart={exec_start} ExecStop=/usr/bin/docker stop --time=30 {CONTAINER_NAME} ExecStopPost=-/usr/bin/docker rm --force {CONTAINER_NAME} Restart=on-failure RestartSec=5s TimeoutStartSec=120s TimeoutStopSec=45s UMask=0077 NoNewPrivileges=yes CapabilityBoundingSet= AmbientCapabilities= PrivateDevices=yes PrivateTmp=yes ProtectSystem=strict ProtectHome=yes ProtectKernelTunables=yes ProtectKernelModules=yes ProtectKernelLogs=yes ProtectControlGroups=yes LockPersonality=yes RestrictSUIDSGID=yes RestrictRealtime=yes RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6 SystemCallArchitectures=native [Install] WantedBy=multi-user.target """ _require("--publish" not in unit and " -p " not in unit, "service unit publishes a port") _require("docker.sock" not in unit, "service unit mounts the Docker socket") _require( "EnvironmentFile" not in unit and "LoadCredential" not in unit, "service unit adds an unreviewed credential source", ) _require("telegram" not in unit.casefold(), "service unit contains a messaging marker") return unit def _git_commit(repo_root: Path) -> str: try: revision = subprocess.run( ["git", "-C", str(repo_root), "rev-parse", "--verify", "HEAD^{commit}"], check=True, capture_output=True, text=True, ).stdout.strip() except (OSError, subprocess.CalledProcessError) as exc: raise PackageError("cannot determine source revision") from exc _require(_is_revision(revision), "source revision is invalid") return revision def _verify_git_source_closure( repo_root: Path, revision: str, expected_sha256: Mapping[str, str], ) -> None: """Require every executable package input to equal its reviewed Git blob and mode.""" _require(_is_revision(revision), "source revision is invalid") _require(repo_root.is_dir() and not repo_root.is_symlink(), "source repository root is missing or unsafe") try: top_level = subprocess.run( ["git", "-C", str(repo_root), "rev-parse", "--show-toplevel"], check=True, capture_output=True, text=True, ).stdout.strip() except (OSError, subprocess.CalledProcessError) as exc: raise PackageError("cannot determine source repository root") from exc _require(Path(top_level).resolve() == repo_root.resolve(), "source root is not the Git repository root") _require(_git_commit(repo_root) == revision, "source checkout differs from the reviewed revision") relative_paths = sorted(expected_sha256) _require(relative_paths and len(relative_paths) == len(expected_sha256), "reviewed source closure is invalid") try: dirty = subprocess.run( [ "git", "-C", str(repo_root), "status", "--porcelain=v1", "-z", "--untracked-files=all", "--", *relative_paths, ], check=True, capture_output=True, ).stdout except (OSError, subprocess.CalledProcessError) as exc: raise PackageError("cannot inspect reviewed source closure") from exc _require(not dirty, "reviewed source closure is dirty") for relative in relative_paths: _require(_is_sha256(expected_sha256[relative]), f"reviewed source hash is invalid: {relative}") _require(":" not in relative and "\x00" not in relative, "source path is unsafe") path = _safe_bound_file(repo_root, relative, label=f"reviewed source: {relative}") try: entry = subprocess.run( ["git", "-C", str(repo_root), "ls-tree", "-z", revision, "--", relative], check=True, capture_output=True, ).stdout _require(entry.endswith(b"\x00") and entry.count(b"\x00") == 1, f"source is not tracked: {relative}") metadata, tracked_path = entry[:-1].split(b"\t", 1) mode, object_type, object_id = metadata.split(b" ", 2) _require( tracked_path.decode("utf-8") == relative and object_type == b"blob" and mode in {b"100644", b"100755"}, f"reviewed source entry is invalid: {relative}", ) committed = subprocess.run( ["git", "-C", str(repo_root), "cat-file", "blob", object_id.decode("ascii")], check=True, capture_output=True, ).stdout except (OSError, subprocess.CalledProcessError, UnicodeDecodeError, ValueError) as exc: raise PackageError(f"cannot verify reviewed source: {relative}") from exc _require(path.read_bytes() == committed, f"source differs from reviewed revision: {relative}") _require( hashlib.sha256(committed).hexdigest() == expected_sha256[relative], f"source hash differs from package binding: {relative}", ) executable = bool(stat.S_IMODE(path.stat().st_mode) & 0o111) _require(executable == (mode == b"100755"), f"source mode differs from reviewed revision: {relative}") def _git_head(repo_root: Path) -> str: revision = _git_commit(repo_root) expected = { relative: sha256_file(_safe_bound_file(repo_root, relative, label=f"reviewed source: {relative}")) for relative in PACKAGE_SOURCE_PATHS } _verify_git_source_closure(repo_root, revision, expected) return revision def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) subparsers = parser.add_subparsers(dest="command", required=True) prepare = subparsers.add_parser("prepare") prepare.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1]) prepare.add_argument("--artifact", type=Path, required=True) prepare.add_argument("--receipt", type=Path, required=True) prepare.add_argument("--identity", type=Path, required=True) prepare.add_argument("--identity-source-root", type=Path, required=True) prepare.add_argument("--output", type=Path, required=True) compile_identity = subparsers.add_parser("compile-identity") compile_identity.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1]) compile_identity.add_argument("--artifact", type=Path, required=True) compile_identity.add_argument("--receipt", type=Path, required=True) compile_identity.add_argument("--identity-manifest", type=Path, required=True) compile_identity.add_argument("--identity-source-root", type=Path, required=True) compile_identity.add_argument("--output", type=Path, required=True) identity_contract = subparsers.add_parser("identity-contract") identity_contract.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1]) identity_contract.add_argument("--artifact", type=Path, required=True) identity_contract.add_argument("--receipt", type=Path, required=True) identity_contract.add_argument("--output", type=Path, required=True) generate_identity = subparsers.add_parser("generate-identity-manifest") generate_identity.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1]) generate_identity.add_argument("--artifact", type=Path, required=True) generate_identity.add_argument("--receipt", type=Path, required=True) generate_identity.add_argument("--database-fingerprint", type=Path, required=True) generate_identity.add_argument("--constitution", type=Path, required=True) generate_identity.add_argument("--database-identity", type=Path, required=True) generate_identity.add_argument("--identity-source-root", type=Path, required=True) generate_identity.add_argument("--output", type=Path, required=True) finalize = subparsers.add_parser("finalize") finalize.add_argument("--image-input", type=Path, required=True) finalize.add_argument("--image-reference", required=True) finalize.add_argument("--output-release", type=Path, required=True) finalize.add_argument("--output-unit", type=Path, required=True) validate = subparsers.add_parser("validate") validate.add_argument("--release", type=Path, required=True) return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: try: args = _parse_args(argv) if args.command == "prepare": repo_root = args.repo_root.resolve() result = prepare_context( repo_root=repo_root, artifact=args.artifact.resolve(), receipt=args.receipt.resolve(), identity=args.identity.resolve(), identity_source_root=args.identity_source_root.resolve(), output=args.output.absolute(), source_revision=_git_head(repo_root), ) elif args.command == "compile-identity": repo_root = args.repo_root.resolve() revision = _git_head(repo_root) artifact = args.artifact.resolve() receipt = args.receipt.resolve() runtime = validate_artifact(artifact, receipt, revision) contract = build_expected_identity_contract( artifact=artifact, receipt_path=receipt, runtime=runtime, source_revision=revision, source_root=repo_root, ) binding = compile_identity_bundle( manifest=load_json(args.identity_manifest.resolve(), "identity manifest"), expected_identity_contract=contract, source_root=args.identity_source_root.resolve(), output=args.output.absolute(), ) result = {"schema": COMPILED_IDENTITY_SCHEMA, **binding} elif args.command == "identity-contract": repo_root = args.repo_root.resolve() revision = _git_head(repo_root) artifact = args.artifact.resolve() receipt = args.receipt.resolve() runtime = validate_artifact(artifact, receipt, revision) result = build_expected_identity_contract( artifact=artifact, receipt_path=receipt, runtime=runtime, source_revision=revision, source_root=repo_root, ) _require(not args.output.exists() and not args.output.is_symlink(), "identity contract output exists") args.output.parent.mkdir(parents=True, exist_ok=True) write_json(args.output, result) elif args.command == "generate-identity-manifest": repo_root = args.repo_root.resolve() result = build_synthetic_identity_manifest( artifact=args.artifact.resolve(), receipt=args.receipt.resolve(), source_revision=_git_head(repo_root), contract_source_root=repo_root, identity_source_root=args.identity_source_root.resolve(), database_fingerprint=args.database_fingerprint.resolve(), constitution=args.constitution.resolve(), database_identity=args.database_identity.resolve(), ) _require(not args.output.exists() and not args.output.is_symlink(), "identity manifest output exists") args.output.parent.mkdir(parents=True, exist_ok=True) write_json(args.output, result, mode=0o600) elif args.command == "finalize": image_input = load_json(args.image_input, "image input") result = build_release_descriptor(image_input, args.image_reference) for path in (args.output_release, args.output_unit): _require(not path.exists() and not path.is_symlink(), "finalize output already exists") path.parent.mkdir(parents=True, exist_ok=True) write_json(args.output_release, result) args.output_unit.write_text(render_systemd_unit(result), encoding="utf-8") args.output_unit.chmod(0o644) else: result = load_json(args.release, "release descriptor") validate_release_descriptor(result) print(json.dumps({"status": "pass", "schema": result["schema"]}, sort_keys=True)) return 0 except PackageError as exc: print(json.dumps({"status": "fail", "error": str(exc)}, sort_keys=True), file=sys.stderr) return 65 if __name__ == "__main__": raise SystemExit(main())