import copy import io import json import subprocess import sys from pathlib import Path import pytest from scripts import build_mcp_hub_release_images as producer from scripts import prepare_release_security_bases as preparation def test_actual_cli_loader_preserves_closed_exception_identity() -> None: root = Path(__file__).resolve().parents[1] program = """import ast,sys,types from pathlib import Path root=Path.cwd() sys.path.insert(0,str(root/'scripts')) path=root/'scripts/build_mcp_hub_release_images.py' syntax=ast.parse(path.read_text()) assert isinstance(syntax.body[-1],ast.If) syntax.body.pop() module=types.ModuleType('__main__') module.__file__=str(path) module.__package__='' sys.modules['__main__']=module exec(compile(syntax,str(path),'exec'),module.__dict__) helper=module._security_preparation_module() assert helper.producer is module assert helper.producer.BuildRejected is module.BuildRejected assert helper.producer.VulnerabilityRejected is module.VulnerabilityRejected try: helper.tag('unsupported','a'*40,123,1) except module.BuildRejected as error: assert str(error)=='security base publication identity is invalid' else: raise AssertionError('closed rejection missing') print('CLI exception identity PASS') """ result = subprocess.run( # noqa: S603 - fixed interpreter and static CLI identity harness [sys.executable, "-c", program], cwd=root, capture_output=True, text=True, timeout=20 ) assert result.returncode == 0, result.stderr assert result.stdout.strip() == "CLI exception identity PASS" def test_fixed_publication_names_bind_exact_protected_attempt() -> None: assert preparation.tag("postgres", "a" * 40, 123, 2) == ( producer.IMAGE_PREFIX + "/postgres:" + "a" * 40 + "-security-run-123-attempt-2" ) for args in [ ("worker", "a" * 40, 123, 2), ("postgres", "secret", 123, 2), ("postgres", "a" * 40, 0, 2), ("postgres", "a" * 40, 123, 0), ]: with pytest.raises(producer.BuildRejected): preparation.tag(*args) @pytest.mark.parametrize( "field", ["User", "Entrypoint", "Cmd", "Volumes", "Env", "Healthcheck", "WorkingDir", "StopSignal"], ) def test_base_lifecycle_cannot_be_changed_or_echoed(field: str) -> None: base = {"config": {field: "private-value"}} preparation.verify_lifecycle(base, {"config": dict(base["config"])}) with pytest.raises(producer.BuildRejected) as error: preparation.verify_lifecycle(base, {"config": {field: "private-new-value"}}) assert str(error.value) == "security base runtime lifecycle differs from its exact parent" def test_unknown_download_asset_is_denied_before_network(tmp_path: Path) -> None: with pytest.raises(producer.BuildRejected): preparation.download_asset( "secret", tmp_path, opener=lambda *args, **kwargs: pytest.fail("network called") ) def test_arbitrary_scanner_subset_cannot_replace_full_release(tmp_path: Path) -> None: with pytest.raises(producer.BuildRejected, match="fixed release or preparation"): producer._install_and_scan( images={}, temp_root=tmp_path, env={}, cwd=tmp_path, runner=lambda *args, **kwargs: None, services=("connector-gitea",), ) @pytest.mark.parametrize("data", [b"bad", b"longer-than-fixed-size"]) def test_corrupt_assets_are_rejected_without_writing( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, data: bytes ) -> None: monkeypatch.setitem(preparation.ASSETS, "go", ("https://go.dev/fixed", "asset", "a" * 64, 3)) with pytest.raises(producer.BuildRejected) as error: preparation.download_asset("go", tmp_path, opener=lambda *args, **kwargs: io.BytesIO(data)) assert "bad" not in str(error.value) assert not list(tmp_path.iterdir()) @pytest.mark.parametrize("failure", [None, "scan", "lifecycle", "pull"]) def test_only_verified_scanned_images_publish_and_no_candidate_is_emitted( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, failure: str | None ) -> None: root = Path(__file__).resolve().parents[1] output = tmp_path / "published" events: list[str] = [] built: dict[str, dict] = {} parent = { "imageId": "sha256:" + "b" * 64, "config": { "User": "", "Entrypoint": ["/original"], "Cmd": ["original"], "Env": ["STATIC=unchanged"], "Volumes": {"/original": {}}, }, } def execute(command: list[str], **kwargs: object) -> None: stage = str(kwargs["stage"]) events.append(stage) if stage == "security_base_build": reference = command[command.index("--tag") + 1] config = copy.deepcopy(parent) config["config"]["Labels"] = dict( command[index + 1].split("=", 1) for index, value in enumerate(command) if value == "--label" ) if failure == "lifecycle": config["config"]["User"] = "10001" built[reference] = config def inspect(reference: str, *args: object, **kwargs: object) -> dict: if reference in {value[0] for value in producer.EXTERNAL_IMAGES.values()}: return copy.deepcopy(parent) result = copy.deepcopy(built[reference.split("@", 1)[0]]) if "@" in reference and failure == "pull": result["imageId"] = "sha256:" + "c" * 64 return result def scan(**kwargs: object) -> Path: events.append("scan-both") assert kwargs["services"] == producer.SECURITY_BASE_SERVICES assert set(kwargs["images"]) == set(producer.SECURITY_BASE_SERVICES) if failure == "scan": raise producer.BuildRejected("synthetic scan denial") reports = tmp_path / "reports" reports.mkdir() names = {"trivy-db-metadata.json"} | { f"{kind}-{service}.json" for service in producer.SECURITY_BASE_SERVICES for kind in ("vulnerability", "secret") } for name in names: producer._write_new(reports / name, b"{}\n") return reports monkeypatch.setattr(producer, "_execute", execute) monkeypatch.setattr(producer, "_inspect_image", inspect) monkeypatch.setattr(producer, "_registry_tag_absent", lambda *args, **kwargs: None) monkeypatch.setattr( producer, "_capture", lambda *args, **kwargs: json.dumps( { "digest": "sha256:" + "d" * 64, "mediaType": "application/vnd.oci.image.manifest.v1+json", "size": 123, } ), ) monkeypatch.setattr(producer, "_install_and_scan", scan) monkeypatch.setattr(preparation, "download_asset", lambda *args, **kwargs: None) arguments = dict( root=root, temp_root=tmp_path, output=output, env={}, revision="a" * 40, tree="e" * 40, epoch="123", built_at="2026-10-14T00:00:00Z", run_id=123, attempt=1, runner=lambda *args, **kwargs: None, ) if failure is not None: with pytest.raises(producer.BuildRejected): preparation.prepare(**arguments) assert not output.exists() if failure != "pull": assert "security_base_push" not in events else: receipt = preparation.prepare(**arguments) assert ( events.count("security_base_build") == events.count("security_base_runtime_verify") == 2 ) assert events.count("security_base_push") == 2 assert events.index("scan-both") < events.index("security_base_push") assert receipt["candidateEmitted"] is False and receipt["reportCount"] == 5 assert [row["service"] for row in receipt["preparedImages"]] == list( producer.SECURITY_BASE_SERVICES ) assert all( row["archiveRef"].endswith("-security-run-123-attempt-1") for row in receipt["preparedImages"] ) assert not (output / "candidate.json").exists() assert not (output / "images.tar").exists()