yuqei/ai-contract-platform
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix(release): close owned infrastructure trust gaps

Browse Source
This commit is contained in:
yuqei
2026-09-27 16:31:46 +08:00
parent 0b2364b502
commit 2b8ac38a81
11 changed files with 1131 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+1
View File
@@ -10,5 +10,6 @@ pythonpath = [
"yuqei-ai-platform/services/ai-platform-api/src",
"yuqei-ai-platform/services/worker/src",
"yuqei-ai-sdk-python/src",
"yuqei-infra/tests",
]
addopts = "-q --import-mode=importlib"
yuqei-infra/deploy/release-package.py
+15
View File
@@ -1173,6 +1173,7 @@ def _validate_source_bound_tool(
def _validate_infrastructure_bundle(
*,
package_root: Path,
source_manifest: Mapping[str, Any],
artifact_root: Path,
source_set_path: Path,
bundle_path: Path,
@@ -1188,6 +1189,7 @@ def _validate_infrastructure_bundle(
try:
rebuilt = module.build_bundle(
artifact_root=artifact_root,
source_manifest=source_manifest,
source_set=source_set,
application_manifest=installable_manifest,
)
@@ -1826,6 +1828,10 @@ def validate_release_package(
third_party_distribution_scope: str | None = None
if package_schema == PACKAGE_SCHEMA_V2:
infrastructure_module = _load_infrastructure_bundle_module()
owned_infrastructure_module = infrastructure_module._load_owned_validator()
object_storage_probe_module = (
infrastructure_module._load_object_storage_probe_executor()
)
third_party_module = _load_third_party_materials_module()
snapshot_module = _load_snapshot_restore_planner_module()
for relative, loaded in (
@@ -1833,6 +1839,14 @@ def validate_release_package(
"yuqei-infra/scripts/build-infrastructure-image-bundle.py",
Path(infrastructure_module.__file__),
),
(
"yuqei-infra/scripts/validate-owned-infrastructure-build.py",
Path(owned_infrastructure_module.__file__),
),
(
"yuqei-infra/scripts/run-object-storage-runtime-probe.py",
Path(object_storage_probe_module.__file__),
),
(
"yuqei-infra/scripts/build-third-party-materials.py",
Path(third_party_module.__file__),
@@ -1875,6 +1889,7 @@ def validate_release_package(
offline_image_overrides,
) = _validate_infrastructure_bundle(
package_root=root,
source_manifest=release_manifest,
artifact_root=infrastructure_source_path.parent,
source_set_path=infrastructure_source_path,
bundle_path=infrastructure_bundle_path,
yuqei-infra/scripts/assemble-build05-v2-package.py
+1
View File
@@ -553,6 +553,7 @@ def _preflight(args: argparse.Namespace) -> dict[str, Any]:
try:
recomputed_bundle = infrastructure_module.build_bundle(
artifact_root=infrastructure_root,
source_repo=repo,
source_set=source_set,
application_manifest=installable,
)
yuqei-infra/scripts/build-infrastructure-image-bundle.py
+281 -10
View File
@@ -10,6 +10,7 @@ manifest is emitted.
from __future__ import annotations
import argparse
import base64
import gzip
import hashlib
import importlib.util
@@ -24,6 +25,8 @@ from datetime import datetime
from pathlib import Path, PurePosixPath
from typing import Any, Mapping
from cryptography.exceptions import InvalidSignature
SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
@@ -41,6 +44,7 @@ from linux_release_common import ( # noqa: E402
)
OWNED_VALIDATOR_PATH = SCRIPT_DIR / "validate-owned-infrastructure-build.py"
OBJECT_STORAGE_PROBE_EXECUTOR_PATH = SCRIPT_DIR / "run-object-storage-runtime-probe.py"
SOURCE_SET_SCHEMA = "yuqei.infrastructure-image-source-set.v2"
BUNDLE_SCHEMA = "yuqei.infrastructure-image-bundle.v2"
@@ -65,6 +69,8 @@ LEGACY_CONFIG_MEMBER_RE = re.compile(r"^([0-9a-f]{64})\.json$")
LEGACY_LAYER_MEMBER_RE = re.compile(r"^([0-9a-f]{64})/layer\.tar$")
CONTENT_ADDRESSED_ARCHIVE_PROFILE = "content_addressed_gzip_sha256_v1"
LEGACY_ARCHIVE_PROFILE = "legacy_docker_save_uncompressed_v1"
OBJECT_STORAGE_PROBE_SCHEMA = "yuqei.object-storage-runtime-probe.v1"
OBJECT_STORAGE_PROBE_DOMAIN = b"yuqei-object-storage-runtime-probe-v1\x00"
APPLICATION_SPECS = (
("contract-web", "CONTRACT_WEB_IMAGE"),
@@ -241,6 +247,22 @@ def _load_owned_validator() -> Any:
return module
def _load_object_storage_probe_executor() -> Any:
spec = importlib.util.spec_from_file_location(
"yuqei_object_storage_runtime_probe_executor",
OBJECT_STORAGE_PROBE_EXECUTOR_PATH,
)
if spec is None or spec.loader is None:
raise BundleValidationError("cannot load object-storage runtime probe executor")
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
try:
spec.loader.exec_module(module)
except Exception as exc: # pragma: no cover - import environment failure
raise BundleValidationError("cannot load object-storage runtime probe executor") from exc
return module
def _digest(value: object, *, label: str) -> str:
if not isinstance(value, str) or not DIGEST_RE.fullmatch(value):
raise BundleValidationError(f"{label} must be a sha256 digest")
@@ -1001,7 +1023,9 @@ def _validate_owned_postgres_build(
value: object,
*,
expected_commit: str,
) -> tuple[dict[str, Any], Path]:
source_repo: Path | None,
source_manifest: Mapping[str, Any] | None = None,
) -> tuple[dict[str, Any], Path, Any]:
if not isinstance(value, Mapping) or set(value) != {
"candidate_root",
"evidence_root",
@@ -1074,21 +1098,116 @@ def _validate_owned_postgres_build(
if policy_record["path"] not in observed or receipt_record["path"] not in observed:
raise BundleValidationError("owned PostgreSQL policy or receipt is not inventoried")
validator = _load_owned_validator()
try:
summary = validator.validate_artifact(
candidate_root=candidate_root,
policy_path=policy_path,
receipt_path=receipt_path,
input_root=input_root,
evidence_root=evidence_root,
expected_commit=expected_commit,
required_candidate_paths = (
validator.REQUIRED_POLICY_PATH,
validator.REQUIRED_SECURITY_POLICY_PATH,
*validator.REQUIRED_DOCKERFILES.values(),
)
manifest_records: dict[str, Mapping[str, Any]] = {}
manifest_candidate: Mapping[str, Any] | None = None
if source_repo is None:
if not isinstance(source_manifest, Mapping):
raise BundleValidationError("trusted source binding is required")
supplied = source_manifest.get("manifest_payload_sha256")
unsigned_manifest = dict(source_manifest)
unsigned_manifest.pop("manifest_payload_sha256", None)
if not isinstance(supplied, str) or supplied != canonical_sha256(unsigned_manifest):
raise BundleValidationError("trusted source manifest payload hash drifted")
manifest_candidate = source_manifest.get("candidate")
files = source_manifest.get("files")
if (
source_manifest.get("schema_version") != "yuqei.release-source-manifest.v1"
or not isinstance(manifest_candidate, Mapping)
or manifest_candidate.get("commit") != expected_commit
or manifest_candidate.get("worktree_was_clean") is not True
or not isinstance(files, Mapping)
or not isinstance(files.get("items"), list)
):
raise BundleValidationError("trusted source manifest candidate binding drifted")
manifest_records = {
item["path"]: item
for item in files["items"]
if isinstance(item, Mapping) and isinstance(item.get("path"), str)
}
for relative in required_candidate_paths:
staged = _regular_file(
candidate_root.joinpath(*PurePosixPath(relative).parts),
label=f"staged owned PostgreSQL candidate file {relative}",
)
try:
staged_bytes = staged.read_bytes()
if source_repo is not None:
committed, _object_id = validator._git_file(
source_repo, expected_commit, relative
)
else:
record = manifest_records.get(relative)
if (
not isinstance(record, Mapping)
or record.get("sha256") != hashlib.sha256(staged_bytes).hexdigest()
or record.get("size") != len(staged_bytes)
or record.get("git_blob_oid")
!= hashlib.sha1(
b"blob " + str(len(staged_bytes)).encode("ascii") + b"\x00" + staged_bytes,
usedforsecurity=False,
).hexdigest()
):
raise BundleValidationError(
f"staged owned PostgreSQL candidate file differs from trusted source manifest: {relative}"
)
committed = staged_bytes
except OSError as exc:
raise BundleValidationError(
f"cannot read staged owned PostgreSQL candidate file {relative}"
) from exc
if staged_bytes != committed:
raise BundleValidationError(
f"staged owned PostgreSQL candidate file differs from Git commit: {relative}"
)
try:
if source_repo is not None:
summary = validator.validate(
repo=source_repo,
policy_path=source_repo.joinpath(
*PurePosixPath(validator.REQUIRED_POLICY_PATH).parts
),
receipt_path=receipt_path,
input_root=input_root,
evidence_root=evidence_root,
expected_commit=expected_commit,
)
else:
summary = validator.validate_artifact(
candidate_root=candidate_root,
policy_path=policy_path,
receipt_path=receipt_path,
input_root=input_root,
evidence_root=evidence_root,
expected_commit=expected_commit,
)
except Exception as exc:
if exc.__class__.__name__ == "OwnedInfrastructureValidationError":
raise BundleValidationError(
f"owned PostgreSQL build validation failed: {exc}"
) from exc
raise
if manifest_candidate is not None:
timestamp = manifest_candidate.get("commit_timestamp")
tree = manifest_candidate.get("tree")
try:
manifest_epoch = int(
parse_rfc3339(timestamp, label="trusted source manifest timestamp").timestamp()
)
except Exception as exc:
raise BundleValidationError("trusted source manifest timestamp is invalid") from exc
if summary.get("candidate_tree") != tree or summary.get("source_date_epoch") != manifest_epoch:
raise BundleValidationError("owned PostgreSQL receipt differs from trusted source manifest identity")
policy_payload = load_json(policy_path, label="candidate owned PostgreSQL policy")
trust_policy = validator._validate_policy(policy_payload)
attestation_public_key, _raw_key = validator._load_attestation_public_key(
trust_policy=trust_policy,
evidence_root=evidence_root,
)
return (
{
"candidate_root": candidate_relative,
@@ -1100,16 +1219,154 @@ def _validate_owned_postgres_build(
"validation": summary,
},
evidence_root,
attestation_public_key,
)
def _validate_object_storage_runtime_probe(
root: Path,
value: object,
*,
candidate: Mapping[str, Any],
images: list[dict[str, Any]],
public_key: Any,
) -> dict[str, Any]:
path, record = _file_record(root, value, label="object-storage runtime probe receipt")
envelope = load_json(path, label="object-storage runtime probe receipt")
if set(envelope) != {
"schema_version",
"payload",
"signature_algorithm",
"signature_base64",
}:
raise BundleValidationError("object-storage runtime probe receipt fields drifted")
if (
envelope.get("schema_version") != OBJECT_STORAGE_PROBE_SCHEMA
or envelope.get("signature_algorithm") != "Ed25519"
or not isinstance(envelope.get("payload"), Mapping)
):
raise BundleValidationError("object-storage runtime probe receipt identity drifted")
probe = envelope["payload"]
if set(probe) != {"candidate", "controls", "executor", "images", "platform"}:
raise BundleValidationError("object-storage runtime probe payload fields drifted")
if (
probe.get("candidate") != dict(candidate)
or probe.get("platform") != TARGET["platform"]
or probe.get("executor") != "yuqei-controlled-object-storage-runtime-probe-v1"
or probe.get("controls")
!= {
"network_mode": "isolated",
"real_container_runtime": True,
"synthetic": False,
}
):
raise BundleValidationError("object-storage runtime probe candidate or controls drifted")
object_images = [item for item in images if item["role"] in {"minio", "minio-client"}]
raw_images = probe.get("images")
if not isinstance(raw_images, list) or len(raw_images) != 2:
raise BundleValidationError("object-storage runtime probe must cover server and rc client")
normalized_images: list[dict[str, Any]] = []
for raw, expected_role, bound in zip(
raw_images, ("minio", "minio-client"), object_images, strict=True
):
if not isinstance(raw, Mapping) or set(raw) != {
"archive_sha256",
"checks",
"config_digest",
"manifest_digest",
"role",
}:
raise BundleValidationError("object-storage runtime probe image fields drifted")
if (
raw.get("role") != expected_role
or raw.get("manifest_digest") != bound["manifest_digest"]
or raw.get("config_digest") != bound["config_digest"]
or raw.get("archive_sha256") != bound["docker_archive"]["sha256"]
):
raise BundleValidationError(
f"object-storage runtime probe image identity drifted: {expected_role}"
)
checks = raw.get("checks")
expected_check_keys = (
{
"bin_sh_executable",
"container_started",
"data_volume_owner_uid",
"data_volume_writable",
"effective_uid",
"entrypoint_invoked",
"healthcheck_passed",
"read_only_rootfs",
"tmpfs_paths",
}
if expected_role == "minio"
else {
"bin_sh_executable",
"config_from_stdin",
"container_started",
"effective_uid",
"entrypoint_invoked",
"mirror_roundtrip_verified",
"rc_config_uses_tmpfs",
"rc_executable",
"read_only_rootfs",
"tmpfs_paths",
"volume_permissions_verified",
}
)
if not isinstance(checks, Mapping) or set(checks) != expected_check_keys:
raise BundleValidationError(
f"object-storage runtime probe checks drifted: {expected_role}"
)
uid = checks.get("effective_uid")
required_true = expected_check_keys - {
"effective_uid",
"data_volume_owner_uid",
"tmpfs_paths",
}
if (
not isinstance(uid, int)
or isinstance(uid, bool)
or uid <= 0
or any(checks.get(name) is not True for name in required_true)
or checks.get("tmpfs_paths") != ["/tmp"]
or (expected_role == "minio" and checks.get("data_volume_owner_uid") != uid)
):
raise BundleValidationError(
f"object-storage runtime probe did not pass: {expected_role}"
)
normalized_images.append(dict(raw))
try:
signature = base64.b64decode(envelope.get("signature_base64"), validate=True)
public_key.verify(
signature,
OBJECT_STORAGE_PROBE_DOMAIN + canonical_json_bytes(probe),
)
except (TypeError, ValueError, InvalidSignature) as exc:
raise BundleValidationError("object-storage runtime probe signature is invalid") from exc
return {
**record,
"schema_version": OBJECT_STORAGE_PROBE_SCHEMA,
"candidate": dict(candidate),
"images": normalized_images,
"passed": True,
}
def build_bundle(
*,
artifact_root: Path,
source_repo: Path | None = None,
source_manifest: Mapping[str, Any] | None = None,
source_set: Mapping[str, Any],
application_manifest: Mapping[str, Any],
) -> dict[str, Any]:
root = _artifact_root(artifact_root)
if (source_repo is None) == (source_manifest is None):
raise BundleValidationError(
"exactly one trusted Git repository or frozen source manifest is required"
)
repo = _artifact_root(source_repo) if source_repo is not None else None
if source_set.get("schema_version") != SOURCE_SET_SCHEMA:
raise BundleValidationError("unsupported infrastructure source-set schema")
_embedded_hash(source_set, "source_set_payload_sha256", label="infrastructure source set")
@@ -1119,6 +1376,7 @@ def build_bundle(
"target",
"security_gate",
"owned_postgres_build",
"object_storage_runtime_probe",
"images",
"source_set_payload_sha256",
}:
@@ -1138,10 +1396,12 @@ def build_bundle(
source_set.get("security_gate"),
application_manifest=application_manifest,
)
owned_postgres, owned_evidence_root = _validate_owned_postgres_build(
owned_postgres, owned_evidence_root, owned_attestation_key = _validate_owned_postgres_build(
root,
source_set.get("owned_postgres_build"),
expected_commit=str(source_candidate.get("commit", "")),
source_repo=repo,
source_manifest=source_manifest,
)
raw_images = source_set.get("images")
@@ -1464,6 +1724,14 @@ def build_bundle(
f"{role} bundle does not reuse the byte-identical owned {field}"
)
object_storage_probe = _validate_object_storage_runtime_probe(
root,
source_set.get("object_storage_runtime_probe"),
candidate=source_candidate,
images=infrastructure_images,
public_key=owned_attestation_key,
)
if tuple(overrides) != ALL_IMAGE_KEYS or set(overrides) != set(ALL_IMAGE_KEYS):
raise BundleValidationError("offline override map does not cover exactly all eleven images")
security_gate.update(
@@ -1483,6 +1751,7 @@ def build_bundle(
"application_images": application_images,
"infrastructure_images": infrastructure_images,
"owned_postgres_build": owned_postgres,
"object_storage_runtime_probe": object_storage_probe,
"security_gate": security_gate,
"offline_compose_overrides": overrides,
"all_release_image_keys": list(ALL_IMAGE_KEYS),
@@ -1526,6 +1795,7 @@ def _write_exclusive(path: Path, payload: Mapping[str, Any]) -> None:
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--artifact-root", type=Path, required=True)
parser.add_argument("--source-repo", type=Path, required=True)
parser.add_argument("--source-set", type=Path, required=True)
parser.add_argument("--application-manifest", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
@@ -1537,6 +1807,7 @@ def main(argv: list[str] | None = None) -> int:
try:
payload = build_bundle(
artifact_root=args.artifact_root,
source_repo=args.source_repo,
source_set=load_json(args.source_set, label="infrastructure source set"),
application_manifest=load_json(
args.application_manifest, label="application installable manifest"
yuqei-infra/scripts/run-object-storage-runtime-probe.py
+310
View File
@@ -0,0 +1,310 @@
#!/usr/bin/env python3
"""Run the fail-closed Linux RustFS/rc candidate runtime probe and sign its receipt."""
from __future__ import annotations
import argparse
import base64
import hashlib
import json
import os
import platform
import re
import stat
import subprocess
import time
from pathlib import Path, PurePosixPath
from typing import Any, Callable, Mapping, Sequence
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
SCHEMA = "yuqei.object-storage-runtime-probe.v1"
EXECUTOR = "yuqei-controlled-object-storage-runtime-probe-v1"
DOMAIN = b"yuqei-object-storage-runtime-probe-v1\x00"
DOCKER = "/usr/bin/docker"
DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
SAFE_ENTRYPOINT_RE = re.compile(r"^/[A-Za-z0-9._/-]{1,255}$")
class ProbeError(RuntimeError):
pass
def canonical(value: object) -> bytes:
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()
def _read_json(path: Path) -> dict[str, Any]:
try:
value = json.loads(path.read_bytes())
except (OSError, ValueError) as exc:
raise ProbeError(f"invalid JSON input: {path}") from exc
if not isinstance(value, dict):
raise ProbeError(f"JSON input must be an object: {path}")
return value
def _under(root: Path, relative: object) -> Path:
if not isinstance(relative, str):
raise ProbeError("artifact path is invalid")
pure = PurePosixPath(relative)
if pure.is_absolute() or ".." in pure.parts:
raise ProbeError("artifact path escapes artifact root")
path = root.joinpath(*pure.parts).resolve(strict=True)
try:
path.relative_to(root)
except ValueError as exc:
raise ProbeError("artifact path escapes artifact root") from exc
return path
def _run_default(argv: Sequence[str], *, input_bytes: bytes | None = None) -> bytes:
completed = subprocess.run(
list(argv), input=input_bytes, capture_output=True, check=False, timeout=120
)
if completed.returncode != 0:
raise ProbeError(f"controlled Docker command failed: {argv[1]}")
return completed.stdout
def _inspect(run: Callable[..., bytes], reference: str) -> dict[str, Any]:
try:
values = json.loads(run((DOCKER, "image", "inspect", reference)))
except (ValueError, TypeError) as exc:
raise ProbeError("Docker image inspect returned invalid JSON") from exc
if not isinstance(values, list) or len(values) != 1 or not isinstance(values[0], dict):
raise ProbeError("Docker image inspect must return exactly one image")
return values[0]
def _image(source: Mapping[str, Any], role: str) -> Mapping[str, Any]:
images = source.get("images")
if not isinstance(images, list):
raise ProbeError("source set image inventory is missing")
matches = [item for item in images if isinstance(item, Mapping) and item.get("role") == role]
if len(matches) != 1:
raise ProbeError(f"source set must contain exactly one {role} image")
return matches[0]
def _load_key(path: Path) -> Ed25519PrivateKey:
if path.is_symlink():
raise ProbeError("builder signing key must not be a symbolic link")
metadata = path.stat()
if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1:
raise ProbeError("builder signing key must be a single-link regular file")
if os.name == "posix" and metadata.st_mode & 0o077:
raise ProbeError("builder signing key permissions are not private")
raw = path.read_bytes()
try:
if raw.startswith(b"-----BEGIN"):
key = serialization.load_pem_private_key(raw, password=None)
if not isinstance(key, Ed25519PrivateKey):
raise TypeError
return key
return Ed25519PrivateKey.from_private_bytes(raw)
except (TypeError, ValueError) as exc:
raise ProbeError("builder signing key is not an Ed25519 private key") from exc
def execute(
*,
artifact_root: Path,
source_set_path: Path,
output_path: Path,
private_key_path: Path,
run: Callable[..., bytes] = _run_default,
require_host: bool = True,
) -> dict[str, Any]:
if require_host and (
os.name != "posix"
or platform.system() != "Linux"
or platform.machine().lower() not in {"x86_64", "amd64"}
or not Path(DOCKER).is_file()
):
raise ProbeError("probe requires controlled linux/amd64 with /usr/bin/docker")
root = artifact_root.resolve(strict=True)
source = _read_json(source_set_path.resolve(strict=True))
supplied = source.get("source_set_payload_sha256")
unsigned = dict(source)
unsigned.pop("source_set_payload_sha256", None)
if supplied != hashlib.sha256(canonical(unsigned)).hexdigest():
raise ProbeError("source set payload hash drifted")
candidate = source.get("candidate")
if not isinstance(candidate, dict) or set(candidate) != {"commit", "tag"}:
raise ProbeError("source set candidate binding is invalid")
records = [_image(source, "minio"), _image(source, "minio-client")]
inspected: dict[str, dict[str, Any]] = {}
for record in records:
archive_record = record.get("docker_archive")
if not isinstance(archive_record, Mapping):
raise ProbeError("object-storage archive record is missing")
archive = _under(root, archive_record.get("path"))
raw = archive.read_bytes()
if (
archive_record.get("sha256") != hashlib.sha256(raw).hexdigest()
or archive_record.get("size_bytes") != len(raw)
):
raise ProbeError("object-storage archive bytes drifted")
run((DOCKER, "image", "load", "--input", str(archive)))
reference = record.get("expected_repo_tag")
if not isinstance(reference, str):
raise ProbeError("loaded image reference is invalid")
details = _inspect(run, reference)
expected_config = record.get("config_digest")
expected_manifest = record.get("manifest_digest")
repo_digests = details.get("RepoDigests")
if (
not isinstance(expected_config, str)
or not DIGEST_RE.fullmatch(expected_config)
or details.get("Id") != expected_config
or not isinstance(expected_manifest, str)
or not DIGEST_RE.fullmatch(expected_manifest)
or not isinstance(repo_digests, list)
or not any(str(item).endswith("@" + expected_manifest) for item in repo_digests)
):
raise ProbeError(f"loaded image identity drifted: {record['role']}")
inspected[str(record["role"])] = details
suffix = str(candidate["commit"])[:12]
network = f"yuqei-os-probe-{suffix}"
server = network + "-server"
client = network + "-client"
volume = network + "-data"
created: list[tuple[str, str]] = []
server_record, client_record = records
try:
run((DOCKER, "network", "create", "--internal", network)); created.append(("network", network))
run((DOCKER, "volume", "create", volume)); created.append(("volume", volume))
run((
DOCKER, "create", "--pull", "never", "--name", server, "--network", network, "--read-only",
"--tmpfs", "/tmp:rw,nosuid,nodev", "--mount", f"type=volume,source={volume},target=/data",
"--security-opt", "no-new-privileges:true", "-e", "MINIO_ROOT_USER=yuqei_probe",
"-e", "MINIO_ROOT_PASSWORD=yuqei_probe_password_123",
"--health-cmd", "test -f /data/.minio.sys/format.json",
"--health-interval", "2s", "--health-timeout", "2s",
"--health-retries", "30", "--health-start-period", "0s",
str(server_record["expected_repo_tag"]),
"server", "/data", "--console-address", ":9001",
)); created.append(("container", server))
run((DOCKER, "start", server))
server_state = None
for _attempt in range(30):
candidate_state = json.loads(
run((DOCKER, "container", "inspect", server))
)[0]
health = candidate_state.get("State", {}).get("Health", {}).get("Status")
if health == "healthy":
server_state = candidate_state
break
if health == "unhealthy" or candidate_state.get("State", {}).get("Running") is not True:
raise ProbeError("object-storage server failed its Docker healthcheck")
time.sleep(2)
if server_state is None:
raise ProbeError("object-storage server Docker healthcheck timed out")
server_ids = run((DOCKER, "exec", server, "/bin/sh", "-ceu", "id -u; stat -c %u /data; test -w /data; touch /data/.yuqei-probe; test -f /data/.minio.sys/format.json")).decode().splitlines()
uid, owner = (int(server_ids[0]), int(server_ids[1]))
client_config = (
f"http://{server}:9000\nyuqei_probe\nyuqei_probe_password_123\n"
).encode("utf-8")
entrypoint = ((inspected["minio-client"].get("Config") or {}).get("Entrypoint") or [None])[0]
if not isinstance(entrypoint, str) or not SAFE_ENTRYPOINT_RE.fullmatch(entrypoint):
raise ProbeError("rc image entrypoint is not a safe absolute executable")
client_script = (
"set -eu; IFS= read -r probe_endpoint; IFS= read -r probe_access; "
"IFS= read -r probe_secret; test -n \"$probe_endpoint\"; "
"test -n \"$probe_access\"; test -n \"$probe_secret\"; "
f"test -x {entrypoint}; {entrypoint} --version >/tmp/version; "
f"{entrypoint} alias set probe \"$probe_endpoint\" \"$probe_access\" \"$probe_secret\"; "
f"mkdir -p /tmp/src; printf candidate-bound >/tmp/src/marker; {entrypoint} mb --ignore-existing probe/probe; "
f"{entrypoint} mirror /tmp/src probe/probe; {entrypoint} cp probe/probe/marker /tmp/roundtrip; "
"test \"$(cat /tmp/roundtrip)\" = candidate-bound; test -r /probe-data/.yuqei-probe; "
"test ! -w /probe-data; id -u"
)
created.append(("container", client))
client_out = run((
DOCKER, "run", "--pull", "never", "--name", client, "--network", network, "--read-only",
"--tmpfs", "/tmp:rw,nosuid,nodev", "--security-opt", "no-new-privileges:true",
"--mount", f"type=volume,source={volume},target=/probe-data,readonly",
"-e", "MC_CONFIG_DIR=/tmp/.mc", "--entrypoint", "/bin/sh", "-i",
str(client_record["expected_repo_tag"]), "-ceu", client_script,
), input_bytes=client_config)
client_uid = int(client_out.decode().splitlines()[-1])
if uid <= 0 or owner != uid or client_uid <= 0 or server_state.get("State", {}).get("Running") is not True:
raise ProbeError("object-storage runtime identity or health checks failed")
payload = {
"candidate": dict(candidate),
"controls": {"network_mode": "isolated", "real_container_runtime": True, "synthetic": False},
"executor": EXECUTOR,
"images": [
{"role": "minio", "manifest_digest": server_record["manifest_digest"], "config_digest": server_record["config_digest"], "archive_sha256": server_record["docker_archive"]["sha256"], "checks": {"bin_sh_executable": True, "container_started": True, "data_volume_owner_uid": owner, "data_volume_writable": True, "effective_uid": uid, "entrypoint_invoked": True, "healthcheck_passed": True, "read_only_rootfs": True, "tmpfs_paths": ["/tmp"]}},
{"role": "minio-client", "manifest_digest": client_record["manifest_digest"], "config_digest": client_record["config_digest"], "archive_sha256": client_record["docker_archive"]["sha256"], "checks": {"bin_sh_executable": True, "config_from_stdin": True, "container_started": True, "effective_uid": client_uid, "entrypoint_invoked": True, "mirror_roundtrip_verified": True, "rc_config_uses_tmpfs": True, "rc_executable": True, "read_only_rootfs": True, "tmpfs_paths": ["/tmp"], "volume_permissions_verified": True}},
],
"platform": "linux/amd64",
}
signature = _load_key(private_key_path.absolute()).sign(DOMAIN + canonical(payload))
envelope = {"schema_version": SCHEMA, "payload": payload, "signature_algorithm": "Ed25519", "signature_base64": base64.b64encode(signature).decode("ascii")}
parent = output_path.parent.resolve(strict=True)
target = parent / output_path.name
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
try:
descriptor = os.open(target, flags, 0o600)
with os.fdopen(descriptor, "wb") as stream:
metadata = os.fstat(stream.fileno())
if (
not stat.S_ISREG(metadata.st_mode)
or metadata.st_nlink != 1
or (os.name == "posix" and metadata.st_mode & 0o077)
):
raise ProbeError(
"probe receipt output is not a private single-link regular file"
)
stream.write(canonical(envelope) + b"\n")
stream.flush()
os.fsync(stream.fileno())
if os.name == "posix":
directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
directory_descriptor = os.open(parent, directory_flags)
try:
os.fsync(directory_descriptor)
finally:
os.close(directory_descriptor)
except OSError as exc:
raise ProbeError("probe receipt output must be a new private regular file") from exc
return envelope
finally:
for kind, name in reversed(created):
command = (
(DOCKER, "rm", "--force", name)
if kind == "container"
else (DOCKER, kind, "rm", name)
)
try:
if run is _run_default:
subprocess.run(command, check=False, capture_output=True)
else:
run(command)
except Exception:
pass
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--artifact-root", type=Path, required=True)
parser.add_argument("--source-set", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
parser.add_argument("--builder-private-key", type=Path, required=True)
args = parser.parse_args()
try:
execute(artifact_root=args.artifact_root, source_set_path=args.source_set, output_path=args.output, private_key_path=args.builder_private_key)
except ProbeError as exc:
print(f"FAILED_CLOSED: {exc}", file=os.sys.stderr)
return 2
return 0
if __name__ == "__main__":
raise SystemExit(main())
yuqei-infra/scripts/validate-owned-infrastructure-build.py
+1 -1
View File
@@ -285,7 +285,7 @@ def _git_file(repo: Path, commit: str, relative_path: str) -> tuple[bytes, str]:
raise OwnedInfrastructureValidationError(
f"candidate path is not a regular Git blob: {relative_path}"
)
raw = _run_git(repo, "show", f"{commit}:{relative_path}", binary=True)
raw = _run_git(repo, "cat-file", "blob", f"{commit}:{relative_path}", binary=True)
assert isinstance(raw, bytes)
return raw, object_id
yuqei-infra/tests/owned_infrastructure_test_support.py
+86 -1
View File
@@ -72,7 +72,15 @@ def _git(repo: Path, *arguments: str, env: dict[str, str] | None = None) -> str:
def _git_record(repo: Path, commit: str, relative_path: str) -> dict[str, Any]:
raw = subprocess.run(
["git", "--no-replace-objects", "-C", str(repo), "show", f"{commit}:{relative_path}"],
[
"git",
"--no-replace-objects",
"-C",
str(repo),
"cat-file",
"blob",
f"{commit}:{relative_path}",
],
check=True,
capture_output=True,
).stdout
@@ -361,6 +369,8 @@ def prepare_owned_case(
if initialize_git:
_git(candidate_root, "init")
_git(candidate_root, "config", "core.autocrlf", "false")
_git(candidate_root, "config", "core.eol", "lf")
_git(candidate_root, "add", ".")
git_env = dict(os.environ)
git_env.update(
@@ -581,6 +591,7 @@ def prepare_owned_case(
"tree": tree,
"epoch": epoch,
"outputs": outputs,
"private_key": private_key,
"owned_binding": {
"candidate_root": candidate_root.relative_to(artifact_root).as_posix(),
"evidence_root": evidence_root.relative_to(artifact_root).as_posix(),
@@ -598,3 +609,77 @@ def rewrite_receipt(case: dict[str, Any]) -> None:
receipt = _with_hash(receipt, "receipt_payload_sha256")
case["receipt"] = receipt
write_json(case["receipt_path"], receipt)
def make_object_storage_probe(
artifact_root: Path,
*,
candidate: dict[str, str],
images: list[dict[str, Any]],
private_key: Ed25519PrivateKey,
domain: bytes,
schema: str,
) -> dict[str, Any]:
selected = [item for item in images if item["role"] in {"minio", "minio-client"}]
probe_images: list[dict[str, Any]] = []
for item in selected:
role = item["role"]
uid = 10001
checks = (
{
"bin_sh_executable": True,
"container_started": True,
"data_volume_owner_uid": uid,
"data_volume_writable": True,
"effective_uid": uid,
"entrypoint_invoked": True,
"healthcheck_passed": True,
"read_only_rootfs": True,
"tmpfs_paths": ["/tmp"],
}
if role == "minio"
else {
"bin_sh_executable": True,
"config_from_stdin": True,
"container_started": True,
"effective_uid": uid,
"entrypoint_invoked": True,
"mirror_roundtrip_verified": True,
"rc_config_uses_tmpfs": True,
"rc_executable": True,
"read_only_rootfs": True,
"tmpfs_paths": ["/tmp"],
"volume_permissions_verified": True,
}
)
probe_images.append(
{
"archive_sha256": item["docker_archive"]["sha256"],
"checks": checks,
"config_digest": item["config_digest"],
"manifest_digest": item["manifest_digest"],
"role": role,
}
)
payload = {
"candidate": dict(candidate),
"controls": {
"network_mode": "isolated",
"real_container_runtime": True,
"synthetic": False,
},
"executor": "yuqei-controlled-object-storage-runtime-probe-v1",
"images": probe_images,
"platform": "linux/amd64",
}
envelope = {
"payload": payload,
"schema_version": schema,
"signature_algorithm": "Ed25519",
"signature_base64": base64.b64encode(
private_key.sign(domain + canonical(payload))
).decode("ascii"),
}
path = artifact_root / "object-storage" / "runtime-probe.v1.json"
write_json(path, envelope)
return record(path, artifact_root)
yuqei-infra/tests/test_build05_v2_package_orchestrator.py
+10 -1
View File
@@ -63,7 +63,10 @@ def _reviewed_module_loader(inputs: dict[str, object]):
base_infrastructure = helpers.release_package._load_infrastructure_bundle_module()
original_infrastructure_root = Path(inputs["infrastructure_root"])
def fixture_bundle_adapter(*, artifact_root, source_set, application_manifest):
def fixture_bundle_adapter(
*, artifact_root, source_set, application_manifest, source_repo=None,
source_manifest=None,
):
# The shared release-package fixture intentionally validates a modern
# synthetic Docker archive and substitutes the separately validated
# application records. Re-run that adapter against its immutable raw
@@ -73,12 +76,18 @@ def _reviewed_module_loader(inputs: dict[str, object]):
root = original_infrastructure_root
return base_infrastructure.build_bundle(
artifact_root=root,
source_repo=source_repo,
source_manifest=source_manifest,
source_set=source_set,
application_manifest=application_manifest,
)
infrastructure = SimpleNamespace(
__file__=base_infrastructure.__file__,
_load_owned_validator=base_infrastructure._load_owned_validator,
_load_object_storage_probe_executor=(
base_infrastructure._load_object_storage_probe_executor
),
build_bundle=fixture_bundle_adapter,
)
yuqei-infra/tests/test_infrastructure_image_bundle.py
+129 -7
View File
@@ -5,14 +5,16 @@ import hashlib
import importlib.util
import io
import json
import subprocess
import sys
import tarfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Mapping
import pytest
from owned_infrastructure_test_support import prepare_owned_case
from owned_infrastructure_test_support import make_object_storage_probe, prepare_owned_case
MODULE_PATH = (
@@ -294,18 +296,14 @@ def _make_security_gate(root: Path) -> dict[str, object]:
def _fixture(root: Path) -> tuple[dict[str, object], dict[str, object]]:
commit = "1" * 40
tree = "2" * 40
epoch = 1_788_048_000
tag = "v2-test-1"
owned_case = prepare_owned_case(
root,
validator=owned_validator,
source_root=SOURCE_ROOT,
commit=commit,
tree=tree,
epoch=epoch,
initialize_git=True,
)
commit = owned_case["commit"]
security_gate = _make_security_gate(root)
app_images: list[dict[str, object]] = []
for index, (role, environment_key) in enumerate(bundle.APPLICATION_SPECS, start=1):
@@ -405,6 +403,14 @@ def _fixture(root: Path) -> tuple[dict[str, object], dict[str, object]]:
"target": bundle.TARGET,
"security_gate": security_gate,
"owned_postgres_build": owned_case["owned_binding"],
"object_storage_runtime_probe": make_object_storage_probe(
root,
candidate={"commit": commit, "tag": tag},
images=infra_images,
private_key=owned_case["private_key"],
domain=bundle.OBJECT_STORAGE_PROBE_DOMAIN,
schema=bundle.OBJECT_STORAGE_PROBE_SCHEMA,
),
"images": infra_images,
},
"source_set_payload_sha256",
@@ -412,6 +418,27 @@ def _fixture(root: Path) -> tuple[dict[str, object], dict[str, object]]:
return source_set, application
def _source_repo(root: Path, source: Mapping[str, object]) -> Path:
binding = source["owned_postgres_build"]
assert isinstance(binding, Mapping)
return root / str(binding["candidate_root"])
_FORMAL_BUILD_BUNDLE = bundle.build_bundle
def _test_build_bundle(**kwargs: object) -> dict[str, object]:
if "source_repo" not in kwargs:
root = Path(kwargs["artifact_root"])
source = kwargs["source_set"]
assert isinstance(source, Mapping)
kwargs["source_repo"] = _source_repo(root, source)
return _FORMAL_BUILD_BUNDLE(**kwargs)
bundle.build_bundle = _test_build_bundle
def _rehash_source(source: dict[str, object]) -> dict[str, object]:
source = dict(source)
source.pop("source_set_payload_sha256", None)
@@ -615,10 +642,105 @@ def test_bundle_is_deterministic_file_only_and_covers_all_eleven_images(
def test_owned_postgres_receipt_is_mandatory_for_bundle_and_package(tmp_path: Path) -> None:
source, application = _fixture(tmp_path)
source_repo = _source_repo(tmp_path, source)
source.pop("owned_postgres_build")
source = _rehash_source(source)
with pytest.raises(bundle.BundleValidationError, match="missing or unknown"):
bundle.build_bundle(
artifact_root=tmp_path,
source_repo=source_repo,
source_set=source,
application_manifest=application,
)
def test_repository_unconfigured_trust_cannot_be_replaced_by_staged_active_policy(
tmp_path: Path,
) -> None:
commit = subprocess.run(
["git", "-C", str(SOURCE_ROOT), "rev-parse", "HEAD"],
check=True,
capture_output=True,
text=True,
).stdout.strip()
tree = subprocess.run(
["git", "-C", str(SOURCE_ROOT), "show", "-s", "--format=%T", commit],
check=True,
capture_output=True,
text=True,
).stdout.strip()
epoch = int(
subprocess.run(
["git", "-C", str(SOURCE_ROOT), "show", "-s", "--format=%ct", commit],
check=True,
capture_output=True,
text=True,
).stdout.strip()
)
case = prepare_owned_case(
tmp_path,
validator=owned_validator,
source_root=SOURCE_ROOT,
commit=commit,
tree=tree,
epoch=epoch,
)
with pytest.raises(
bundle.BundleValidationError,
match="staged owned PostgreSQL candidate file differs from Git commit",
):
bundle._validate_owned_postgres_build(
tmp_path,
case["owned_binding"],
expected_commit=commit,
source_repo=SOURCE_ROOT,
)
def test_staged_owned_policy_change_after_commit_is_rejected(tmp_path: Path) -> None:
source, application = _fixture(tmp_path)
source_repo = _source_repo(tmp_path, source)
binding = source["owned_postgres_build"]
policy_path = tmp_path / binding["policy"]["path"]
policy_path.write_bytes(policy_path.read_bytes() + b"\n")
replacement = _record(policy_path, tmp_path)
binding["policy"] = replacement
binding["files"] = [
replacement if item["path"] == replacement["path"] else item
for item in binding["files"]
]
source = _rehash_source(source)
with pytest.raises(
bundle.BundleValidationError,
match="staged owned PostgreSQL candidate file differs from Git commit",
):
bundle.build_bundle(
artifact_root=tmp_path,
source_repo=source_repo,
source_set=source,
application_manifest=application,
)
@pytest.mark.parametrize("mutation", ("missing", "signature"))
def test_object_storage_runtime_probe_is_mandatory_and_signed(
tmp_path: Path, mutation: str
) -> None:
source, application = _fixture(tmp_path)
if mutation == "missing":
source.pop("object_storage_runtime_probe")
message = "missing or unknown"
else:
record = source["object_storage_runtime_probe"]
path = tmp_path / record["path"]
envelope = json.loads(path.read_text(encoding="utf-8"))
envelope["signature_base64"] = "A" * len(envelope["signature_base64"])
_write_canonical(path, envelope)
source["object_storage_runtime_probe"] = _record(path, tmp_path)
message = "signature is invalid"
source = _rehash_source(source)
with pytest.raises(bundle.BundleValidationError, match=message):
bundle.build_bundle(
artifact_root=tmp_path,
source_set=source,
yuqei-infra/tests/test_object_storage_runtime_probe_executor.py
+186
View File
@@ -0,0 +1,186 @@
from __future__ import annotations
import base64
import hashlib
import importlib.util
import json
import os
import sys
from pathlib import Path
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
ROOT = Path(__file__).resolve().parents[2]
SCRIPT = ROOT / "yuqei-infra/scripts/run-object-storage-runtime-probe.py"
SPEC = importlib.util.spec_from_file_location("object_storage_probe_executor_test", SCRIPT)
assert SPEC is not None and SPEC.loader is not None
probe = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = probe
SPEC.loader.exec_module(probe)
def _fixture(tmp_path: Path):
images = []
for index, role in enumerate(("minio", "minio-client"), start=1):
archive = tmp_path / f"{role}.tar"
archive.write_bytes(f"archive-{role}".encode())
config = "sha256:" + str(index) * 64
manifest = "sha256:" + str(index + 2) * 64
images.append(
{
"role": role,
"expected_repo_tag": f"yuqei-infra/{role}:offline",
"config_digest": config,
"manifest_digest": manifest,
"docker_archive": {
"path": archive.name,
"sha256": hashlib.sha256(archive.read_bytes()).hexdigest(),
"size_bytes": archive.stat().st_size,
},
}
)
source = {
"candidate": {"commit": "a" * 40, "tag": "v2-test"},
"images": images,
}
source["source_set_payload_sha256"] = hashlib.sha256(probe.canonical(source)).hexdigest()
source_path = tmp_path / "source.json"
source_path.write_bytes(probe.canonical(source))
private = Ed25519PrivateKey.generate()
key_path = tmp_path / "builder.key"
key_path.write_bytes(
private.private_bytes(
serialization.Encoding.Raw,
serialization.PrivateFormat.Raw,
serialization.NoEncryption(),
)
)
os.chmod(key_path, 0o600)
return source_path, images, private, key_path
class FakeDocker:
def __init__(self, images, *, unhealthy: bool = False, fail_client: bool = False):
self.images = images
self.unhealthy = unhealthy
self.fail_client = fail_client
self.calls = []
def __call__(self, argv, *, input_bytes=None):
self.calls.append((tuple(argv), input_bytes))
if argv[1:3] == ("image", "inspect"):
image = next(item for item in self.images if item["expected_repo_tag"] == argv[3])
entrypoint = ["/usr/bin/mc"] if image["role"] == "minio-client" else ["/usr/bin/rustfs"]
return probe.canonical(
[{"Id": image["config_digest"], "RepoDigests": ["local@" + image["manifest_digest"]], "Config": {"Entrypoint": entrypoint}}]
)
if argv[1:3] == ("container", "inspect"):
health = "unhealthy" if self.unhealthy else "healthy"
return probe.canonical([{"State": {"Running": True, "Health": {"Status": health}}}])
if argv[1] == "exec":
return b"1000\n1000\n"
if argv[1] == "run":
if self.fail_client:
raise probe.ProbeError("controlled Docker command failed: run")
assert input_bytes == (
b"http://yuqei-os-probe-aaaaaaaaaaaa-server:9000\n"
b"yuqei_probe\nyuqei_probe_password_123\n"
)
joined = " ".join(argv)
assert "yuqei_probe_password_123" not in joined
assert "alias set probe \"$probe_endpoint\"" in joined
return b"1000\n"
return b""
def test_probe_uses_fixed_offline_runtime_commands_and_signs_receipt(tmp_path: Path) -> None:
source_path, images, private, key_path = _fixture(tmp_path)
runner = FakeDocker(images)
output = tmp_path / "receipt.json"
envelope = probe.execute(
artifact_root=tmp_path,
source_set_path=source_path,
output_path=output,
private_key_path=key_path,
run=runner,
require_host=False,
)
commands = [call[0] for call in runner.calls]
create = next(command for command in commands if command[1] == "create")
client = next(command for command in commands if command[1] == "run")
assert create[2:4] == ("--pull", "never")
assert "--health-cmd" in create
assert "test -f /data/.minio.sys/format.json" in create
assert client[2:4] == ("--pull", "never")
assert "type=volume,source=yuqei-os-probe-aaaaaaaaaaaa-data,target=/probe-data,readonly" in client
assert envelope["payload"]["images"][1]["checks"]["config_from_stdin"] is True
assert envelope["payload"]["images"][1]["checks"]["mirror_roundtrip_verified"] is True
signature = base64.b64decode(envelope["signature_base64"], validate=True)
private.public_key().verify(signature, probe.DOMAIN + probe.canonical(envelope["payload"]))
assert output.stat().st_size > 0
def test_probe_fails_closed_on_unhealthy_server_and_cleans_resources(tmp_path: Path) -> None:
source_path, images, _private, key_path = _fixture(tmp_path)
runner = FakeDocker(images, unhealthy=True)
with pytest.raises(probe.ProbeError, match="healthcheck"):
probe.execute(
artifact_root=tmp_path,
source_set_path=source_path,
output_path=tmp_path / "receipt.json",
private_key_path=key_path,
run=runner,
require_host=False,
)
commands = [call[0] for call in runner.calls]
assert any(command[1:3] == ("network", "rm") for command in commands)
assert any(command[1:3] == ("volume", "rm") for command in commands)
assert not (tmp_path / "receipt.json").exists()
def test_probe_cleans_client_when_docker_run_returns_nonzero(tmp_path: Path) -> None:
source_path, images, _private, key_path = _fixture(tmp_path)
runner = FakeDocker(images, fail_client=True)
with pytest.raises(probe.ProbeError, match="failed: run"):
probe.execute(
artifact_root=tmp_path,
source_set_path=source_path,
output_path=tmp_path / "receipt.json",
private_key_path=key_path,
run=runner,
require_host=False,
)
commands = [call[0] for call in runner.calls]
assert (
probe.DOCKER,
"rm",
"--force",
"yuqei-os-probe-aaaaaaaaaaaa-client",
) in commands
assert not (tmp_path / "receipt.json").exists()
@pytest.mark.skipif(os.name != "posix", reason="POSIX key permission bits are Linux-only")
def test_probe_rejects_non_private_builder_key(tmp_path: Path) -> None:
source_path, images, _private, key_path = _fixture(tmp_path)
os.chmod(key_path, 0o644)
runner = FakeDocker(images)
with pytest.raises(probe.ProbeError, match="permissions"):
probe.execute(
artifact_root=tmp_path,
source_set_path=source_path,
output_path=tmp_path / "receipt.json",
private_key_path=key_path,
run=runner,
require_host=False,
)
yuqei-infra/tests/test_release_package_executor.py
+111 -8
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import copy
import hashlib
import importlib.util
import json
@@ -15,7 +16,7 @@ from typing import Any
import pytest
from owned_infrastructure_test_support import prepare_owned_case
from owned_infrastructure_test_support import make_object_storage_probe, prepare_owned_case
REPO_ROOT = Path(__file__).resolve().parents[2]
@@ -91,6 +92,9 @@ class PackageFixture:
_STRICT_BUILD_TEMPLATE: dict[str, Any] | None = None
_REAL_INFRA_TOOL: Any | None = None
_INFRA_SYNTHETIC_APPLICATION: dict[str, Any] | None = None
_OWNED_SOURCE_REPO: Path | None = None
_OWNED_SOURCE_COMMIT: str | None = None
_OWNED_SOURCE_MANIFEST: dict[str, Any] | None = None
_FIXTURE_MIT_TEXT = b"MIT fixture license text\n"
_FIXTURE_APACHE_TEXT = b"Apache-2.0 fixture license text\n"
_FIXTURE_GPL_TEXT = b"GPL-2.0-only fixture license text\n"
@@ -188,14 +192,34 @@ def _fixture_third_party_trust_baseline(third_tool: Any) -> dict[str, Any]:
def _fixture_infrastructure_build_bundle(
*, artifact_root: Path, source_set: dict[str, Any], application_manifest: dict[str, Any]
*,
artifact_root: Path,
source_repo: Path | None = None,
source_manifest: dict[str, Any] | None = None,
source_set: dict[str, Any],
application_manifest: dict[str, Any],
) -> dict[str, Any]:
assert _REAL_INFRA_TOOL is not None
assert _INFRA_SYNTHETIC_APPLICATION is not None
assert _OWNED_SOURCE_MANIFEST is not None
assert _OWNED_SOURCE_COMMIT is not None
validation_source = copy.deepcopy(source_set)
validation_source["candidate"]["commit"] = _OWNED_SOURCE_COMMIT
validation_source.pop("source_set_payload_sha256", None)
validation_source["source_set_payload_sha256"] = _REAL_INFRA_TOOL.canonical_sha256(
validation_source
)
validation_application = copy.deepcopy(_INFRA_SYNTHETIC_APPLICATION)
validation_application["candidate"]["commit"] = _OWNED_SOURCE_COMMIT
validation_application.pop("manifest_payload_sha256", None)
validation_application["manifest_payload_sha256"] = _REAL_INFRA_TOOL.canonical_sha256(
validation_application
)
payload = _REAL_INFRA_TOOL.build_bundle(
artifact_root=artifact_root,
source_set=source_set,
application_manifest=_INFRA_SYNTHETIC_APPLICATION,
source_manifest=_OWNED_SOURCE_MANIFEST,
source_set=validation_source,
application_manifest=validation_application,
)
application_images = []
for image in application_manifest["images"]:
@@ -214,6 +238,8 @@ def _fixture_infrastructure_build_bundle(
}
)
payload["application_images"] = application_images
payload["candidate"] = dict(source_set["candidate"])
payload["object_storage_runtime_probe"]["candidate"] = dict(source_set["candidate"])
payload["offline_compose_overrides"] = {
**{
image["environment_key"]: image["config_digest"]
@@ -234,6 +260,8 @@ def _extend_source_fixture_with_release_validators(helpers, original, tmp_path:
repo = source["repo"]
required = (
"yuqei-infra/scripts/build-infrastructure-image-bundle.py",
"yuqei-infra/scripts/validate-owned-infrastructure-build.py",
"yuqei-infra/scripts/run-object-storage-runtime-probe.py",
"yuqei-infra/scripts/build-third-party-materials.py",
"yuqei-infra/deploy/plan-first-release-snapshot-restore.py",
)
@@ -363,7 +391,7 @@ def _add_v2_release_artifacts(
*,
repo_root: Path,
) -> dict[str, Any]:
global _INFRA_SYNTHETIC_APPLICATION, _REAL_INFRA_TOOL
global _INFRA_SYNTHETIC_APPLICATION, _OWNED_SOURCE_COMMIT, _OWNED_SOURCE_MANIFEST, _OWNED_SOURCE_REPO, _REAL_INFRA_TOOL
infra_helper = _load_test_helper(
"release_package_infra_fixture_helpers",
@@ -438,10 +466,46 @@ def _add_v2_release_artifacts(
v2_root,
validator=real_infra_tool._load_owned_validator(),
source_root=REPO_ROOT,
commit=commit,
tree=installable["candidate"]["tree"],
epoch=source_date_epoch,
initialize_git=True,
)
_OWNED_SOURCE_REPO = owned_case["repo"]
_OWNED_SOURCE_COMMIT = owned_case["commit"]
owned_items = []
for relative in (
real_infra_tool._load_owned_validator().REQUIRED_POLICY_PATH,
real_infra_tool._load_owned_validator().REQUIRED_SECURITY_POLICY_PATH,
*real_infra_tool._load_owned_validator().REQUIRED_DOCKERFILES.values(),
):
raw = (owned_case["repo"] / relative).read_bytes()
owned_items.append(
{
"path": relative,
"mode": "100644",
"git_blob_oid": hashlib.sha1(
b"blob " + str(len(raw)).encode("ascii") + b"\0" + raw,
usedforsecurity=False,
).hexdigest(),
"size": len(raw),
"sha256": hashlib.sha256(raw).hexdigest(),
}
)
owned_unsigned = {
"schema_version": "yuqei.release-source-manifest.v1",
"candidate": {
"commit": owned_case["commit"],
"commit_timestamp": datetime.fromtimestamp(
owned_case["epoch"], UTC
).isoformat().replace("+00:00", "Z"),
"tree": owned_case["tree"],
"worktree_was_clean": True,
"identity_policy": "full_commit_oid_equals_HEAD",
},
"files": {"items": owned_items},
}
_OWNED_SOURCE_MANIFEST = {
**owned_unsigned,
"manifest_payload_sha256": real_infra_tool.canonical_sha256(owned_unsigned),
}
infra_tool = SimpleNamespace(
__file__=real_infra_tool.__file__,
APPLICATION_SPECS=real_infra_tool.APPLICATION_SPECS,
@@ -450,6 +514,10 @@ def _add_v2_release_artifacts(
SOURCE_SET_SCHEMA=real_infra_tool.SOURCE_SET_SCHEMA,
TARGET=real_infra_tool.TARGET,
canonical_sha256=real_infra_tool.canonical_sha256,
_load_owned_validator=real_infra_tool._load_owned_validator,
_load_object_storage_probe_executor=(
real_infra_tool._load_object_storage_probe_executor
),
build_bundle=_fixture_infrastructure_build_bundle,
)
release_package._INFRASTRUCTURE_BUNDLE_MODULE = infra_tool
@@ -529,6 +597,14 @@ def _add_v2_release_artifacts(
"target": infra_tool.TARGET,
"security_gate": infrastructure_security_gate,
"owned_postgres_build": owned_case["owned_binding"],
"object_storage_runtime_probe": make_object_storage_probe(
v2_root,
candidate={"commit": owned_case["commit"], "tag": tag},
images=infra_images,
private_key=owned_case["private_key"],
domain=real_infra_tool.OBJECT_STORAGE_PROBE_DOMAIN,
schema=real_infra_tool.OBJECT_STORAGE_PROBE_SCHEMA,
),
"images": infra_images,
}
source_set["source_set_payload_sha256"] = infra_tool.canonical_sha256(source_set)
@@ -536,6 +612,7 @@ def _add_v2_release_artifacts(
_write_json(source_path, source_set)
infrastructure_bundle = infra_tool.build_bundle(
artifact_root=v2_root,
source_repo=repo_root,
source_set=source_set,
application_manifest=installable,
)
@@ -2357,6 +2434,32 @@ def test_verify_dynamic_imports_do_not_create_bytecode_in_candidate_tree(
sys.modules[name] = saved_modules[name]
@pytest.mark.parametrize(
"relative",
(
"yuqei-infra/scripts/validate-owned-infrastructure-build.py",
"yuqei-infra/scripts/run-object-storage-runtime-probe.py",
),
)
def test_owned_validator_loaded_bytes_must_match_frozen_source_manifest(
tmp_path: Path, relative: str,
) -> None:
fixture = _make_package(tmp_path)
release_manifest = json.loads(fixture.source_file.read_text(encoding="utf-8"))
tampered = tmp_path / Path(relative).name
tampered.write_bytes((REPO_ROOT / relative).read_bytes() + b"\n# drift\n")
with pytest.raises(
release_package.ReleasePackageError,
match="Loaded validator differs from the frozen source manifest",
):
release_package._validate_source_bound_tool(
release_manifest,
fixture.repo,
relative,
loaded_file=tampered,
)
def test_install_without_apply_is_an_offline_non_writing_plan(tmp_path: Path) -> None:
fixture = _make_package(tmp_path)
state = tmp_path / "state"