#!/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", ) 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 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": ["skill_view", "skills_list", "terminal"], } def _runtime_surface(value: Any, label: str) -> dict[str, Any]: _require(isinstance(value, dict), f"{label} is missing") surface = { key: value.get(key) for key in ( "registry_sealed", "send_message_present", "terminal_handler", "terminal_restricted_to", "tools", ) } _require( surface == { "registry_sealed": True, "send_message_present": False, "terminal_handler": "livingip.restricted_teleo_kb.v1", "terminal_restricted_to": "profile/bin/teleo-kb", "tools": ["skill_view", "skills_list", "terminal"], }, f"{label} is not the exact sealed no-send tool surface", ) return surface 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") == { "registry_sealed": True, "send_message_present": False, "terminal_handler": "livingip.restricted_teleo_kb.v1", "terminal_restricted_to": "profile/bin/teleo-kb", "tools": ["skill_view", "skills_list", "terminal"], }, "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) 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": sha256_file(dockerfile), "entrypoint.py": sha256_file(entrypoint), "package_contract.py": sha256_file(package_contract), **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, ) 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) identity_manifest = load_json(identity / "identity-manifest.json", "identity manifest") _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_head(repo_root: Path) -> str: try: return subprocess.run( ["git", "-C", str(repo_root), "rev-parse", "HEAD"], check=True, capture_output=True, text=True, ).stdout.strip() except (OSError, subprocess.CalledProcessError) as exc: raise PackageError("cannot determine source revision") from exc 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())