teleo/teleo-infrastructure
8
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions 1
teleo-infrastructure/ops/gcp_leoclean_nosend_package.py

2410 lines
101 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""Build and validate the immutable GCP leoclean no-send service package."""
from __future__ import annotations
import argparse
import ctypes
import errno
import fcntl
import hashlib
import json
import os
import re
import shutil
import stat
import subprocess
import sys
import tomllib
import uuid
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"
IMAGE_INSPECTION_SCHEMA = "livingip.leocleanNoSendImageInspection.v1"
RELEASE_SCHEMA = "livingip.leocleanNoSendRelease.v2"
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
MAX_DOCKER_INSPECTION_BYTES = 1024 * 1024
IMAGE_RUNTIME_CONFIG = {
"user": None,
"environment": [
"PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/usr/lib/postgresql/16/bin",
"GOSU_VERSION=1.19",
"LANG=en_US.utf8",
"PG_MAJOR=16",
"PG_VERSION=16.14-1.pgdg12+1",
"PGDATA=/var/lib/postgresql/data",
"PYTHONDONTWRITEBYTECODE=1",
"PYTHONNOUSERSITE=1",
"PYTHONSAFEPATH=1",
],
"working_dir": "",
"exposed_ports": {"5432/tcp": {}},
"volumes": {"/var/lib/postgresql/data": {}},
"shell": None,
"on_build": None,
"args_escaped": False,
"stop_timeout": None,
"tty": False,
"open_stdin": False,
"stdin_once": False,
"attach_stdin": False,
"attach_stdout": False,
"attach_stderr": False,
"network_disabled": False,
"hostname": "",
"domainname": "",
"mac_address": "",
"entrypoint": [
"/opt/livingip/leoclean-nosend/venv/bin/python",
"/opt/livingip/leoclean-nosend/entrypoint.py",
],
"cmd": ["service"],
"healthcheck": {
"test": [
"CMD",
"/opt/livingip/leoclean-nosend/venv/bin/python",
"/opt/livingip/leoclean-nosend/entrypoint.py",
"health",
],
"interval_ns": 15_000_000_000,
"timeout_ns": 10_000_000_000,
"start_period_ns": 30_000_000_000,
"start_interval_ns": 0,
"retries": 4,
},
"stop_signal": "SIGTERM",
}
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:
try:
encoded = json.dumps(
value,
allow_nan=False,
ensure_ascii=True,
separators=(",", ":"),
sort_keys=True,
).encode("utf-8")
except (TypeError, ValueError) as exc:
raise PackageError("cannot canonically encode non-finite or unsupported JSON") from exc
return hashlib.sha256(encoded).hexdigest()
def _strict_json_loads(encoded: str, label: str) -> Any:
def reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
value: dict[str, Any] = {}
for key, item in pairs:
if key in value:
raise ValueError("duplicate JSON key")
value[key] = item
return value
def reject_constant(_value: str) -> Any:
raise ValueError("invalid JSON constant")
try:
return json.loads(
encoded,
object_pairs_hook=reject_duplicate_keys,
parse_constant=reject_constant,
)
except (json.JSONDecodeError, ValueError) as exc:
raise PackageError(f"{label} is invalid JSON") from exc
def load_json(path: Path, label: str) -> dict[str, Any]:
try:
encoded = path.read_text(encoding="utf-8")
except (OSError, UnicodeError) as exc:
raise PackageError(f"cannot read {label}: {type(exc).__name__}") from exc
value = _strict_json_loads(encoded, label)
_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:
try:
encoded = json.dumps(value, allow_nan=False, indent=2, sort_keys=True) + "\n"
except (TypeError, ValueError) as exc:
raise PackageError("cannot encode non-finite or unsupported JSON") from exc
path.write_text(encoded, 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 validate_image_runtime_configuration(inspected: dict[str, Any]) -> dict[str, Any]:
"""Validate the runtime-bearing Docker config shared by local and registry inspection."""
config = inspected.get("Config")
healthcheck = config.get("Healthcheck") if isinstance(config, dict) else None
raw_start_interval = healthcheck.get("StartInterval") if isinstance(healthcheck, dict) else None
raw_args_escaped = config.get("ArgsEscaped") if isinstance(config, dict) else None
expected_healthcheck_keys = {"Test", "Interval", "Timeout", "StartPeriod", "Retries"}
allowed_healthcheck_keys = (expected_healthcheck_keys, expected_healthcheck_keys | {"StartInterval"})
boolean_defaults = {
"Tty": False,
"OpenStdin": False,
"StdinOnce": False,
"AttachStdin": False,
"AttachStdout": False,
"AttachStderr": False,
"NetworkDisabled": False,
}
string_defaults = {"WorkingDir": "", "Hostname": "", "Domainname": "", "MacAddress": ""}
allowed_config_keys = {
"ArgsEscaped",
"AttachStderr",
"AttachStdin",
"AttachStdout",
"Cmd",
"Domainname",
"Entrypoint",
"Env",
"ExposedPorts",
"Healthcheck",
"Hostname",
"Image",
"Labels",
"MacAddress",
"NetworkDisabled",
"OnBuild",
"OpenStdin",
"Shell",
"StdinOnce",
"StopSignal",
"StopTimeout",
"Tty",
"User",
"Volumes",
"WorkingDir",
}
_require(
isinstance(config, dict)
and isinstance(healthcheck, dict)
and set(config) <= allowed_config_keys
and config.get("User") in (None, "")
and config.get("Image") in (None, "")
# ArgsEscaped is a deprecated, Windows-only compatibility field. Linux
# builders may omit it or serialize either boolean; exact argv arrays
# below remain authoritative for this linux/amd64 image.
and (raw_args_escaped is None or type(raw_args_escaped) is bool)
and all(type(config.get(name, default)) is bool for name, default in boolean_defaults.items())
and all(config.get(name, default) is default for name, default in boolean_defaults.items())
and all(isinstance(config.get(name, default), str) for name, default in string_defaults.items())
and all(config.get(name, default) == default for name, default in string_defaults.items())
and config.get("StopTimeout") is None
and set(healthcheck) in allowed_healthcheck_keys
and all(
type(healthcheck.get(name)) is int
for name in ("Interval", "Timeout", "StartPeriod", "Retries")
)
and (
raw_start_interval is None
or (
isinstance(raw_start_interval, int)
and not isinstance(raw_start_interval, bool)
and raw_start_interval == 0
)
),
"inspected image runtime config is invalid",
)
start_interval = 0
runtime_config = {
"user": None,
"environment": config.get("Env"),
"working_dir": config.get("WorkingDir", ""),
"exposed_ports": config.get("ExposedPorts"),
"volumes": config.get("Volumes"),
"shell": config.get("Shell"),
"on_build": config.get("OnBuild"),
"args_escaped": False,
"stop_timeout": config.get("StopTimeout"),
"tty": config.get("Tty", False),
"open_stdin": config.get("OpenStdin", False),
"stdin_once": config.get("StdinOnce", False),
"attach_stdin": config.get("AttachStdin", False),
"attach_stdout": config.get("AttachStdout", False),
"attach_stderr": config.get("AttachStderr", False),
"network_disabled": config.get("NetworkDisabled", False),
"hostname": config.get("Hostname", ""),
"domainname": config.get("Domainname", ""),
"mac_address": config.get("MacAddress", ""),
"entrypoint": config.get("Entrypoint"),
"cmd": config.get("Cmd"),
"healthcheck": {
"test": healthcheck.get("Test"),
"interval_ns": healthcheck.get("Interval"),
"timeout_ns": healthcheck.get("Timeout"),
"start_period_ns": healthcheck.get("StartPeriod"),
"start_interval_ns": start_interval,
"retries": healthcheck.get("Retries"),
},
"stop_signal": config.get("StopSignal"),
}
_require(runtime_config == IMAGE_RUNTIME_CONFIG, "inspected image runtime config is invalid")
return runtime_config
def validate_image_configuration(image_input: dict[str, Any], inspected: dict[str, Any]) -> dict[str, Any]:
"""Validate and normalize the immutable runtime-bearing image configuration."""
validate_image_input(image_input)
config_digest = inspected.get("Id")
_require(
isinstance(config_digest, str)
and config_digest.startswith("sha256:")
and _is_sha256(config_digest.removeprefix("sha256:")),
"inspected image config digest is invalid",
)
variant = inspected.get("Variant")
_require(
inspected.get("Os") == "linux"
and inspected.get("Architecture") == "amd64"
and (variant is None or variant == ""),
"inspected image platform is not linux/amd64",
)
config = inspected.get("Config")
labels = config.get("Labels") if isinstance(config, dict) else None
expected_labels = expected_image_labels(image_input)
_require(
isinstance(labels, dict)
and all(isinstance(key, str) and isinstance(item, str) for key, item in labels.items())
and all(labels.get(key) == item for key, item in expected_labels.items())
and not any(key.startswith("livingip.") and key not in expected_labels for key in labels),
"inspected image labels are not exact",
)
runtime_config = validate_image_runtime_configuration(inspected)
return {
"config_digest": config_digest,
"platform": PLATFORM,
"labels": expected_labels,
"runtime_config": runtime_config,
}
def build_image_inspection(
image_input: dict[str, Any],
image_reference: str,
inspected: dict[str, Any],
) -> dict[str, Any]:
"""Normalize one Docker inspection after an exact digest pull."""
validate_image_input(image_input)
match = IMAGE_REFERENCE.fullmatch(image_reference)
_require(match is not None, "inspected image must be the exact staging repository at a sha256 digest")
digest = f"sha256:{match.group(1)}"
observed = validate_image_configuration(image_input, inspected)
repository_digests = inspected.get("RepoDigests")
_require(
isinstance(repository_digests, list)
and all(isinstance(item, str) for item in repository_digests)
and len(repository_digests) == len(set(repository_digests))
and image_reference in repository_digests,
"inspected image is not bound to the requested repository digest",
)
value = {
"schema": IMAGE_INSPECTION_SCHEMA,
"status": "pass",
"reference": image_reference,
"digest": digest,
"config_digest": observed["config_digest"],
"platform": PLATFORM,
"labels": observed["labels"],
"runtime_config": observed["runtime_config"],
"input_sha256": image_input["input_sha256"],
"source_revision": image_input["runtime"]["teleo_git_head"],
}
validate_image_inspection(value, image_input=image_input)
return value
def validate_image_inspection(value: dict[str, Any], *, image_input: dict[str, Any]) -> None:
validate_image_input(image_input)
_require(
set(value)
== {
"schema",
"status",
"reference",
"digest",
"config_digest",
"platform",
"labels",
"runtime_config",
"input_sha256",
"source_revision",
},
"image inspection fields are not exact",
)
_require(
value.get("schema") == IMAGE_INSPECTION_SCHEMA and value.get("status") == "pass",
"image inspection did not pass",
)
reference = value.get("reference")
match = IMAGE_REFERENCE.fullmatch(reference) if isinstance(reference, str) else None
_require(match is not None, "image inspection reference is not the exact staging digest")
_require(value.get("digest") == f"sha256:{match.group(1)}", "image inspection digest differs from its reference")
config_digest = value.get("config_digest")
_require(
isinstance(config_digest, str)
and config_digest.startswith("sha256:")
and _is_sha256(config_digest.removeprefix("sha256:")),
"image inspection config digest is invalid",
)
_require(value.get("platform") == PLATFORM, "image inspection platform is not exact")
_require(value.get("labels") == expected_image_labels(image_input), "image inspection labels are not exact")
_require(value.get("runtime_config") == IMAGE_RUNTIME_CONFIG, "image inspection runtime config is not exact")
_require(value.get("input_sha256") == image_input["input_sha256"], "image inspection input binding drifted")
_require(
value.get("source_revision") == image_input["runtime"]["teleo_git_head"],
"image inspection source revision drifted",
)
_assert_no_forbidden_descriptor_markers(value)
def _load_docker_inspection(raw: bytes) -> Any:
_require(len(raw) <= MAX_DOCKER_INSPECTION_BYTES, "registry image inspection output is too large")
try:
encoded = raw.decode("utf-8")
return _strict_json_loads(encoded, "registry image inspection output")
except (UnicodeDecodeError, PackageError) as exc:
raise PackageError("registry image inspection output is invalid") from exc
def _docker_environment() -> dict[str, str]:
return {name: value for name, value in os.environ.items() if not name.startswith("DOCKER_")}
def _validate_docker_authority(docker_binary: Path, docker_host: str, docker_config: Path) -> None:
_require(docker_binary.is_absolute(), "Docker binary must be absolute")
_require("\x00" not in str(docker_binary) and "\n" not in str(docker_binary), "Docker binary is invalid")
_require(docker_host.startswith("unix://"), "Docker host must be a local Unix endpoint")
socket_path = Path(docker_host.removeprefix("unix://"))
_require(
socket_path.is_absolute() and ".." not in socket_path.parts and "\x00" not in docker_host and "\n" not in docker_host,
"Docker host must be a local Unix endpoint",
)
_require(
docker_config.is_absolute() and docker_config.is_dir() and not docker_config.is_symlink(),
"Docker config directory is missing or unsafe",
)
class _DockerCacheLock:
def __init__(self, docker_config: Path) -> None:
self._docker_config = docker_config
self._descriptor: int | None = None
def __enter__(self) -> None:
flags = os.O_RDONLY
if hasattr(os, "O_DIRECTORY"):
flags |= os.O_DIRECTORY
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
try:
descriptor = os.open(self._docker_config, flags)
fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError as exc:
if "descriptor" in locals():
os.close(descriptor)
raise PackageError("Docker cache authority is already in use or unsafe") from exc
self._descriptor = descriptor
def __exit__(self, _kind: object, _value: object, _traceback: object) -> None:
if self._descriptor is None:
return
try:
fcntl.flock(self._descriptor, fcntl.LOCK_UN)
finally:
os.close(self._descriptor)
self._descriptor = None
def _run_docker(
docker_binary: Path,
docker_host: str,
docker_config: Path,
arguments: list[str],
*,
check: bool,
timeout: int,
) -> subprocess.CompletedProcess[bytes]:
_validate_docker_authority(docker_binary, docker_host, docker_config)
return subprocess.run(
[str(docker_binary), "--host", docker_host, "--config", str(docker_config), *arguments],
check=check,
capture_output=True,
timeout=timeout,
env=_docker_environment(),
)
def _local_image_reference_config_id(
docker_binary: Path,
docker_host: str,
docker_config: Path,
image_reference: str,
) -> str | None:
inventory = _run_docker(
docker_binary,
docker_host,
docker_config,
["image", "ls", "--digests", "--no-trunc", "--format", "{{json .}}", IMAGE_REPOSITORY],
check=True,
timeout=60,
)
_require(len(inventory.stdout) <= MAX_DOCKER_INSPECTION_BYTES, "local image inventory output is too large")
match = IMAGE_REFERENCE.fullmatch(image_reference)
_require(match is not None, "local image reference is invalid")
expected_digest = f"sha256:{match.group(1)}"
matching_ids: set[str] = set()
for raw_line in inventory.stdout.splitlines():
if not raw_line.strip():
continue
try:
row = _load_docker_inspection(raw_line)
except PackageError as exc:
raise PackageError("local image inventory output is invalid") from exc
_require(isinstance(row, dict), "local image inventory row is invalid")
if row.get("Repository") != IMAGE_REPOSITORY or row.get("Digest") != expected_digest:
continue
config_id = row.get("ID")
_require(
isinstance(config_id, str)
and config_id.startswith("sha256:")
and _is_sha256(config_id.removeprefix("sha256:")),
"local image inventory config digest is invalid",
)
matching_ids.add(config_id)
_require(len(matching_ids) <= 1, "local image inventory is ambiguous")
if not matching_ids:
return None
inventory_config_id = next(iter(matching_ids))
result = _run_docker(
docker_binary,
docker_host,
docker_config,
["image", "inspect", image_reference],
check=True,
timeout=60,
)
inspected = _load_docker_inspection(result.stdout)
_require(isinstance(inspected, list) and len(inspected) == 1, "local image reference inspection is not exact")
_require(isinstance(inspected[0], dict), "local image reference inspection payload is invalid")
config_id = inspected[0].get("Id")
_require(
isinstance(config_id, str)
and config_id.startswith("sha256:")
and _is_sha256(config_id.removeprefix("sha256:")),
"local image reference config digest is invalid",
)
_require(config_id == inventory_config_id, "local image inventory and inspection disagree")
return config_id
def _remove_new_image_reference(
docker_binary: Path,
docker_host: str,
docker_config: Path,
image_reference: str,
expected_config_id: str,
) -> None:
observed_config_id = _local_image_reference_config_id(
docker_binary,
docker_host,
docker_config,
image_reference,
)
if observed_config_id is None:
return
_require(observed_config_id == expected_config_id, "new registry image reference changed before cleanup")
_run_docker(
docker_binary,
docker_host,
docker_config,
["image", "rm", image_reference],
check=True,
timeout=120,
)
_require(
_local_image_reference_config_id(docker_binary, docker_host, docker_config, image_reference) is None,
"new registry image reference cleanup did not pass",
)
def inspect_registry_image(
image_input: dict[str, Any],
image_reference: str,
*,
expected_revision: str,
expected_input_sha256: str,
expected_config_digest: str,
docker_binary: Path,
docker_host: str,
docker_config: Path,
cleanup_new_reference: bool = False,
) -> dict[str, Any]:
"""Pull and inspect one immutable Artifact Registry image without starting it."""
validate_image_input(image_input)
_require(
IMAGE_REFERENCE.fullmatch(image_reference) is not None,
"inspection image must be the exact staging repository at a sha256 digest",
)
_require(_is_revision(expected_revision), "expected reviewed revision is invalid")
_require(_is_sha256(expected_input_sha256), "expected reviewed input SHA-256 is invalid")
_require(
isinstance(expected_config_digest, str)
and expected_config_digest.startswith("sha256:")
and _is_sha256(expected_config_digest.removeprefix("sha256:")),
"expected reviewed config digest is invalid",
)
_require(
image_input["runtime"]["teleo_git_head"] == expected_revision,
"image input differs from the expected reviewed revision",
)
_require(
image_input["input_sha256"] == expected_input_sha256,
"image input differs from the expected reviewed input SHA-256",
)
_validate_docker_authority(docker_binary, docker_host, docker_config)
prestate_observed = False
pulled_config_id: str | None = None
inspection: dict[str, Any] | None = None
primary_error: BaseException | None = None
with _DockerCacheLock(docker_config):
try:
if cleanup_new_reference:
_require(
_local_image_reference_config_id(docker_binary, docker_host, docker_config, image_reference) is None,
"exact registry image reference must be absent before inspection cleanup",
)
prestate_observed = True
_run_docker(
docker_binary,
docker_host,
docker_config,
["pull", "--quiet", "--platform", PLATFORM, image_reference],
check=True,
timeout=600,
)
pulled_config_id = _local_image_reference_config_id(
docker_binary,
docker_host,
docker_config,
image_reference,
)
_require(pulled_config_id is not None, "pulled registry image reference is missing")
result = _run_docker(
docker_binary,
docker_host,
docker_config,
["image", "inspect", image_reference],
check=True,
timeout=60,
)
inspected = _load_docker_inspection(result.stdout)
_require(isinstance(inspected, list) and len(inspected) == 1, "registry image inspection is not exact")
_require(isinstance(inspected[0], dict), "registry image inspection payload is invalid")
inspection = build_image_inspection(image_input, image_reference, inspected[0])
_require(
inspection["config_digest"] == expected_config_digest == pulled_config_id,
"inspected image differs from the expected reviewed config digest",
)
except (OSError, subprocess.SubprocessError, PackageError) as exc:
primary_error = exc
finally:
try:
if cleanup_new_reference and prestate_observed:
if pulled_config_id is None:
pulled_config_id = _local_image_reference_config_id(
docker_binary,
docker_host,
docker_config,
image_reference,
)
if pulled_config_id is not None:
_remove_new_image_reference(
docker_binary,
docker_host,
docker_config,
image_reference,
pulled_config_id,
)
except (OSError, subprocess.SubprocessError, PackageError) as cleanup_error:
raise PackageError("immutable registry image cleanup failed") from cleanup_error
if primary_error is not None:
raise PackageError("immutable registry image inspection failed") from primary_error
_require(inspection is not None, "immutable registry image inspection failed")
return inspection
def build_release_descriptor(
image_input: dict[str, Any],
inspection: dict[str, Any],
) -> dict[str, Any]:
validate_image_input(image_input)
validate_image_inspection(inspection, image_input=image_input)
image_reference = inspection["reference"]
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,
"config_digest": inspection["config_digest"],
"platform": PLATFORM,
"labels": labels,
"input_sha256": image_input["input_sha256"],
},
"inspection": inspection,
"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", "inspection", "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", "config_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")
inspection = value.get("inspection")
_require(isinstance(inspection, dict), "release image inspection is missing")
validate_image_inspection(inspection, image_input=image_input)
_require(inspection.get("reference") == reference, "release image differs from its inspection")
_require(image.get("config_digest") == inspection.get("config_digest"), "release config digest drifted")
_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 _fsync_path(path: Path, *, directory: bool = False) -> None:
flags = os.O_RDONLY
if directory and hasattr(os, "O_DIRECTORY"):
flags |= os.O_DIRECTORY
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
descriptor = os.open(path, flags)
try:
mode = os.fstat(descriptor).st_mode
if directory and not stat.S_ISDIR(mode):
raise OSError(errno.ENOTDIR, "fsync target is not a directory", str(path))
if not directory and not stat.S_ISREG(mode):
raise OSError(errno.EINVAL, "fsync target is not a regular file", str(path))
os.fsync(descriptor)
finally:
os.close(descriptor)
def _rename_noreplace(source: Path, destination: Path) -> None:
"""Atomically publish one directory without replacing an existing path."""
library = ctypes.CDLL(None, use_errno=True)
source_bytes = os.fsencode(source)
destination_bytes = os.fsencode(destination)
if sys.platform.startswith("linux"):
try:
rename = library.renameat2
except AttributeError as exc:
raise OSError(errno.ENOSYS, "renameat2 is unavailable") from exc
rename.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint]
rename.restype = ctypes.c_int
result = rename(-100, source_bytes, -100, destination_bytes, 1)
elif sys.platform == "darwin":
try:
rename = library.renamex_np
except AttributeError as exc:
raise OSError(errno.ENOSYS, "renamex_np is unavailable") from exc
rename.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_uint]
rename.restype = ctypes.c_int
result = rename(source_bytes, destination_bytes, 0x00000004)
else:
raise OSError(errno.ENOSYS, "atomic no-replace rename is unsupported")
if result != 0:
error = ctypes.get_errno()
if error == errno.EEXIST:
raise FileExistsError(error, "finalize output bundle already exists", str(destination))
raise OSError(error, os.strerror(error), str(destination))
def _remove_published_bundle(output_bundle: Path, expected_identity: tuple[int, int]) -> None:
try:
observed = output_bundle.stat(follow_symlinks=False)
except OSError as exc:
raise PackageError("finalize bundle rollback cannot identify published output") from exc
_require(
stat.S_ISDIR(observed.st_mode) and (observed.st_dev, observed.st_ino) == expected_identity,
"finalize bundle rollback refused to remove a replaced output",
)
try:
shutil.rmtree(output_bundle)
_fsync_path(output_bundle.parent, directory=True)
except OSError as exc:
raise PackageError("finalize bundle rollback failed") from exc
_require(
not output_bundle.exists() and not output_bundle.is_symlink(),
"finalize bundle rollback did not pass",
)
def _remove_staged_bundle(staged_bundle: Path, expected_identity: tuple[int, int]) -> None:
try:
observed = staged_bundle.stat(follow_symlinks=False)
except OSError as exc:
raise PackageError("finalize temporary cleanup cannot identify staged output") from exc
_require(
stat.S_ISDIR(observed.st_mode) and (observed.st_dev, observed.st_ino) == expected_identity,
"finalize temporary cleanup refused to remove a replaced output",
)
try:
shutil.rmtree(staged_bundle)
_fsync_path(staged_bundle.parent, directory=True)
except OSError as exc:
raise PackageError("finalize temporary cleanup failed") from exc
_require(
not staged_bundle.exists() and not staged_bundle.is_symlink(),
"finalize temporary cleanup did not pass",
)
def publish_finalize_bundle(output_bundle: Path, release: dict[str, Any]) -> None:
"""Publish release.json and the unit through one directory rename."""
validate_release_descriptor(release)
_require(not output_bundle.exists() and not output_bundle.is_symlink(), "finalize output bundle already exists")
try:
release_payload = json.dumps(release, allow_nan=False, indent=2, sort_keys=True) + "\n"
except (TypeError, ValueError) as exc:
raise PackageError("cannot encode finalized release JSON") from exc
unit_payload = render_systemd_unit(release)
staged_bundle = output_bundle.with_name(f".{output_bundle.name}.{uuid.uuid4().hex}.tmp")
staged_created = False
published = False
staged_identity: tuple[int, int] | None = None
try:
_require(
output_bundle.is_absolute()
and output_bundle.parent.is_dir()
and not output_bundle.parent.is_symlink(),
"finalize output parent must be an existing safe absolute directory",
)
staged_bundle.mkdir(mode=0o700)
staged_created = True
staged_stat = staged_bundle.stat(follow_symlinks=False)
staged_identity = (staged_stat.st_dev, staged_stat.st_ino)
staged_release = staged_bundle / "release.json"
staged_unit = staged_bundle / SERVICE
staged_release.write_text(release_payload, encoding="utf-8")
staged_release.chmod(0o644)
staged_unit.write_text(unit_payload, encoding="utf-8")
staged_unit.chmod(0o644)
validate_release_descriptor(load_json(staged_release, "staged release descriptor"))
_require(staged_unit.read_text(encoding="utf-8") == render_systemd_unit(release), "staged unit drifted")
_fsync_path(staged_release)
_fsync_path(staged_unit)
staged_bundle.chmod(0o755)
_fsync_path(staged_bundle, directory=True)
_rename_noreplace(staged_bundle, output_bundle)
published = True
_fsync_path(output_bundle.parent, directory=True)
except OSError as exc:
if published:
_require(staged_identity is not None, "finalize bundle rollback identity is missing")
_remove_published_bundle(output_bundle, staged_identity)
raise PackageError("cannot publish finalized bundle") from exc
finally:
if staged_created and (staged_bundle.exists() or staged_bundle.is_symlink()):
_require(staged_identity is not None, "finalize temporary cleanup identity is missing")
_remove_staged_bundle(staged_bundle, staged_identity)
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("--expected-revision", required=True)
finalize.add_argument("--expected-input-sha256", required=True)
finalize.add_argument("--expected-config-digest", required=True)
finalize.add_argument("--docker-binary", type=Path, required=True)
finalize.add_argument("--docker-host", required=True)
finalize.add_argument("--docker-config", type=Path, required=True)
finalize.add_argument("--output-bundle", 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")
inspection = inspect_registry_image(
image_input,
args.image_reference,
expected_revision=args.expected_revision,
expected_input_sha256=args.expected_input_sha256,
expected_config_digest=args.expected_config_digest,
docker_binary=args.docker_binary,
docker_host=args.docker_host,
docker_config=args.docker_config,
cleanup_new_reference=False,
)
result = build_release_descriptor(image_input, inspection)
publish_finalize_bundle(args.output_bundle, result)
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())
Reference in a new issue View git blame Copy permalink