61 lines
3.0 KiB
Python
61 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from test_hub_candidate_import import modules # noqa: F401 - reuse importer isolation fixture
|
|
|
|
|
|
RECEIPT = json.loads(
|
|
(Path(__file__).parent / "fixtures/hub_security_base_publication_23280.json").read_text()
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("row", RECEIPT["preparedImages"], ids=lambda row: row["service"])
|
|
def test_actual_published_security_image_requires_exact_archive_id_and_digest(modules, monkeypatch, row):
|
|
importer, controller = modules.importer, modules.controller
|
|
service = row["service"]
|
|
repository, digest = row["imageRef"].split("@", 1)
|
|
repo_digest = repository.rsplit(":", 1)[0] + "@" + digest
|
|
assert RECEIPT["revision"] == "3237e4d8aec312d63d13b4e80664422a7520b5f9"
|
|
assert RECEIPT["runId"] == 23280 and RECEIPT["runAttempt"] == 1
|
|
assert RECEIPT["candidateEmitted"] is False and RECEIPT["reportCount"] == 5
|
|
assert row["runtimeVerified"] and row["vulnerabilityAndSecretVerified"]
|
|
inspected = {
|
|
row["archiveRef"]: {"Id": row["imageId"]},
|
|
row["imageRef"]: {"Id": row["imageId"], "RepoDigests": [repo_digest]},
|
|
}
|
|
monkeypatch.setattr(importer, "_inspect_image", inspected.get)
|
|
assert importer._require_loaded_image(service, row, external=True)["Id"] == row["imageId"]
|
|
binding = controller.mcp_hub_external_image_binding(service, row["imageRef"])
|
|
assert binding == (row["imageRef"], row["archiveRef"], repo_digest)
|
|
old_binding = controller.mcp_hub_external_image_binding(service, row["parentRef"])
|
|
assert old_binding is not None
|
|
assert old_binding[1] != row["archiveRef"]
|
|
with pytest.raises(importer.CandidateRejected, match="reviewed external image binding"):
|
|
importer._require_loaded_image(service, {**row, "archiveRef": old_binding[1]}, external=True)
|
|
inspected[row["archiveRef"]]["Id"] = "sha256:" + "0" * 64
|
|
with pytest.raises(importer.CandidateRejected, match="Loaded Docker image ID differs"):
|
|
importer._require_loaded_image(service, row, external=True)
|
|
inspected[row["archiveRef"]]["Id"] = row["imageId"]
|
|
inspected[row["imageRef"]]["RepoDigests"] = [old_binding[2]]
|
|
with pytest.raises(importer.CandidateRejected, match="Pinned external digest"):
|
|
importer._require_loaded_image(service, row, external=True)
|
|
|
|
|
|
@pytest.mark.parametrize("row", RECEIPT["preparedImages"], ids=lambda row: row["service"])
|
|
def test_actual_security_pin_preserves_only_reviewed_forward_predecessors(modules, row):
|
|
controller = modules.controller
|
|
allowed = controller._mcp_hub_observed_external_ref_allowed
|
|
service, new, old = row["service"], row["imageRef"], row["parentRef"]
|
|
assert allowed(service, new, old)
|
|
assert allowed(service, new, new)
|
|
assert not allowed(service, old, new)
|
|
assert not allowed(service, new, old + "-drift")
|
|
assert not allowed(service, new[:-1] + "0", old)
|
|
if service == "postgres":
|
|
assert allowed(service, new, controller.MCP_HUB_POSTGRES_LEGACY_IMAGE)
|
|
assert allowed(service, new, controller.MCP_HUB_POSTGRES_PREVIOUS_IMAGE)
|