#!/usr/bin/env python3 """Publish only the frozen MCP Hub connector-ingestion image by digest.""" from __future__ import annotations import argparse import hashlib import json import os import re import stat import subprocess import sys import tempfile from collections.abc import Callable from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Any FIXED_REPOSITORY = "jens/itworx_mcp_hub" FIXED_IMAGE_REPOSITORY = "gitea.example.com/jens/itworx-mcp-hub/connector-ingestion" FIXED_SOURCE_URI = "https://gitea.example.com/jens/itworx-mcp-hub" FIXED_SERVICE = "connector-ingestion" FIXED_CONTAINER = "gitea-deploy-control" FIXED_DOCKER_BINARY = "/usr/bin/docker" FIXED_DOCKER_CONTEXT = "default" FIXED_DOCKER_SOCKET = "unix:///var/run/docker.sock" FIXED_DOCKER_ROOT = "/var/lib/docker" FIXED_DOCKER_HOST_NAME = "Tower" FIXED_DOCKER_VERSION = "27.5.1" FIXED_BUILDX_VERSION = "v0.20.1" FIXED_BUILDER_NAME = "itworx-mcp-hub-ingestion" FIXED_BUILDKIT_IMAGE = ( "moby/buildkit:v0.33.0@sha256:6c2fa84a6b61ccd72899dde4239f8d5717f05f9a8ca6f3cad185fb1a95a94de3" ) FIXED_BUILDKIT_CONTAINER = "buildx_buildkit_itworx-mcp-hub-ingestion0" FIXED_BUILDKIT_CONFIG_PATH = Path("/root/.docker/buildx/buildkitd.default.toml") FIXED_BUILDKIT_ENTRYPOINT = ["/usr/bin/buildkitd-entrypoint"] FIXED_BUILDKIT_COMMAND = ["--allow-insecure-entitlement=network.host"] FIXED_BUILDKIT_STATE_VOLUME = f"{FIXED_BUILDKIT_CONTAINER}_state" FIXED_BUILDKIT_STATE_PATH = "/var/lib/buildkit" FIXED_VOLUME_NAME = "itworx-controller-hub-releases" FIXED_VOLUME_ROOT = Path("/var/lib/docker/volumes/itworx-controller-hub-releases/_data") FIXED_RELEASES_ROOT = FIXED_VOLUME_ROOT / "releases" FIXED_STATE_ROOT = Path("/mnt/cache/appdata/gitea-deploy-runners/state") ATTESTATION_SCHEMA = "itworx.mcp-hub-ingestion-freeze-attestation/v1" RECEIPT_SCHEMA = "itworx.mcp-hub-ingestion-publication/v1" MAX_METADATA_BYTES = 1024 * 1024 COMMAND_TIMEOUT_SECONDS = 30 BUILDER_BOOTSTRAP_TIMEOUT_SECONDS = 300 BUILD_TIMEOUT_SECONDS = 1800 OCI_LABELS = ( ("org.opencontainers.image.source", FIXED_SOURCE_URI), ("org.opencontainers.image.revision", "revision"), ("org.opencontainers.image.source-tree", "tree"), ("org.opencontainers.image.version", "id"), ("org.opencontainers.image.created", "builtAt"), ("org.opencontainers.image.source-date-epoch", "epoch"), ) ALLOWED_MANIFEST_MEDIA_TYPES = { "application/vnd.docker.distribution.manifest.v2+json", "application/vnd.oci.image.manifest.v1+json", } EXPECTED_VOLUME_LABELS = { "itworx.managed-by": "deployment-controller", "itworx.purpose": "mcp-hub-protected-release", } DOCKER_INFO_FORMAT = "{{.ID}}|{{.DockerRootDir}}|{{.ServerVersion}}|{{.Name}}" IMAGE_INSPECT_FORMAT = "|".join( [ "{{.Id}}", "{{.Config.User}}", *(f'{{{{index .Config.Labels "{name}"}}}}' for name, _ in OCI_LABELS), ] ) PINNED_BUILDKIT_IMAGE_FIELDS = ( "Id", "Config.Env", "Config.Entrypoint", "Config.User", "Config.WorkingDir", "Config.Labels", ) PINNED_BUILDKIT_IMAGE_FORMAT = "\n".join( f"{{{{json .{field}}}}}" for field in PINNED_BUILDKIT_IMAGE_FIELDS ) BUILDKIT_CONTAINER_FIELDS = ( "Image", "Config.Image", "Config.Env", "Config.Entrypoint", "Config.Cmd", "Config.User", "Config.WorkingDir", "Config.Labels", "State.Running", "HostConfig.Privileged", "HostConfig.Init", "HostConfig.RestartPolicy.Name", "HostConfig.RestartPolicy.MaximumRetryCount", "HostConfig.NetworkMode", "HostConfig.UsernsMode", "HostConfig.CgroupParent", "HostConfig.CgroupnsMode", "HostConfig.Mounts", "Mounts", "HostConfig.Binds", "HostConfig.Devices", "HostConfig.DeviceRequests", "HostConfig.CapAdd", "HostConfig.CapDrop", "HostConfig.SecurityOpt", "HostConfig.PortBindings", "HostConfig.PublishAllPorts", "HostConfig.ReadonlyRootfs", "HostConfig.PidMode", "HostConfig.IpcMode", "HostConfig.UTSMode", "HostConfig.AutoRemove", "HostConfig.Runtime", "HostConfig.Memory", "HostConfig.MemorySwap", "HostConfig.CpuShares", "HostConfig.CpuQuota", "HostConfig.CpuPeriod", "HostConfig.CpusetCpus", "HostConfig.CpusetMems", "HostConfig.OomKillDisable", ) BUILDKIT_CONTAINER_FORMAT = "\n".join( f"{{{{json .{field}}}}}" for field in BUILDKIT_CONTAINER_FIELDS ) class PublishError(RuntimeError): """A fixed publication precondition or attestation failed.""" @dataclass(frozen=True) class FrozenSource: repository: str revision: str build: dict[str, str] source_date_epoch: str freeze_sha256: str controller_source_path: str Runner = Callable[..., subprocess.CompletedProcess[str]] _CONTROLLER_ATTESTATION_PROGRAM = r""" import hashlib, json, sys from pathlib import Path sys.path.insert(0, "/opt/gitea-deploy") import deploy as controller import prepare_mcp_hub_candidate as candidate repository = "jens/itworx_mcp_hub" revision = sys.argv[1] project, repo = candidate._fixed_target(repository, revision) record = candidate._load_freeze_record(revision) build, epoch = candidate._validate_freeze_record(record, repository, revision, repo) source = controller.MCP_HUB_RELEASES_ROOT / revision if not controller.exact_source_matches_revision(repo, revision, source): raise RuntimeError("staged source mismatch") controller._mcp_hub_verify_release_tree_metadata(source) controller._mcp_hub_require_trusted_state_root() freeze_path = candidate._freeze_directory(revision, create=False) / candidate.FREEZE_FILENAME raw = freeze_path.read_bytes() if json.loads(raw) != record: raise RuntimeError("freeze changed") result = { "schema": "itworx.mcp-hub-ingestion-freeze-attestation/v1", "repository": repository, "revision": revision, "build": build, "sourceDateEpoch": epoch, "freezeSha256": hashlib.sha256(raw).hexdigest(), "sourcePath": source.as_posix(), } print(json.dumps(result, sort_keys=True, separators=(",", ":"))) """ def _docker(*arguments: str) -> list[str]: return [FIXED_DOCKER_BINARY, "--context", FIXED_DOCKER_CONTEXT, *arguments] def _run( runner: Runner, command: list[str], *, cwd: Path | None = None, timeout: int = COMMAND_TIMEOUT_SECONDS, stage: str, ) -> str: try: result = runner(command, cwd=cwd, timeout=timeout) except (OSError, subprocess.SubprocessError): raise PublishError(f"{stage} command failed") from None if result.returncode != 0 or not isinstance(result.stdout, str): raise PublishError(f"{stage} command failed") return result.stdout.strip() def _docker_runner( command: list[str], *, cwd: Path | None, timeout: int ) -> subprocess.CompletedProcess[str]: return subprocess.run( # noqa: S603 -- command is fixed by this module; callers inject only the test runner. command, cwd=cwd, timeout=timeout, check=False, capture_output=True, text=True, encoding="utf-8", errors="replace", ) def _validate_revision(revision: str) -> None: if re.fullmatch(r"[0-9a-f]{40}", revision) is None: raise PublishError("revision must be a full lowercase Git SHA") def _validate_digest(value: object) -> str: digest = str(value or "") if re.fullmatch(r"sha256:[0-9a-f]{64}", digest) is None: raise PublishError("image digest is invalid") return digest def _validate_frozen_attestation(document: object, revision: str) -> FrozenSource: _validate_revision(revision) required = { "schema", "repository", "revision", "build", "sourceDateEpoch", "freezeSha256", "sourcePath", } if not isinstance(document, dict) or set(document) != required: raise PublishError("official source-freeze attestation has an invalid shape") build = document.get("build") build_fields = {"id", "repository", "revision", "tree", "treeDigest", "builtAt"} if ( document.get("schema") != ATTESTATION_SCHEMA or document.get("repository") != FIXED_REPOSITORY or document.get("revision") != revision or not isinstance(build, dict) or set(build) != build_fields or any(not isinstance(value, str) for value in build.values()) ): raise PublishError("official source-freeze attestation does not match the fixed target") tree = build["tree"] epoch = document.get("sourceDateEpoch") if re.fullmatch(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z", build["builtAt"]) is None: raise PublishError("official source-freeze build time is invalid") from None try: datetime.strptime(build["builtAt"], "%Y-%m-%dT%H:%M:%S.%fZ") except ValueError: raise PublishError("official source-freeze build time is invalid") from None if ( build["id"] != revision[:12] or build["repository"] != FIXED_REPOSITORY or build["revision"] != revision or re.fullmatch(r"[0-9a-f]{40}", tree) is None or build["treeDigest"] != hashlib.sha256(f"git-tree:{tree}".encode("ascii")).hexdigest() or not isinstance(epoch, str) or re.fullmatch(r"[1-9][0-9]{0,11}", epoch) is None or re.fullmatch(r"[0-9a-f]{64}", str(document.get("freezeSha256") or "")) is None or document.get("sourcePath") != f"/controller-hub-releases/releases/{revision}" ): raise PublishError("official source-freeze metadata is invalid") return FrozenSource( repository=FIXED_REPOSITORY, revision=revision, build=dict(build), source_date_epoch=epoch, freeze_sha256=document["freezeSha256"], controller_source_path=document["sourcePath"], ) def _attest_official_freeze(revision: str, runner: Runner) -> FrozenSource: _validate_revision(revision) command = _docker( "exec", "--user", "0", FIXED_CONTAINER, "python3", "-c", _CONTROLLER_ATTESTATION_PROGRAM, revision, ) raw = _run(runner, command, stage="official source-freeze attestation") if len(raw) > 16 * 1024: raise PublishError("official source-freeze attestation is oversized") try: document = json.loads(raw) except (ValueError, TypeError): raise PublishError("official source-freeze attestation is invalid") from None return _validate_frozen_attestation(document, revision) def _parse_engine_identity(value: str, *, source: str) -> tuple[str, str, str, str]: parts = value.split("|") if ( len(parts) != 4 or re.fullmatch(r"[A-Za-z0-9:_-]{16,128}", parts[0]) is None or parts[1] != FIXED_DOCKER_ROOT or parts[2] != FIXED_DOCKER_VERSION or parts[3] != FIXED_DOCKER_HOST_NAME ): raise PublishError(f"{source} Docker engine identity is not the fixed Tower engine") return parts[0], parts[1], parts[2], parts[3] def _preflight_engine(runner: Runner) -> tuple[str, str]: selected = _run(runner, [FIXED_DOCKER_BINARY, "context", "show"], stage="Docker context") endpoint = _run( runner, [ FIXED_DOCKER_BINARY, "context", "inspect", FIXED_DOCKER_CONTEXT, "--format", '{{(index .Endpoints "docker").Host}}', ], stage="Docker endpoint inspection", ) if selected != FIXED_DOCKER_CONTEXT or endpoint != FIXED_DOCKER_SOCKET: raise PublishError("Docker CLI is not bound to the fixed local Tower socket") host = _parse_engine_identity( _run( runner, _docker("info", "--format", DOCKER_INFO_FORMAT), stage="host Docker identity inspection", ), source="host", ) controller = _parse_engine_identity( _run( runner, _docker( "exec", "--user", "0", FIXED_CONTAINER, "docker", "--host", FIXED_DOCKER_SOCKET, "info", "--format", DOCKER_INFO_FORMAT, ), stage="controller Docker identity inspection", ), source="controller-socket", ) if host != controller: raise PublishError("host and controller Docker socket identities differ") inspected = _run( runner, _docker("inspect", "--format", "{{.Id}}|{{.State.Running}}", FIXED_CONTAINER), stage="controller container inspection", ).split("|") if ( len(inspected) != 2 or re.fullmatch(r"[0-9a-f]{64}", inspected[0]) is None or inspected[1] != "true" ): raise PublishError("fixed deployment controller is not running") buildx_version = _run(runner, _docker("buildx", "version"), stage="Buildx availability") if re.search(rf"(?:^|\s){re.escape(FIXED_BUILDX_VERSION)}(?:\s|$)", buildx_version) is None: raise PublishError("Buildx version differs from the fixed Tower toolchain") return host[0], inspected[0] def _builder_rows(runner: Runner) -> set[tuple[str, str]]: raw = _run( runner, _docker("buildx", "ls", "--format", "{{.Name}}|{{.DriverEndpoint}}"), stage="buildx builder inventory", ) rows: set[tuple[str, str]] = set() for line in raw.splitlines(): parts = line.strip().split("|", 1) if len(parts) != 2 or not parts[0] or not parts[1]: raise PublishError("buildx builder inventory is invalid") rows.add((parts[0], parts[1])) return rows def _json_inspection_record(raw: str, fields: tuple[str, ...], *, stage: str) -> dict[str, Any]: lines = raw.splitlines() if len(lines) != len(fields): raise PublishError(f"{stage} returned an invalid fixed record") try: values = [json.loads(line) for line in lines] except (ValueError, TypeError): raise PublishError(f"{stage} returned an invalid fixed record") from None return dict(zip(fields, values, strict=True)) def _empty_host_config(value: Any) -> bool: return value in (None, [], {}, "") def _mount_extras_are_empty(mount: dict[str, Any], allowed: set[str]) -> bool: return all(key in allowed or value in (None, "", False, [], {}) for key, value in mount.items()) def _verify_dedicated_builder_container(runner: Runner, *, require_running: bool) -> None: image_raw = _run( runner, _docker( "image", "inspect", "--format", PINNED_BUILDKIT_IMAGE_FORMAT, FIXED_BUILDKIT_IMAGE, ), stage="pinned BuildKit image configuration inspection", ) image = _json_inspection_record( image_raw, PINNED_BUILDKIT_IMAGE_FIELDS, stage="pinned BuildKit image configuration inspection", ) image_id = image["Id"] image_env = image["Config.Env"] image_entrypoint = image["Config.Entrypoint"] image_user = image["Config.User"] image_working_dir = image["Config.WorkingDir"] image_labels = image["Config.Labels"] if ( not isinstance(image_id, str) or re.fullmatch(r"sha256:[0-9a-f]{64}", image_id) is None or image_entrypoint != FIXED_BUILDKIT_ENTRYPOINT or not isinstance(image_env, list) or not all(isinstance(item, str) for item in image_env) or image_user is not None and not isinstance(image_user, str) or image_working_dir is not None and not isinstance(image_working_dir, str) or image_labels is not None and not isinstance(image_labels, dict) ): raise PublishError("pinned BuildKit image configuration is outside its fixed contract") container_raw = _run( runner, _docker( "inspect", "--format", BUILDKIT_CONTAINER_FORMAT, FIXED_BUILDKIT_CONTAINER, ), stage="dedicated BuildKit container inspection", ) container = _json_inspection_record( container_raw, BUILDKIT_CONTAINER_FIELDS, stage="dedicated BuildKit container inspection", ) config_matches_image = all( container[container_field] == image[image_field] for container_field, image_field in ( ("Config.Env", "Config.Env"), ("Config.Entrypoint", "Config.Entrypoint"), ("Config.User", "Config.User"), ("Config.WorkingDir", "Config.WorkingDir"), ("Config.Labels", "Config.Labels"), ) ) mounts = container["HostConfig.Mounts"] actual_mounts = container["Mounts"] expected_host_mount = { "Type": "volume", "Source": FIXED_BUILDKIT_STATE_VOLUME, "Target": FIXED_BUILDKIT_STATE_PATH, } expected_actual_mount = { "Type": "volume", "Name": FIXED_BUILDKIT_STATE_VOLUME, "Source": f"{FIXED_DOCKER_ROOT}/volumes/{FIXED_BUILDKIT_STATE_VOLUME}/_data", "Destination": FIXED_BUILDKIT_STATE_PATH, "Driver": "local", "Mode": "", "RW": True, "Propagation": "", } mounts_match = ( isinstance(mounts, list) and len(mounts) == 1 and isinstance(mounts[0], dict) and all(mounts[0].get(key) == value for key, value in expected_host_mount.items()) # Docker's Mount.ReadOnly is tagged json:",omitempty". Buildx leaves it # unset for the writable state volume, so the daemon omits the key. # Accept only omission or the explicit JSON boolean false; reject null, # numbers, strings, and true rather than treating them as equivalent. and ("ReadOnly" not in mounts[0] or mounts[0]["ReadOnly"] is False) and _mount_extras_are_empty(mounts[0], set(expected_host_mount) | {"ReadOnly"}) and isinstance(actual_mounts, list) and len(actual_mounts) == 1 and isinstance(actual_mounts[0], dict) and all(actual_mounts[0].get(key) == value for key, value in expected_actual_mount.items()) and _mount_extras_are_empty(actual_mounts[0], set(expected_actual_mount)) ) empty_config_fields = ( "HostConfig.Binds", "HostConfig.Devices", "HostConfig.DeviceRequests", "HostConfig.CapAdd", "HostConfig.CapDrop", "HostConfig.SecurityOpt", "HostConfig.PortBindings", ) no_extra_host_config = all( _empty_host_config(container[field]) for field in empty_config_fields ) zero_resource_limits = all( container[field] in (None, 0, "") for field in ( "HostConfig.Memory", "HostConfig.MemorySwap", "HostConfig.CpuShares", "HostConfig.CpuQuota", "HostConfig.CpuPeriod", "HostConfig.CpusetCpus", "HostConfig.CpusetMems", ) ) if ( container["Image"] != image_id or container["Config.Image"] != FIXED_BUILDKIT_IMAGE or not config_matches_image or container["Config.Cmd"] != FIXED_BUILDKIT_COMMAND or not mounts_match or container["HostConfig.Privileged"] is not True or container["HostConfig.Init"] is not True or container["HostConfig.RestartPolicy.Name"] != "unless-stopped" or container["HostConfig.RestartPolicy.MaximumRetryCount"] != 0 or container["HostConfig.NetworkMode"] != "default" or container["HostConfig.UsernsMode"] not in (None, "") or container["HostConfig.CgroupParent"] != "/docker/buildx" or container["HostConfig.CgroupnsMode"] != "private" or not no_extra_host_config or container["HostConfig.PublishAllPorts"] is not False or container["HostConfig.ReadonlyRootfs"] is not False or any( container[field] not in (None, "") for field in ( "HostConfig.PidMode", "HostConfig.UTSMode", ) ) or container["HostConfig.IpcMode"] != "private" or container["HostConfig.Runtime"] != "runc" or container["HostConfig.AutoRemove"] is not False or not zero_resource_limits or container["HostConfig.OomKillDisable"] not in (None, False) or (require_running and container["State.Running"] is not True) ): raise PublishError("dedicated BuildKit container runtime configuration is unsafe") def _verify_buildkit_default_config_absent() -> None: path = FIXED_BUILDKIT_CONFIG_PATH current = Path(path.anchor) for part in path.parts[1:]: current = current / part try: metadata = current.lstat() except FileNotFoundError: continue except OSError: raise PublishError("default BuildKit config path cannot be verified") from None if stat.S_ISLNK(metadata.st_mode): raise PublishError("default BuildKit config path contains a symlink") try: path.lstat() except FileNotFoundError: return except OSError: raise PublishError("default BuildKit config path cannot be verified") from None raise PublishError("default BuildKit config would alter the pinned daemon command") def _inspect_dedicated_builder( runner: Runner, *, require_running: bool, validate_container: bool = False ) -> None: rows = _builder_rows(runner) expected_rows = { (FIXED_BUILDER_NAME, "docker-container"), (f"{FIXED_BUILDER_NAME}0", FIXED_DOCKER_SOCKET), } own_rows = {row for row in rows if row[0].startswith(FIXED_BUILDER_NAME)} if own_rows != expected_rows: raise PublishError("dedicated BuildKit builder does not match its fixed local contract") details = _run( runner, _docker("buildx", "inspect", FIXED_BUILDER_NAME), stage="dedicated builder inspection", ) names = re.findall(r"(?m)^Name:\s*(\S+)\s*$", details) options = re.search(r"(?m)^Driver Options:\s*(.*?)\s*$", details) if ( names != [FIXED_BUILDER_NAME, f"{FIXED_BUILDER_NAME}0"] or not re.search(r"(?m)^Driver:\s+docker-container\s*$", details) or not re.search(rf"(?m)^Endpoint:\s+{re.escape(FIXED_DOCKER_SOCKET)}\s*$", details) or options is None or options.group(1) != f'image="{FIXED_BUILDKIT_IMAGE}"' ): raise PublishError("dedicated BuildKit builder is not bound to the fixed local engine") if require_running: version_lines = re.findall(r"(?m)^BuildKit(?: version)?:\s*(\S+)\s*$", details) if version_lines != ["v0.33.0"] or not re.search(r"(?m)^Status:\s+running\s*$", details): raise PublishError("dedicated BuildKit version differs from the pinned release") if validate_container or require_running: _verify_dedicated_builder_container(runner, require_running=require_running) def _ensure_dedicated_builder(runner: Runner, frozen: FrozenSource) -> None: checked_frozen = _validate_frozen_attestation( { "schema": ATTESTATION_SCHEMA, "repository": frozen.repository, "revision": frozen.revision, "build": frozen.build, "sourceDateEpoch": frozen.source_date_epoch, "freezeSha256": frozen.freeze_sha256, "sourcePath": frozen.controller_source_path, }, frozen.revision, ) if checked_frozen != frozen: raise PublishError("a validated source freeze is required before builder setup") _verify_buildkit_default_config_absent() rows = _builder_rows(runner) builder_registered = any( name in {FIXED_BUILDER_NAME, f"{FIXED_BUILDER_NAME}0"} for name, _ in rows ) containers = _run( runner, _docker("container", "ls", "--all", "--format", "{{.Names}}"), stage="dedicated builder container inventory", ).splitlines() has_reserved_container = FIXED_BUILDKIT_CONTAINER in containers if not builder_registered and has_reserved_container: raise PublishError("reserved BuildKit container name is occupied without its builder") if builder_registered: # Validate the configured pin before starting any existing builder. _inspect_dedicated_builder( runner, require_running=False, validate_container=has_reserved_container, ) else: command = _docker( "buildx", "create", "--name", FIXED_BUILDER_NAME, "--driver", "docker-container", "--driver-opt", f"image={FIXED_BUILDKIT_IMAGE}", FIXED_DOCKER_CONTEXT, ) _run(runner, command, stage="fixed dedicated BuildKit builder creation") _inspect_dedicated_builder(runner, require_running=False) _run( runner, _docker("buildx", "inspect", "--bootstrap", FIXED_BUILDER_NAME), timeout=BUILDER_BOOTSTRAP_TIMEOUT_SECONDS, stage="pinned dedicated BuildKit bootstrap", ) _inspect_dedicated_builder(runner, require_running=True, validate_container=True) def _verify_release_volume(runner: Runner) -> None: raw = _run( runner, _docker( "volume", "inspect", "--format", "{{json .}}", FIXED_VOLUME_NAME, ), stage="fixed release-volume inspection", ) try: document = json.loads(raw) except (ValueError, TypeError): raise PublishError("fixed release volume metadata is invalid") from None if isinstance(document, list): if len(document) != 1: raise PublishError("fixed release volume metadata is ambiguous") document = document[0] if ( not isinstance(document, dict) or document.get("Name") != FIXED_VOLUME_NAME or document.get("Driver") != "local" or document.get("Scope") != "local" or document.get("Mountpoint") != FIXED_VOLUME_ROOT.as_posix() or document.get("Options") not in (None, {}) or document.get("Labels") != EXPECTED_VOLUME_LABELS ): raise PublishError("fixed release volume differs from its reviewed contract") def _require_no_acl(path: Path) -> None: if os.name == "nt": return listxattr = getattr(os, "listxattr", None) if not callable(listxattr): raise PublishError("filesystem ACLs cannot be checked") try: names = listxattr(path, follow_symlinks=False) except OSError: raise PublishError("filesystem ACLs cannot be checked") from None if any(name in {"system.posix_acl_access", "system.posix_acl_default"} for name in names): raise PublishError("source path has an unreviewed POSIX ACL") def _verify_local_release_root(path: Path, revision: str) -> None: expected = FIXED_RELEASES_ROOT / revision if path != expected or path.is_symlink(): raise PublishError("staged source path is not the fixed release generation") try: info = path.lstat() volume = FIXED_VOLUME_ROOT.lstat() except OSError: raise PublishError("staged source generation is unavailable") from None releases_info = FIXED_RELEASES_ROOT.lstat() if ( path.is_symlink() or not stat.S_ISDIR(info.st_mode) or not stat.S_ISDIR(volume.st_mode) or not stat.S_ISDIR(releases_info.st_mode) or FIXED_VOLUME_ROOT.is_symlink() or FIXED_RELEASES_ROOT.is_symlink() ): raise PublishError("staged source generation metadata is unsafe") if os.name != "nt" and ( info.st_uid != 0 or stat.S_IMODE(info.st_mode) != 0o555 or volume.st_uid != 0 or stat.S_IMODE(volume.st_mode) & 0o022 or releases_info.st_uid != 0 or stat.S_IMODE(releases_info.st_mode) & 0o022 ): raise PublishError("staged source generation metadata is unsafe") _require_no_acl(path) _require_no_acl(FIXED_VOLUME_ROOT) _require_no_acl(FIXED_RELEASES_ROOT) _require_no_acl(path / "infrastructure") dockerfile = path / "infrastructure" / "images" / "python-service" / "Dockerfile" try: dockerfile_info = dockerfile.lstat() except OSError: raise PublishError("fixed ingestion Dockerfile is unavailable") from None if dockerfile.is_symlink() or not stat.S_ISREG(dockerfile_info.st_mode): raise PublishError("fixed ingestion Dockerfile metadata is unsafe") if os.name != "nt" and ( dockerfile_info.st_uid != 0 or stat.S_IMODE(dockerfile_info.st_mode) not in (0o444, 0o555) or dockerfile_info.st_nlink != 1 ): raise PublishError("fixed ingestion Dockerfile metadata is unsafe") _require_no_acl(dockerfile) def _parse_build_metadata(path: Path) -> tuple[str, str]: try: info = path.lstat() if ( path.is_symlink() or not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or (os.name != "nt" and info.st_uid != 0) or info.st_size <= 0 or info.st_size > MAX_METADATA_BYTES ): raise PublishError("build metadata file metadata is unsafe") _require_no_acl(path) metadata = json.loads(path.read_text(encoding="utf-8")) except PublishError: raise except (OSError, ValueError, TypeError): raise PublishError("build metadata file is invalid") from None if not isinstance(metadata, dict): raise PublishError("build metadata file is invalid") return ( _validate_digest(metadata.get("containerimage.digest")), _validate_digest(metadata.get("containerimage.config.digest")), ) def _parse_manifest_descriptor(raw: str, expected_digest: str) -> None: if len(raw) > 16 * 1024: raise PublishError("registry manifest descriptor is oversized") try: descriptor = json.loads(raw) except (ValueError, TypeError): raise PublishError("registry manifest descriptor is invalid") from None if not isinstance(descriptor, dict) or set(descriptor) != {"digest", "mediaType", "size"}: raise PublishError("registry manifest descriptor has an invalid shape") if ( descriptor.get("digest") != expected_digest or descriptor.get("mediaType") not in ALLOWED_MANIFEST_MEDIA_TYPES or isinstance(descriptor.get("size"), bool) or not isinstance(descriptor.get("size"), int) or descriptor["size"] <= 0 ): raise PublishError("registry manifest digest does not match the frozen build") def _validate_image_inspection(raw: str, frozen: FrozenSource, config_digest: str) -> None: fields = raw.split("|") if len(fields) != 2 + len(OCI_LABELS): raise PublishError("registry image inspection has an invalid shape") if fields[0] != config_digest or fields[1] != "10001:10001": raise PublishError("registry image identity or non-root user does not match the build") expected = [ FIXED_SOURCE_URI, frozen.revision, frozen.build["tree"], frozen.build["id"], frozen.build["builtAt"], frozen.source_date_epoch, ] if fields[2:] != expected: raise PublishError("registry image provenance labels do not match the freeze") def _safe_private_dir(path: Path, *, create: bool) -> None: if os.name != "nt" and (not hasattr(os, "geteuid") or os.geteuid() != 0): raise PublishError("private publication receipts require host root on POSIX") fixed_parent = FIXED_STATE_ROOT.parent for trusted in (fixed_parent, FIXED_STATE_ROOT): try: info = trusted.lstat() except OSError: raise PublishError("fixed controller state root is unavailable") from None if ( trusted.is_symlink() or not stat.S_ISDIR(info.st_mode) or (os.name != "nt" and (info.st_uid != 0 or stat.S_IMODE(info.st_mode) & 0o022)) ): raise PublishError("fixed controller state root is unsafe") _require_no_acl(trusted) if not path.is_relative_to(FIXED_STATE_ROOT): raise PublishError("publication receipt path is outside fixed controller state") relative = path.relative_to(FIXED_STATE_ROOT) current = FIXED_STATE_ROOT for part in relative.parts: current = current / part if create: try: current.mkdir(mode=0o700) except FileExistsError: pass except OSError: raise PublishError("private publication directory could not be prepared") from None elif not current.exists() and not current.is_symlink(): raise PublishError("private publication directory is unavailable") try: info = current.lstat() except OSError: raise PublishError("private publication directory is unavailable") from None if current.is_symlink() or not stat.S_ISDIR(info.st_mode): raise PublishError("private publication directory metadata is unsafe") if os.name != "nt" and (info.st_uid != 0 or stat.S_IMODE(info.st_mode) != 0o700): raise PublishError("private publication directory metadata is unsafe") _require_no_acl(current) def _require_publication_slot(state_root: Path, revision: str) -> None: if state_root != FIXED_STATE_ROOT: raise PublishError("publication receipt root is not fixed") _safe_private_dir(state_root, create=False) current = state_root for part in ("mcp-hub-ingestion-publications", revision): current = current / part if not current.exists() and not current.is_symlink(): break try: info = current.lstat() except OSError: raise PublishError("publication receipt path is unsafe") from None if current.is_symlink() or not stat.S_ISDIR(info.st_mode): raise PublishError("publication receipt path is unsafe") if os.name != "nt" and (info.st_uid != 0 or stat.S_IMODE(info.st_mode) != 0o700): raise PublishError("publication receipt path is unsafe") _require_no_acl(current) receipt = state_root / "mcp-hub-ingestion-publications" / revision / "publication.json" if receipt.exists() or receipt.is_symlink(): raise PublishError("publication receipt already exists and is never overwritten") def _write_receipt(path: Path, payload: dict[str, Any]) -> str: _safe_private_dir(path.parent, create=True) if path.exists() or path.is_symlink(): raise PublishError("publication receipt already exists and is never overwritten") receipt_hash = hashlib.sha256( (json.dumps(payload, sort_keys=True, separators=(",", ":")) + "\n").encode("utf-8") ).hexdigest() envelope = {"receipt": payload, "receiptSha256": receipt_hash} serialized = (json.dumps(envelope, sort_keys=True, separators=(",", ":")) + "\n").encode( "utf-8" ) temporary = path.with_name(f".{path.name}.{os.urandom(8).hex()}.tmp") created = False try: descriptor = os.open( temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), 0o600, ) created = True try: if os.name != "nt": os.fchmod(descriptor, 0o600) with os.fdopen(descriptor, "wb", closefd=False) as stream: stream.write(serialized) stream.flush() os.fsync(stream.fileno()) finally: os.close(descriptor) info = temporary.lstat() if temporary.is_symlink() or not stat.S_ISREG(info.st_mode): raise PublishError("private publication receipt temporary file is unsafe") if os.name != "nt" and ( info.st_uid != 0 or info.st_nlink != 1 or stat.S_IMODE(info.st_mode) != 0o600 ): raise PublishError("private publication receipt temporary file is unsafe") _require_no_acl(temporary) os.link(temporary, path, follow_symlinks=False) temporary.unlink() created = False if os.name != "nt": directory_fd = os.open(path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) try: os.fsync(directory_fd) finally: os.close(directory_fd) except PublishError: raise except FileExistsError: raise PublishError("publication receipt already exists and is never overwritten") from None except OSError: raise PublishError("private publication receipt could not be written") from None finally: if created: temporary.unlink(missing_ok=True) return receipt_hash def _publish_frozen_revision( revision: str, *, frozen: FrozenSource, docker_runner: Runner, source_root: Path, state_root: Path, script_path: Path, before_receipt: Callable[[], None] | None = None, ) -> dict[str, str]: """Build and publish only the fixed service from an attested immutable freeze.""" _validate_revision(revision) if frozen.revision != revision or frozen.repository != FIXED_REPOSITORY: raise PublishError("frozen source does not match the fixed target") checked_frozen = _validate_frozen_attestation( { "schema": ATTESTATION_SCHEMA, "repository": frozen.repository, "revision": frozen.revision, "build": frozen.build, "sourceDateEpoch": frozen.source_date_epoch, "freezeSha256": frozen.freeze_sha256, "sourcePath": frozen.controller_source_path, }, revision, ) if checked_frozen != frozen: raise PublishError("frozen source attestation is invalid") expected_state_root = FIXED_STATE_ROOT if os.name != "nt" else state_root if state_root != expected_state_root: raise PublishError("publication receipt root is not the fixed controller state root") _verify_local_release_root(source_root, revision) expected_script = source_root / "scripts" / "publish_mcp_hub_ingestion.py" if script_path.resolve() != expected_script.resolve() or script_path.is_symlink(): raise PublishError("publisher script is not from the frozen source tree") if not script_path.is_file(): raise PublishError("publisher script is not present in the frozen source tree") if os.name != "nt": script_info = script_path.lstat() if ( script_path.is_symlink() or not stat.S_ISREG(script_info.st_mode) or script_info.st_uid != 0 or stat.S_IMODE(script_info.st_mode) not in (0o444, 0o555) or script_info.st_nlink != 1 ): raise PublishError("publisher script metadata is unsafe") _require_no_acl(script_path) _require_publication_slot(state_root, revision) receipt_path = state_root / "mcp-hub-ingestion-publications" / revision / "publication.json" _inspect_dedicated_builder(docker_runner, require_running=True) with tempfile.TemporaryDirectory(prefix="mcp-hub-ingestion-") as temporary: metadata_path = Path(temporary) / "build-metadata.json" output = ( "type=image," f"name={FIXED_IMAGE_REPOSITORY}," "push=true,push-by-digest=true,name-canonical=true" ) command = _docker( "buildx", "build", "--builder", FIXED_BUILDER_NAME, "--platform", "linux/amd64", "--pull", "--provenance=false", "--file", "infrastructure/images/python-service/Dockerfile", "--metadata-file", str(metadata_path), "--output", output, ) for key, value in ( ("SERVICE_NAME", FIXED_SERVICE), ("GIT_REVISION", revision), ("GIT_SOURCE_TREE", frozen.build["tree"]), ("IMAGE_VERSION", frozen.build["id"]), ("SOURCE_DATE_EPOCH", frozen.source_date_epoch), ("BUILD_DATE", frozen.build["builtAt"]), ): command.extend(("--build-arg", f"{key}={value}")) command.append(".") _run( docker_runner, command, cwd=source_root, timeout=BUILD_TIMEOUT_SECONDS, stage="fixed ingestion image build and digest publication", ) _inspect_dedicated_builder(docker_runner, require_running=True) manifest_digest, config_digest = _parse_build_metadata(metadata_path) image_ref = f"{FIXED_IMAGE_REPOSITORY}@{manifest_digest}" descriptor = _run( docker_runner, _docker( "buildx", "imagetools", "inspect", "--format", "{{json .Manifest}}", image_ref, ), stage="registry manifest readback", ) _parse_manifest_descriptor(descriptor, manifest_digest) _run( docker_runner, _docker("pull", "--quiet", image_ref), timeout=BUILD_TIMEOUT_SECONDS, stage="immutable registry image pull", ) inspected = _run( docker_runner, _docker("image", "inspect", "--format", IMAGE_INSPECT_FORMAT, image_ref), stage="immutable image provenance inspection", ) _validate_image_inspection(inspected, frozen, config_digest) receipt = { "schema": RECEIPT_SCHEMA, "repository": FIXED_REPOSITORY, "service": FIXED_SERVICE, "revision": revision, "build": dict(frozen.build), "sourceDateEpoch": frozen.source_date_epoch, "sourceFreezeSha256": frozen.freeze_sha256, "imageRepository": FIXED_IMAGE_REPOSITORY, "imageRef": image_ref, "manifestDigest": manifest_digest, "imageId": config_digest, "platform": "linux/amd64", "registryReadbackVerified": True, } if before_receipt is not None: before_receipt() receipt_hash = _write_receipt(receipt_path, receipt) return { "status": "published", "revision": revision, "imageRef": image_ref, "imageId": config_digest, "receiptSha256": receipt_hash, } def _reject_docker_environment(environ: dict[str, str]) -> None: forbidden = { "DOCKER_HOST", "DOCKER_CONTEXT", "DOCKER_CONFIG", "DOCKER_TLS_VERIFY", "DOCKER_CERT_PATH", "BUILDX_BUILDER", "BUILDKIT_HOST", "DOCKER_AUTH_CONFIG", "BUILDX_CONFIG", } if any(key in environ for key in forbidden): raise PublishError("Docker endpoint, builder, or credential-path overrides are not allowed") if environ.get("HOME") != "/root": raise PublishError("publisher must use the existing host-root Docker credential context") def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("revision", help="exact frozen protected-main revision") args = parser.parse_args(argv) try: _validate_revision(args.revision) _reject_docker_environment(dict(os.environ)) if os.name != "posix" or not hasattr(os, "geteuid") or os.geteuid() != 0: raise PublishError("publisher requires POSIX host root") runner = _docker_runner initial_engine, initial_container = _preflight_engine(runner) _verify_release_volume(runner) frozen = _attest_official_freeze(args.revision, runner) source_root = FIXED_RELEASES_ROOT / args.revision _verify_local_release_root(source_root, args.revision) _require_publication_slot(FIXED_STATE_ROOT, args.revision) _ensure_dedicated_builder(runner, frozen) def reattest_before_receipt() -> None: final_engine, final_container = _preflight_engine(runner) _verify_release_volume(runner) final_frozen = _attest_official_freeze(args.revision, runner) if (initial_engine, initial_container) != ( final_engine, final_container, ) or final_frozen != frozen: raise PublishError( "controller, Docker engine, or source freeze changed during publication" ) result = _publish_frozen_revision( args.revision, frozen=frozen, docker_runner=runner, source_root=source_root, state_root=FIXED_STATE_ROOT, script_path=Path(__file__), before_receipt=reattest_before_receipt, ) print(json.dumps(result, sort_keys=True)) return 0 except PublishError as exc: print(f"publication denied: {exc}", file=sys.stderr) return 2 except (OSError, ValueError): print("publication denied: fixed publication precondition failed", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())