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

fix(release): close cleanup and source loader gaps

Browse Source
This commit is contained in:
yuqei
2026-09-27 17:33:14 +08:00
parent 2b8ac38a81
commit bc739f63b9
7 changed files with 485 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yuqei-infra/deploy/plan-first-release-snapshot-restore.py
+67 -22
View File
@@ -17,6 +17,7 @@ import os
import re
import stat
import sys
import types
from datetime import UTC, datetime, timedelta
from pathlib import Path, PurePosixPath
from typing import Any, Callable, Mapping
@@ -83,6 +84,8 @@ PACKAGE_VALIDATOR_SOURCE_CLOSURE = (
"yuqei-infra/deploy/verify-production-compose.py",
"yuqei-infra/scripts/build-installable-release.py",
"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/release-trust/third-party-materials-baseline.json",
"yuqei-infra/scripts/linux_release_common.py",
@@ -377,34 +380,74 @@ def _bootstrap_release_package(
return root, package, actual, release_manifest, release_sha
def _load_release_package_validator(repo_root: Path) -> Callable[..., Any]:
def _verified_source_bytes(
source_manifest: Mapping[str, Any], repo_root: Path, relative: str
) -> tuple[bytes, Path]:
records = source_manifest.get("files", {}).get("items", [])
matches = [
item
for item in records
if isinstance(item, Mapping) and item.get("path") == relative
]
if len(matches) != 1:
raise SnapshotRestorePlanError(
f"release source manifest does not uniquely bind source: {relative}"
)
record = matches[0]
expected_sha = _sha(record.get("sha256"), label=f"{relative} source SHA-256")
expected_size = record.get("size")
if (
not isinstance(expected_size, int)
or isinstance(expected_size, bool)
or expected_size <= 0
):
raise SnapshotRestorePlanError(f"{relative} source size is invalid")
root = _regular_directory(repo_root, label="repository root")
relative = "yuqei-infra/deploy/release-package.py"
path = _regular_file(
root.joinpath(*PurePosixPath(relative).parts),
label="official release-package validator",
label=f"verified source {relative}",
)
expected_common = root / "yuqei-infra/scripts/linux_release_common.py"
ambient_common = sys.modules.get("linux_release_common")
if ambient_common is not None:
ambient_file = getattr(ambient_common, "__file__", None)
if ambient_file is None or Path(ambient_file).resolve() != expected_common.resolve():
raise SnapshotRestorePlanError("ambient linux_release_common module is forbidden")
module_name = "yuqei_first_release_official_release_package"
spec = importlib.util.spec_from_file_location(module_name, path)
if spec is None or spec.loader is None:
raise SnapshotRestorePlanError("official release-package validator cannot be loaded")
module = importlib.util.module_from_spec(spec)
sys.modules[module_name] = module
previous = sys.dont_write_bytecode
sys.dont_write_bytecode = True
raw = path.read_bytes()
if len(raw) != expected_size or hashlib.sha256(raw).hexdigest() != expected_sha:
raise SnapshotRestorePlanError(
f"repository source differs from the release manifest: {relative}"
)
return raw, path
def _exec_verified_module(raw: bytes, path: Path, module_name: str) -> Any:
try:
spec.loader.exec_module(module)
code = compile(raw, os.fspath(path), "exec", dont_inherit=True)
except (SyntaxError, UnicodeError, ValueError) as exc:
raise SnapshotRestorePlanError("verified validator source cannot be compiled") from exc
module = types.ModuleType(module_name)
module.__file__ = os.fspath(path)
module.__package__ = ""
module.__loader__ = None
module.__spec__ = importlib.util.spec_from_loader(
module_name, loader=None, origin=os.fspath(path)
)
sys.modules[module_name] = module
try:
exec(code, module.__dict__)
except Exception as exc:
sys.modules.pop(module_name, None)
raise SnapshotRestorePlanError("official release-package validator failed to load") from exc
finally:
sys.dont_write_bytecode = previous
raise SnapshotRestorePlanError("verified validator source failed to load") from exc
return module
def _load_release_package_validator(
repo_root: Path, source_manifest: Mapping[str, Any]
) -> Callable[..., Any]:
root = _regular_directory(repo_root, label="repository root")
relative = "yuqei-infra/deploy/release-package.py"
common_raw, common_path = _verified_source_bytes(
source_manifest, root, "yuqei-infra/scripts/linux_release_common.py"
)
_exec_verified_module(common_raw, common_path, "linux_release_common")
raw, path = _verified_source_bytes(source_manifest, root, relative)
module_name = "yuqei_first_release_official_release_package"
module = _exec_verified_module(raw, path, module_name)
validator = getattr(module, "validate_release_package", None)
if not callable(validator):
raise SnapshotRestorePlanError("official release-package validator entrypoint is missing")
@@ -451,7 +494,9 @@ def _validate_package(
expected_package_sha256,
repo_root=repo_root,
)
validator = package_validator or _load_release_package_validator(repo_root)
validator = package_validator or _load_release_package_validator(
repo_root, bootstrapped_release
)
try:
validated = validator(
root,
yuqei-infra/deploy/release-package.py
+183 -40
View File
@@ -20,10 +20,12 @@ import importlib.util
import json
import os
import re
import shutil
import stat
import subprocess
import sys
import time
import types
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import UTC, datetime
@@ -32,6 +34,7 @@ from typing import Any, Callable, Iterator, Mapping
REPO_ROOT = Path(__file__).resolve().parents[2]
_BOOTSTRAP_GIT_EXECUTABLE = shutil.which("git")
PACKAGE_FILE = "release-package.json"
BASE_COMPOSE_RELATIVE = "yuqei-infra/compose/docker-compose.yml"
PRODUCTION_COMPOSE_RELATIVE = "yuqei-infra/compose/docker-compose.production.yml"
@@ -1170,6 +1173,173 @@ def _validate_source_bound_tool(
)
def _verified_source_bytes(
source_manifest: Mapping[str, Any],
repo_root: Path,
relative: str,
) -> tuple[bytes, Path]:
matches = [
item
for item in source_manifest["files"]["items"]
if isinstance(item, Mapping) and item.get("path") == relative
]
if len(matches) != 1:
raise ReleasePackageError(
f"Release source manifest does not uniquely bind required source: {relative}"
)
record = matches[0]
expected = record.get("sha256")
size = record.get("size")
if (
not isinstance(expected, str)
or not SHA256_RE.fullmatch(expected)
or not isinstance(size, int)
or isinstance(size, bool)
or size <= 0
):
raise ReleasePackageError(f"Release source record is invalid: {relative}")
source_path = _artifact_file(
_validated_directory(repo_root, label="repository root"),
relative,
label=f"source-bound file {relative}",
)
raw = source_path.read_bytes()
if len(raw) != size or hashlib.sha256(raw).hexdigest() != expected:
raise ReleasePackageError(
f"Repository source differs from the frozen source manifest: {relative}"
)
return raw, source_path
def _exec_verified_source_module(
raw: bytes,
source_path: Path,
module_name: str,
*,
label: str,
) -> Any:
try:
code = compile(raw, os.fspath(source_path), "exec", dont_inherit=True)
except (SyntaxError, UnicodeError, ValueError) as exc:
raise ReleasePackageError(f"Cannot compile verified {label} source bytes.") from exc
module = types.ModuleType(module_name)
module.__file__ = os.fspath(source_path)
module.__package__ = ""
module.__loader__ = None
module.__spec__ = importlib.util.spec_from_loader(
module_name, loader=None, origin=os.fspath(source_path)
)
sys.modules[module_name] = module
try:
exec(code, module.__dict__)
except Exception as exc:
sys.modules.pop(module_name, None)
raise ReleasePackageError(f"Cannot execute verified {label} source bytes.") from exc
return module
def _load_verified_release_modules(
source_manifest: Mapping[str, Any], repo_root: Path
) -> dict[str, Any]:
sources = {
"common": (
"yuqei-infra/scripts/linux_release_common.py",
"linux_release_common",
"Linux release common dependency",
),
"installable": (
"yuqei-infra/scripts/build-installable-release.py",
"yuqei_release_package_installable_validator",
"installable manifest validator",
),
"infrastructure": (
"yuqei-infra/scripts/build-infrastructure-image-bundle.py",
"yuqei_release_package_infrastructure_bundle",
"infrastructure image bundle validator",
),
"owned": (
"yuqei-infra/scripts/validate-owned-infrastructure-build.py",
"yuqei_owned_infrastructure_bundle_validator",
"owned infrastructure validator",
),
"probe": (
"yuqei-infra/scripts/run-object-storage-runtime-probe.py",
"yuqei_object_storage_runtime_probe_executor",
"object-storage runtime probe executor",
),
"third_party": (
"yuqei-infra/scripts/build-third-party-materials.py",
"yuqei_release_package_third_party_materials",
"third-party materials validator",
),
"snapshot": (
"yuqei-infra/deploy/plan-first-release-snapshot-restore.py",
"yuqei_release_package_snapshot_restore_planner",
"snapshot restore planner",
),
"preflight": (
"yuqei-infra/deploy/preflight-production.py",
"yuqei_release_package_preflight",
"production preflight validator",
),
"compose": (
"yuqei-infra/deploy/verify-production-compose.py",
"yuqei_release_package_compose_gate",
"production Compose validator",
),
}
frozen: dict[str, tuple[bytes, Path, str, str]] = {}
for key, (relative, module_name, label) in sources.items():
raw, path = _verified_source_bytes(source_manifest, repo_root, relative)
frozen[key] = (raw, path, module_name, label)
_verified_source_bytes(
source_manifest,
repo_root,
"yuqei-infra/release-trust/third-party-materials-baseline.json",
)
modules: dict[str, Any] = {}
for key in (
"common",
"installable",
"owned",
"probe",
"infrastructure",
"third_party",
"snapshot",
"preflight",
"compose",
):
raw, path, module_name, label = frozen[key]
modules[key] = _exec_verified_source_module(
raw, path, module_name, label=label
)
if (
key == "common"
and sys.platform.startswith("win")
and _BOOTSTRAP_GIT_EXECUTABLE
):
modules[key]._WINDOWS_GIT_EXECUTABLE = _BOOTSTRAP_GIT_EXECUTABLE
modules["infrastructure"]._load_owned_validator = lambda: modules["owned"]
modules["infrastructure"]._load_object_storage_probe_executor = (
lambda: modules["probe"]
)
global _COMPOSE_GATE_MODULE
global _INFRASTRUCTURE_BUNDLE_MODULE
global _INSTALLABLE_VALIDATOR_MODULE
global _PREFLIGHT_MODULE
global _SNAPSHOT_RESTORE_PLANNER_MODULE
global _THIRD_PARTY_MATERIALS_MODULE
_COMPOSE_GATE_MODULE = modules["compose"]
_INFRASTRUCTURE_BUNDLE_MODULE = modules["infrastructure"]
_INSTALLABLE_VALIDATOR_MODULE = modules["installable"]
_PREFLIGHT_MODULE = modules["preflight"]
_SNAPSHOT_RESTORE_PLANNER_MODULE = modules["snapshot"]
_THIRD_PARTY_MATERIALS_MODULE = modules["third_party"]
return modules
def _validate_infrastructure_bundle(
*,
package_root: Path,
@@ -1690,6 +1860,11 @@ def validate_release_package(
release_manifest = _read_json(release_path, label="release source manifest")
source_commit = _validate_source_manifest(release_manifest)
_validate_compose_inputs(release_manifest, repo_root)
verified_modules = (
_load_verified_release_modules(release_manifest, repo_root)
if package_schema == PACKAGE_SCHEMA_V2
else {}
)
identity_path = _artifact_file(root, pins["image_identity"].path, label="image identity")
identity_sha = file_sha256(identity_path)
@@ -1736,7 +1911,11 @@ def validate_release_package(
installable_sha = file_sha256(installable_path)
if installable_sha != pins["installable_manifest"].sha256:
raise ReleasePackageError("Installable release manifest SHA-256 mismatch.")
installable_validator = _load_installable_validator_module()
installable_validator = (
verified_modules["installable"]
if verified_modules
else _load_installable_validator_module()
)
try:
installable_manifest = installable_validator.validate_installable_release_manifest(
artifact_root=installable_path.parent,
@@ -1827,45 +2006,9 @@ def validate_release_package(
third_party_profile: str | None = None
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 (
(
"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__),
),
(
"yuqei-infra/release-trust/third-party-materials-baseline.json",
Path(third_party_module.REPOSITORY_TRUST_BASELINE_PATH),
),
(
"yuqei-infra/deploy/plan-first-release-snapshot-restore.py",
Path(snapshot_module.__file__),
),
):
_validate_source_bound_tool(
release_manifest,
repo_root,
relative,
loaded_file=loaded,
)
infrastructure_module = verified_modules["infrastructure"]
third_party_module = verified_modules["third_party"]
snapshot_module = verified_modules["snapshot"]
candidate = installable_manifest.get("candidate")
candidate_tag = candidate.get("tag") if isinstance(candidate, Mapping) else None
if not isinstance(candidate_tag, str) or not candidate_tag:
yuqei-infra/scripts/run-object-storage-runtime-probe.py
+26 -13
View File
@@ -108,6 +108,26 @@ def _load_key(path: Path) -> Ed25519PrivateKey:
raise ProbeError("builder signing key is not an Ed25519 private key") from exc
def _cleanup_created(
created: list[tuple[str, str]], run: Callable[..., bytes]
) -> None:
failures: list[str] = []
for kind, name in reversed(created):
command = (
(DOCKER, "rm", "--force", name)
if kind == "container"
else (DOCKER, kind, "rm", name)
)
try:
run(command)
except Exception:
failures.append(f"{kind}:{name}")
if failures:
raise ProbeError(
"object-storage probe cleanup failed: " + ", ".join(failures)
)
def execute(
*,
artifact_root: Path,
@@ -173,6 +193,7 @@ def execute(
client = network + "-client"
volume = network + "-data"
created: list[tuple[str, str]] = []
cleanup_attempted = False
server_record, client_record = records
try:
run((DOCKER, "network", "create", "--internal", network)); created.append(("network", network))
@@ -243,6 +264,9 @@ def execute(
],
"platform": "linux/amd64",
}
cleanup_attempted = True
_cleanup_created(created, run)
created.clear()
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)
@@ -276,19 +300,8 @@ def execute(
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
if not cleanup_attempted and created:
_cleanup_created(created, run)
def main() -> int:
yuqei-infra/tests/test_build05_v2_package_orchestrator.py
+9
View File
@@ -61,6 +61,7 @@ def release_inputs(tmp_path_factory: pytest.TempPathFactory) -> dict[str, object
def _reviewed_module_loader(inputs: dict[str, object]):
base_infrastructure = helpers.release_package._load_infrastructure_bundle_module()
base_verified_loader = helpers.release_package._load_verified_release_modules
original_infrastructure_root = Path(inputs["infrastructure_root"])
def fixture_bundle_adapter(
@@ -91,6 +92,14 @@ def _reviewed_module_loader(inputs: dict[str, object]):
build_bundle=fixture_bundle_adapter,
)
def verified_loader(source_manifest, source_repo):
modules = base_verified_loader(source_manifest, source_repo)
modules["infrastructure"] = infrastructure
helpers.release_package._INFRASTRUCTURE_BUNDLE_MODULE = infrastructure
return modules
helpers.release_package._load_verified_release_modules = verified_loader
def load(path: Path, _name: str):
if path.name == "build-installable-release.py":
return helpers.release_package._load_installable_validator_module()
yuqei-infra/tests/test_first_release_snapshot_restore_planner.py
+63 -1
View File
@@ -5,6 +5,8 @@ import hashlib
import importlib.util
import inspect
import json
import os
import py_compile
import sys
from datetime import UTC, datetime, timedelta
from pathlib import Path
@@ -753,6 +755,8 @@ def test_package_rejects_wrong_trusted_build_receipt_from_official_validator(
"relative",
(
"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/release-trust/third-party-materials-baseline.json",
),
@@ -765,6 +769,8 @@ def test_package_source_closure_binds_dynamic_bundle_builders(
case = _case(tmp_path)
for required in (
"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/release-trust/third-party-materials-baseline.json",
):
@@ -786,6 +792,62 @@ def test_package_source_closure_binds_dynamic_bundle_builders(
)
def test_official_release_validator_loader_ignores_timestamp_valid_malicious_pyc(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
repo = tmp_path / "repo"
relative_release = "yuqei-infra/deploy/release-package.py"
relative_common = "yuqei-infra/scripts/linux_release_common.py"
release_path = repo.joinpath(*relative_release.split("/"))
common_path = repo.joinpath(*relative_common.split("/"))
release_path.parent.mkdir(parents=True)
common_path.parent.mkdir(parents=True)
release_raw = (REPO_ROOT / relative_release).read_bytes()
common_raw = (REPO_ROOT / relative_common).read_bytes()
malicious_prefix = (
b"import builtins\n"
b"builtins._YUQEI_MALICIOUS_PYC_EXECUTED = True\n"
b"def validate_release_package(*args, **kwargs): return None\n"
)
assert len(malicious_prefix) < len(release_raw)
malicious = malicious_prefix + b"#" + b"x" * (
len(release_raw) - len(malicious_prefix) - 1
)
release_path.write_bytes(malicious)
fixed_ns = 1_800_000_000_000_000_000
os.utime(release_path, ns=(fixed_ns, fixed_ns))
pyc_path = Path(importlib.util.cache_from_source(str(release_path)))
py_compile.compile(str(release_path), cfile=str(pyc_path), doraise=True)
release_path.write_bytes(release_raw)
os.utime(release_path, ns=(fixed_ns, fixed_ns))
common_path.write_bytes(common_raw)
manifest = {
"files": {
"items": [
{
"path": relative_release,
"size": len(release_raw),
"sha256": hashlib.sha256(release_raw).hexdigest(),
},
{
"path": relative_common,
"size": len(common_raw),
"sha256": hashlib.sha256(common_raw).hexdigest(),
},
]
}
}
import builtins
monkeypatch.delattr(
builtins, "_YUQEI_MALICIOUS_PYC_EXECUTED", raising=False
)
validator = planner._load_release_package_validator(repo, manifest)
assert callable(validator)
assert not hasattr(builtins, "_YUQEI_MALICIOUS_PYC_EXECUTED")
def _validate_saved_receipt(
case: dict[str, Any],
receipt: dict[str, Any],
@@ -798,7 +860,7 @@ def _validate_saved_receipt(
monkeypatch.setattr(
planner,
"_load_release_package_validator",
lambda _repo_root: case["package_validator"],
lambda _repo_root, _source_manifest: case["package_validator"],
)
monkeypatch.setattr(planner, "REPO_ROOT", case["repo_root"])
return planner.validate_install_readiness(
yuqei-infra/tests/test_object_storage_runtime_probe_executor.py
+45 -1
View File
@@ -63,14 +63,33 @@ def _fixture(tmp_path: Path):
class FakeDocker:
def __init__(self, images, *, unhealthy: bool = False, fail_client: bool = False):
def __init__(
self,
images,
*,
unhealthy: bool = False,
fail_client: bool = False,
fail_cleanup: str | None = None,
):
self.images = images
self.unhealthy = unhealthy
self.fail_client = fail_client
self.fail_cleanup = fail_cleanup
self.calls = []
def __call__(self, argv, *, input_bytes=None):
self.calls.append((tuple(argv), input_bytes))
cleanup_kind = None
if argv[1] == "rm" and argv[-1].endswith("-client"):
cleanup_kind = "client"
elif argv[1] == "rm" and argv[-1].endswith("-server"):
cleanup_kind = "server"
elif argv[1:3] == ("volume", "rm"):
cleanup_kind = "volume"
elif argv[1:3] == ("network", "rm"):
cleanup_kind = "network"
if self.fail_cleanup is not None and cleanup_kind == self.fail_cleanup:
raise probe.ProbeError(f"cleanup failed: {cleanup_kind}")
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"]
@@ -169,6 +188,31 @@ def test_probe_cleans_client_when_docker_run_returns_nonzero(tmp_path: Path) ->
assert not (tmp_path / "receipt.json").exists()
@pytest.mark.parametrize("cleanup_kind", ("client", "server", "volume", "network"))
def test_probe_cleanup_failure_prevents_receipt_publication(
tmp_path: Path, cleanup_kind: str
) -> None:
source_path, images, _private, key_path = _fixture(tmp_path)
runner = FakeDocker(images, fail_cleanup=cleanup_kind)
output = tmp_path / "receipt.json"
with pytest.raises(probe.ProbeError, match="probe cleanup failed"):
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]
assert sum(command[1] == "rm" for command in commands) == 2
assert any(command[1:3] == ("volume", "rm") for command in commands)
assert any(command[1:3] == ("network", "rm") for command in commands)
assert not output.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)
yuqei-infra/tests/test_release_package_executor.py
+92 -7
View File
@@ -5,6 +5,7 @@ import hashlib
import importlib.util
import json
import os
import py_compile
import shutil
import subprocess
import sys
@@ -33,6 +34,9 @@ def _load_script():
release_package = _load_script()
_REAL_VERIFIED_SOURCE_LOADER = release_package._load_verified_release_modules
_TEST_PREFLIGHT_MODULE = release_package._load_preflight_module()
_TEST_COMPOSE_MODULE = release_package._load_compose_gate_module()
def test_preflight_summary_binds_tls_material_details(monkeypatch) -> None:
@@ -259,7 +263,9 @@ def _extend_source_fixture_with_release_validators(helpers, original, tmp_path:
source = original(tmp_path)
repo = source["repo"]
required = (
"yuqei-infra/deploy/preflight-production.py",
"yuqei-infra/scripts/build-infrastructure-image-bundle.py",
"yuqei-infra/scripts/build-installable-release.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",
@@ -521,6 +527,21 @@ def _add_v2_release_artifacts(
build_bundle=_fixture_infrastructure_build_bundle,
)
release_package._INFRASTRUCTURE_BUNDLE_MODULE = infra_tool
def fixture_verified_source_loader(source_manifest, source_repo):
snapshot_override = release_package._SNAPSHOT_RESTORE_PLANNER_MODULE
modules = _REAL_VERIFIED_SOURCE_LOADER(source_manifest, source_repo)
modules["infrastructure"] = infra_tool
modules["preflight"] = _TEST_PREFLIGHT_MODULE
modules["compose"] = _TEST_COMPOSE_MODULE
if isinstance(snapshot_override, SimpleNamespace):
modules["snapshot"] = snapshot_override
release_package._INFRASTRUCTURE_BUNDLE_MODULE = infra_tool
release_package._PREFLIGHT_MODULE = modules["preflight"]
release_package._COMPOSE_GATE_MODULE = modules["compose"]
release_package._SNAPSHOT_RESTORE_PLANNER_MODULE = modules["snapshot"]
return modules
release_package._load_verified_release_modules = fixture_verified_source_loader
repositories = (
"yuqei-owned/postgres-runtime",
"yuqei-owned/postgres-client",
@@ -1839,7 +1860,7 @@ def test_v2_package_rejects_profile_distribution_scope_mismatch(tmp_path: Path)
)
def test_v2_source_bound_third_party_trust_baseline_drift_is_rejected(
def test_v2_ambient_third_party_module_drift_is_ignored_for_verified_source(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
@@ -1859,12 +1880,12 @@ def test_v2_source_bound_third_party_trust_baseline_drift_is_rejected(
loaded_baseline_path,
)
with pytest.raises(release_package.ReleasePackageError, match="frozen source manifest"):
release_package.validate_release_package(
fixture.root,
expected_package_sha256=fixture.expected_sha256,
repo_root=fixture.repo,
)
validated = release_package.validate_release_package(
fixture.root,
expected_package_sha256=fixture.expected_sha256,
repo_root=fixture.repo,
)
assert validated.root == fixture.root
def test_v2_self_consistent_third_party_report_drift_from_formal_build_is_rejected(
@@ -2460,6 +2481,70 @@ def test_owned_validator_loaded_bytes_must_match_frozen_source_manifest(
)
def test_verified_source_loader_ignores_timestamp_valid_malicious_pyc(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
repo = tmp_path / "repo"
source = repo / "verified_gate.py"
source.parent.mkdir(parents=True)
safe = b"VALUE = 'safe'\n"
malicious = b"VALUE = 'evil'\n"
assert len(safe) == len(malicious)
source.write_bytes(malicious)
fixed_ns = 1_800_000_000_000_000_000
os.utime(source, ns=(fixed_ns, fixed_ns))
pyc_path = Path(importlib.util.cache_from_source(str(source)))
py_compile.compile(str(source), cfile=str(pyc_path), doraise=True)
source.write_bytes(safe)
os.utime(source, ns=(fixed_ns, fixed_ns))
manifest = {
"files": {
"items": [
{
"path": "verified_gate.py",
"size": len(safe),
"sha256": hashlib.sha256(safe).hexdigest(),
}
]
}
}
ambient_spec = importlib.util.spec_from_file_location("ambient_pyc_gate", source)
assert ambient_spec is not None and ambient_spec.loader is not None
ambient = importlib.util.module_from_spec(ambient_spec)
ambient_spec.loader.exec_module(ambient)
assert ambient.VALUE == "evil"
raw, path = release_package._verified_source_bytes(
manifest, repo, "verified_gate.py"
)
verified = release_package._exec_verified_source_module(
raw, path, "verified_source_gate", label="test gate"
)
assert verified.VALUE == "safe"
def test_verified_source_loader_checks_full_closure_before_executing_code(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
fixture = _make_package(tmp_path)
manifest = json.loads(fixture.source_file.read_text(encoding="utf-8"))
late_source = fixture.repo / "yuqei-infra/deploy/verify-production-compose.py"
late_source.write_bytes(late_source.read_bytes() + b"\n# drift\n")
executions: list[str] = []
def forbidden_exec(*_args, **_kwargs):
executions.append("executed")
raise AssertionError("no source may execute before the full closure is verified")
monkeypatch.setattr(release_package, "_exec_verified_source_module", forbidden_exec)
with pytest.raises(
release_package.ReleasePackageError,
match="Repository source differs from the frozen source manifest",
):
_REAL_VERIFIED_SOURCE_LOADER(manifest, fixture.repo)
assert executions == []
def test_install_without_apply_is_an_offline_non_writing_plan(tmp_path: Path) -> None:
fixture = _make_package(tmp_path)
state = tmp_path / "state"