590 lines
22 KiB
Python
590 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
import gzip
|
|
import io
|
|
import subprocess
|
|
import tarfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from scripts import image_release
|
|
|
|
|
|
COMMIT = "a" * 40
|
|
DIGEST = "sha256:" + "b" * 64
|
|
IMAGE_ID = "sha256:" + "c" * 64
|
|
REPO = Path(__file__).resolve().parents[2]
|
|
MIGRATION_DIR = REPO / "trex_core" / "adapters" / "persistence" / "sql"
|
|
|
|
|
|
class RecordingRunner:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
fail_on: tuple[str, ...] | None = None,
|
|
digest: str = DIGEST,
|
|
revision: str = COMMIT,
|
|
image_id: str = IMAGE_ID,
|
|
loaded_image: str = f"trex-core-ci:{COMMIT}",
|
|
):
|
|
self.commands: list[tuple[list[str], str | None, int | None]] = []
|
|
self.fail_on = fail_on
|
|
self.digest = digest
|
|
self.revision = revision
|
|
self.image_id = image_id
|
|
self.loaded_image = loaded_image
|
|
|
|
def run(self, command, *, input_text=None, timeout=None):
|
|
command = list(command)
|
|
self.commands.append((command, input_text, timeout))
|
|
if self.fail_on and command[: len(self.fail_on)] == list(self.fail_on):
|
|
raise subprocess.CalledProcessError(1, command)
|
|
if command[:2] == ["docker", "load"]:
|
|
return image_release.Result(f"Loaded image: {self.loaded_image}\n")
|
|
if command[:3] == ["docker", "image", "inspect"] and "{{.Id}}" in command:
|
|
return image_release.Result(self.image_id + "\n")
|
|
if command[:3] == ["docker", "image", "inspect"]:
|
|
return image_release.Result(self.revision + "\n")
|
|
if command[:2] == ["docker", "push"]:
|
|
return image_release.Result(f"digest: {self.digest} size: 1234\n")
|
|
if command[:2] == ["docker", "create"]:
|
|
return image_release.Result("container-id\n")
|
|
return image_release.Result()
|
|
|
|
|
|
def _artifact(tmp_path: Path) -> tuple[Path, Path, Path]:
|
|
archive = tmp_path / "trex-core-image.tar"
|
|
checksum = tmp_path / "trex-core-image.tar.sha256"
|
|
image_id = tmp_path / "trex-core-image.id"
|
|
archive.write_bytes(b"tested image")
|
|
image_release.write_checksum(archive, checksum)
|
|
image_id.write_text(IMAGE_ID + "\n", encoding="ascii")
|
|
return archive, checksum, image_id
|
|
|
|
|
|
def test_release_tags_are_deterministic_and_never_latest():
|
|
main = image_release.release_tags(COMMIT, "refs/heads/main")
|
|
release = image_release.release_tags(COMMIT, "refs/tags/v1.2.3")
|
|
|
|
assert main == [f"{image_release.IMAGE}:{COMMIT[:12]}"]
|
|
assert release == [*main, f"{image_release.IMAGE}:v1.2.3"]
|
|
assert all(not tag.endswith(":latest") for tag in main + release)
|
|
|
|
|
|
@pytest.mark.parametrize("commit", ["A" * 40, "a" * 39, "g" * 40, ""])
|
|
def test_invalid_source_commit_is_rejected(commit):
|
|
with pytest.raises(ValueError):
|
|
image_release.release_tags(commit, "refs/heads/main")
|
|
|
|
|
|
def test_smoke_uses_exact_isolated_probes_and_cleans_containers():
|
|
runner = RecordingRunner()
|
|
image_release.smoke_image("candidate", runner, timeout=17, migration_dir=MIGRATION_DIR)
|
|
|
|
runs = [call for call in runner.commands if call[0][:2] == ["docker", "run"]]
|
|
removals = [call for call in runner.commands if call[0][:2] == ["docker", "rm"]]
|
|
expected_migrations = tuple(sorted(path.name for path in MIGRATION_DIR.glob("*.sql")))
|
|
assert expected_migrations
|
|
assert len(runs) == len(removals) == 5
|
|
assert [call[0][call[0].index("--entrypoint") + 1 :] for call in runs] == [
|
|
["python", "candidate", "-c", "import trex_core"],
|
|
["trex-core-daemon", "candidate", "--help"],
|
|
["trex-core", "candidate", "--help"],
|
|
["trex-core-reset", "candidate", "--help"],
|
|
["python", "candidate", "-I", "-c", image_release.migration_inventory_code(expected_migrations)],
|
|
]
|
|
assert all("--network" in call[0] and "none" in call[0] for call in runs)
|
|
assert all("--read-only" in call[0] and "--cap-drop" in call[0] for call in runs)
|
|
assert all(call[2] == 17 for call in runs + removals)
|
|
assert all("--config" not in call[0] and "--yes" not in call[0] for call in runs)
|
|
|
|
|
|
def test_smoke_rejects_empty_canonical_migration_inventory(tmp_path):
|
|
with pytest.raises(ValueError, match="canonical migration inventory is empty"):
|
|
image_release.smoke_image("candidate", RecordingRunner(), migration_dir=tmp_path)
|
|
|
|
|
|
def test_migration_inventory_probe_failure_propagates_after_cleanup():
|
|
class FifthProbeFailsRunner(RecordingRunner):
|
|
def run(self, command, *, input_text=None, timeout=None):
|
|
run_count = len([call for call in self.commands if call[0][:2] == ["docker", "run"]])
|
|
if list(command)[:2] == ["docker", "run"] and run_count == 4:
|
|
self.commands.append((list(command), input_text, timeout))
|
|
raise subprocess.CalledProcessError(1, command)
|
|
return super().run(command, input_text=input_text, timeout=timeout)
|
|
|
|
runner = FifthProbeFailsRunner()
|
|
with pytest.raises(RuntimeError, match="smoke probe 5 failed"):
|
|
image_release.smoke_image("candidate", runner, migration_dir=MIGRATION_DIR)
|
|
assert runner.commands[-1][0][:3] == ["docker", "rm", "--force"]
|
|
|
|
|
|
def test_smoke_failure_propagates_after_cleanup():
|
|
runner = RecordingRunner(fail_on=("docker", "run"))
|
|
with pytest.raises(RuntimeError, match="smoke probe 1 failed"):
|
|
image_release.smoke_image("candidate", runner)
|
|
assert [call[0][:3] for call in runner.commands] == [
|
|
["docker", "run", "--name"],
|
|
["docker", "rm", "--force"],
|
|
]
|
|
|
|
|
|
def test_smoke_fails_if_captured_probe_output_contains_canary():
|
|
class LeakingRunner(RecordingRunner):
|
|
def run(self, command, *, input_text=None, timeout=None):
|
|
result = super().run(command, input_text=input_text, timeout=timeout)
|
|
if list(command)[:2] == ["docker", "run"]:
|
|
return image_release.Result(image_release.CANARY.decode())
|
|
return result
|
|
|
|
runner = LeakingRunner()
|
|
with pytest.raises(RuntimeError, match="smoke probe 1 failed"):
|
|
image_release.smoke_image("candidate", runner)
|
|
assert runner.commands[-1][0][:3] == ["docker", "rm", "--force"]
|
|
|
|
|
|
def test_mutated_archive_stops_before_load_or_registry(tmp_path):
|
|
archive, checksum, image_id = _artifact(tmp_path)
|
|
archive.write_bytes(b"mutated")
|
|
runner = RecordingRunner()
|
|
|
|
with pytest.raises(ValueError, match="checksum mismatch"):
|
|
image_release.publish_image(
|
|
archive=archive,
|
|
checksum=checksum,
|
|
image_id=image_id,
|
|
local_image=f"trex-core-ci:{COMMIT}",
|
|
commit=COMMIT,
|
|
ref="refs/heads/main",
|
|
actor="actor",
|
|
token="fake-token",
|
|
runner=runner,
|
|
)
|
|
assert runner.commands == []
|
|
|
|
|
|
def test_load_failure_stops_before_login_and_push(tmp_path):
|
|
archive, checksum, image_id = _artifact(tmp_path)
|
|
runner = RecordingRunner(fail_on=("docker", "load"))
|
|
|
|
with pytest.raises(subprocess.CalledProcessError):
|
|
image_release.publish_image(
|
|
archive=archive,
|
|
checksum=checksum,
|
|
image_id=image_id,
|
|
local_image=f"trex-core-ci:{COMMIT}",
|
|
commit=COMMIT,
|
|
ref="refs/heads/main",
|
|
actor="actor",
|
|
token="fake-token",
|
|
runner=runner,
|
|
)
|
|
assert not any(call[0][:2] in (["docker", "login"], ["docker", "push"]) for call in runner.commands)
|
|
|
|
|
|
def test_revision_mismatch_stops_before_login_and_push(tmp_path):
|
|
archive, checksum, image_id = _artifact(tmp_path)
|
|
runner = RecordingRunner(revision="c" * 40)
|
|
|
|
with pytest.raises(ValueError, match="revision does not match"):
|
|
image_release.publish_image(
|
|
archive=archive,
|
|
checksum=checksum,
|
|
image_id=image_id,
|
|
local_image=f"trex-core-ci:{COMMIT}",
|
|
commit=COMMIT,
|
|
ref="refs/heads/main",
|
|
actor="actor",
|
|
token="fake-token",
|
|
runner=runner,
|
|
)
|
|
assert not any(call[0][:2] in (["docker", "login"], ["docker", "push"]) for call in runner.commands)
|
|
|
|
|
|
def test_expected_image_id_mismatch_stops_before_credentials_tags_or_push(tmp_path):
|
|
archive, checksum, image_id = _artifact(tmp_path)
|
|
runner = RecordingRunner(image_id="sha256:" + "d" * 64)
|
|
|
|
with pytest.raises(ValueError, match="image ID does not match"):
|
|
image_release.publish_image(
|
|
archive=archive,
|
|
checksum=checksum,
|
|
image_id=image_id,
|
|
local_image=f"trex-core-ci:{COMMIT}",
|
|
commit=COMMIT,
|
|
ref="refs/heads/main",
|
|
actor="actor",
|
|
token="fake-token",
|
|
runner=runner,
|
|
)
|
|
assert not any(
|
|
call[0][:2] in (["docker", "tag"], ["docker", "login"], ["docker", "push"])
|
|
for call in runner.commands
|
|
)
|
|
|
|
|
|
def test_unrelated_load_output_cannot_use_stale_expected_local_tag(tmp_path):
|
|
archive, checksum, image_id = _artifact(tmp_path)
|
|
runner = RecordingRunner(loaded_image="unrelated:image", image_id=IMAGE_ID)
|
|
|
|
with pytest.raises(ValueError, match="did not report the expected local image"):
|
|
image_release.publish_image(
|
|
archive=archive,
|
|
checksum=checksum,
|
|
image_id=image_id,
|
|
local_image=f"trex-core-ci:{COMMIT}",
|
|
commit=COMMIT,
|
|
ref="refs/heads/main",
|
|
actor="actor",
|
|
token="fake-token",
|
|
runner=runner,
|
|
)
|
|
assert not any(
|
|
call[0][:2] in (["docker", "tag"], ["docker", "login"], ["docker", "push"])
|
|
for call in runner.commands
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"invalid_id",
|
|
["", "sha256:" + "c" * 63, "sha256:" + "C" * 64, "sha512:" + "c" * 64],
|
|
)
|
|
def test_invalid_tested_image_id_stops_before_load(tmp_path, invalid_id):
|
|
archive, checksum, image_id = _artifact(tmp_path)
|
|
image_id.write_text(invalid_id, encoding="ascii")
|
|
runner = RecordingRunner()
|
|
|
|
with pytest.raises(ValueError, match="image ID evidence"):
|
|
image_release.publish_image(
|
|
archive=archive,
|
|
checksum=checksum,
|
|
image_id=image_id,
|
|
local_image=f"trex-core-ci:{COMMIT}",
|
|
commit=COMMIT,
|
|
ref="refs/heads/main",
|
|
actor="actor",
|
|
token="fake-token",
|
|
runner=runner,
|
|
)
|
|
assert runner.commands == []
|
|
|
|
|
|
def test_record_image_id_writes_strict_inspected_id(tmp_path):
|
|
output = tmp_path / "trex-core-image.id"
|
|
runner = RecordingRunner()
|
|
assert image_release.write_image_id("candidate", output, runner) == IMAGE_ID
|
|
assert output.read_text(encoding="ascii") == IMAGE_ID + "\n"
|
|
assert runner.commands == [
|
|
(["docker", "image", "inspect", "--format", "{{.Id}}", "candidate"], None, None)
|
|
]
|
|
|
|
|
|
def test_valid_publish_records_digest_only_after_push(tmp_path):
|
|
archive, checksum, image_id = _artifact(tmp_path)
|
|
output = tmp_path / "output"
|
|
summary = tmp_path / "summary"
|
|
runner = RecordingRunner()
|
|
|
|
digest = image_release.publish_image(
|
|
archive=archive,
|
|
checksum=checksum,
|
|
image_id=image_id,
|
|
local_image=f"trex-core-ci:{COMMIT}",
|
|
commit=COMMIT,
|
|
ref="refs/tags/v2.0.0",
|
|
actor="actor",
|
|
token="fake-token",
|
|
runner=runner,
|
|
output_path=output,
|
|
summary_path=summary,
|
|
)
|
|
|
|
commands = [call[0][:2] for call in runner.commands]
|
|
assert commands.index(["docker", "load"]) < commands.index(["docker", "login"])
|
|
assert commands.count(["docker", "push"]) == 2
|
|
assert digest == DIGEST
|
|
assert f"digest={DIGEST}" in output.read_text()
|
|
assert COMMIT in output.read_text()
|
|
assert f"{image_release.IMAGE}@{DIGEST}" in summary.read_text()
|
|
|
|
|
|
@pytest.mark.parametrize("bad_digest", ["", "sha256:" + "b" * 63, "sha512:" + "b" * 64])
|
|
def test_invalid_post_push_digest_emits_no_success_evidence(tmp_path, bad_digest):
|
|
archive, checksum, image_id = _artifact(tmp_path)
|
|
output = tmp_path / "output"
|
|
summary = tmp_path / "summary"
|
|
runner = RecordingRunner(digest=bad_digest)
|
|
|
|
with pytest.raises(ValueError, match="valid digest"):
|
|
image_release.publish_image(
|
|
archive=archive,
|
|
checksum=checksum,
|
|
image_id=image_id,
|
|
local_image=f"trex-core-ci:{COMMIT}",
|
|
commit=COMMIT,
|
|
ref="refs/heads/main",
|
|
actor="actor",
|
|
token="fake-token",
|
|
runner=runner,
|
|
output_path=output,
|
|
summary_path=summary,
|
|
)
|
|
assert any(call[0][:2] == ["docker", "push"] for call in runner.commands)
|
|
assert not output.exists()
|
|
assert not summary.exists()
|
|
|
|
|
|
def test_release_second_push_failure_preserves_nonzero_without_success_evidence(tmp_path):
|
|
class SecondPushFailsRunner(RecordingRunner):
|
|
def run(self, command, *, input_text=None, timeout=None):
|
|
result = super().run(command, input_text=input_text, timeout=timeout)
|
|
pushes = [call for call in self.commands if call[0][:2] == ["docker", "push"]]
|
|
if list(command)[:2] == ["docker", "push"] and len(pushes) == 2:
|
|
return image_release.Result("digest unavailable\n")
|
|
return result
|
|
|
|
archive, checksum, image_id = _artifact(tmp_path)
|
|
output = tmp_path / "output"
|
|
summary = tmp_path / "summary"
|
|
runner = SecondPushFailsRunner()
|
|
|
|
with pytest.raises(ValueError, match="valid digest"):
|
|
image_release.publish_image(
|
|
archive=archive,
|
|
checksum=checksum,
|
|
image_id=image_id,
|
|
local_image=f"trex-core-ci:{COMMIT}",
|
|
commit=COMMIT,
|
|
ref="refs/tags/v2.0.0",
|
|
actor="actor",
|
|
token="fake-token",
|
|
runner=runner,
|
|
output_path=output,
|
|
summary_path=summary,
|
|
)
|
|
assert len([call for call in runner.commands if call[0][:2] == ["docker", "push"]]) == 2
|
|
assert not output.exists()
|
|
assert not summary.exists()
|
|
|
|
|
|
def test_canary_is_found_inside_nested_export_archive(tmp_path):
|
|
layer = io.BytesIO()
|
|
with tarfile.open(fileobj=layer, mode="w") as archive:
|
|
payload = image_release.CANARY
|
|
member = tarfile.TarInfo("app/data")
|
|
member.size = len(payload)
|
|
archive.addfile(member, io.BytesIO(payload))
|
|
|
|
saved = tmp_path / "image.tar"
|
|
with tarfile.open(saved, mode="w") as archive:
|
|
payload = layer.getvalue()
|
|
member = tarfile.TarInfo("layer.tar")
|
|
member.size = len(payload)
|
|
archive.addfile(member, io.BytesIO(payload))
|
|
|
|
with pytest.raises(ValueError, match="canary marker found"):
|
|
image_release._scan_tar(saved)
|
|
|
|
|
|
def test_canary_is_found_inside_compressed_digest_named_blob(tmp_path):
|
|
inner = io.BytesIO()
|
|
with tarfile.open(fileobj=inner, mode="w") as archive:
|
|
payload = image_release.CANARY
|
|
member = tarfile.TarInfo("app/data")
|
|
member.size = len(payload)
|
|
archive.addfile(member, io.BytesIO(payload))
|
|
|
|
saved = tmp_path / "oci-image.tar"
|
|
with tarfile.open(saved, mode="w") as archive:
|
|
payload = gzip.compress(inner.getvalue())
|
|
member = tarfile.TarInfo("blobs/sha256/deadbeef")
|
|
member.size = len(payload)
|
|
archive.addfile(member, io.BytesIO(payload))
|
|
|
|
with pytest.raises(ValueError, match="canary marker found"):
|
|
image_release._scan_tar(saved)
|
|
|
|
|
|
def test_canary_is_found_in_symlink_metadata_inside_compressed_digest_blob(tmp_path):
|
|
inner = io.BytesIO()
|
|
with tarfile.open(fileobj=inner, mode="w") as archive:
|
|
member = tarfile.TarInfo("app/link")
|
|
member.type = tarfile.SYMTYPE
|
|
member.linkname = image_release.CANARY.decode()
|
|
archive.addfile(member)
|
|
|
|
saved = tmp_path / "oci-image.tar"
|
|
with tarfile.open(saved, mode="w") as archive:
|
|
payload = gzip.compress(inner.getvalue())
|
|
member = tarfile.TarInfo("blobs/sha256/deadbeef")
|
|
member.size = len(payload)
|
|
archive.addfile(member, io.BytesIO(payload))
|
|
|
|
with pytest.raises(ValueError, match="canary marker found"):
|
|
image_release._scan_tar(saved)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"pax_headers",
|
|
[
|
|
{"comment": image_release.CANARY.decode()},
|
|
{image_release.CANARY.decode(): "safe"},
|
|
],
|
|
)
|
|
def test_canary_is_found_in_pax_metadata_inside_compressed_digest_blob(tmp_path, pax_headers):
|
|
inner = io.BytesIO()
|
|
with tarfile.open(fileobj=inner, mode="w", format=tarfile.PAX_FORMAT) as archive:
|
|
member = tarfile.TarInfo("app/data")
|
|
member.pax_headers = pax_headers
|
|
archive.addfile(member)
|
|
|
|
saved = tmp_path / "oci-image.tar"
|
|
with tarfile.open(saved, mode="w") as archive:
|
|
payload = gzip.compress(inner.getvalue())
|
|
member = tarfile.TarInfo("blobs/sha256/deadbeef")
|
|
member.size = len(payload)
|
|
archive.addfile(member, io.BytesIO(payload))
|
|
|
|
with pytest.raises(ValueError, match="canary marker found"):
|
|
image_release._scan_tar(saved)
|
|
|
|
|
|
@pytest.mark.parametrize("field", ["uname", "gname"])
|
|
def test_canary_is_found_in_owner_metadata_inside_compressed_digest_blob(tmp_path, field):
|
|
inner = io.BytesIO()
|
|
with tarfile.open(fileobj=inner, mode="w") as archive:
|
|
member = tarfile.TarInfo("app/data")
|
|
setattr(member, field, image_release.CANARY.decode())
|
|
archive.addfile(member)
|
|
|
|
saved = tmp_path / "oci-image.tar"
|
|
with tarfile.open(saved, mode="w") as archive:
|
|
payload = gzip.compress(inner.getvalue())
|
|
member = tarfile.TarInfo("blobs/sha256/deadbeef")
|
|
member.size = len(payload)
|
|
archive.addfile(member, io.BytesIO(payload))
|
|
|
|
with pytest.raises(ValueError, match="canary marker found"):
|
|
image_release._scan_tar(saved)
|
|
|
|
|
|
def test_archive_scan_rejects_excessive_nesting_depth(tmp_path):
|
|
payload = io.BytesIO()
|
|
with tarfile.open(fileobj=payload, mode="w"):
|
|
pass
|
|
nested = payload.getvalue()
|
|
for _ in range(4):
|
|
wrapped = io.BytesIO()
|
|
with tarfile.open(fileobj=wrapped, mode="w") as archive:
|
|
member = tarfile.TarInfo("blob")
|
|
member.size = len(nested)
|
|
archive.addfile(member, io.BytesIO(nested))
|
|
nested = wrapped.getvalue()
|
|
path = tmp_path / "deep.tar"
|
|
path.write_bytes(nested)
|
|
|
|
with pytest.raises(ValueError, match="archive scan resource limit exceeded") as error:
|
|
image_release._scan_tar(path, image_release._ScanBudget(max_depth=2))
|
|
assert image_release.CANARY.decode() not in str(error.value)
|
|
|
|
|
|
def test_archive_scan_rejects_excessive_aggregate_members(tmp_path):
|
|
path = tmp_path / "members.tar"
|
|
with tarfile.open(path, mode="w") as archive:
|
|
for index in range(3):
|
|
member = tarfile.TarInfo(f"directory-{index}")
|
|
member.type = tarfile.DIRTYPE
|
|
archive.addfile(member)
|
|
|
|
with pytest.raises(ValueError, match="archive scan resource limit exceeded"):
|
|
image_release._scan_tar(path, image_release._ScanBudget(max_members=2))
|
|
|
|
|
|
def test_archive_scan_rejects_excessive_aggregate_extracted_bytes(tmp_path):
|
|
path = tmp_path / "aggregate.tar"
|
|
with tarfile.open(path, mode="w") as archive:
|
|
payload = b"123456"
|
|
member = tarfile.TarInfo("payload")
|
|
member.size = len(payload)
|
|
archive.addfile(member, io.BytesIO(payload))
|
|
|
|
budget = image_release._ScanBudget(max_extracted_bytes=5, max_member_bytes=10)
|
|
with pytest.raises(ValueError, match="archive scan resource limit exceeded"):
|
|
image_release._scan_tar(path, budget)
|
|
|
|
|
|
def test_archive_scan_rejects_excessive_per_member_bytes(tmp_path):
|
|
path = tmp_path / "member.tar"
|
|
with tarfile.open(path, mode="w") as archive:
|
|
payload = b"123456"
|
|
member = tarfile.TarInfo("payload")
|
|
member.size = len(payload)
|
|
archive.addfile(member, io.BytesIO(payload))
|
|
|
|
budget = image_release._ScanBudget(max_extracted_bytes=100, max_member_bytes=5)
|
|
with pytest.raises(ValueError, match="archive scan resource limit exceeded"):
|
|
image_release._scan_tar(path, budget)
|
|
|
|
|
|
def test_archive_scan_enforces_top_level_input_bytes_and_accepts_boundary(tmp_path):
|
|
path = tmp_path / "input.tar"
|
|
with tarfile.open(path, mode="w"):
|
|
pass
|
|
input_size = path.stat().st_size
|
|
|
|
image_release._scan_tar(path, image_release._ScanBudget(max_input_bytes=input_size))
|
|
with pytest.raises(ValueError, match="archive scan resource limit exceeded"):
|
|
image_release._scan_tar(path, image_release._ScanBudget(max_input_bytes=input_size - 1))
|
|
|
|
|
|
def test_canary_failure_message_does_not_disclose_marker(tmp_path):
|
|
output = tmp_path / "captured-output"
|
|
output.write_bytes(image_release.CANARY)
|
|
with pytest.raises(ValueError) as error:
|
|
image_release.assert_clean_output([output])
|
|
assert image_release.CANARY.decode() not in str(error.value)
|
|
|
|
|
|
def test_canary_scan_checks_metadata_filesystem_archive_and_cleanup(tmp_path):
|
|
class ScanRunner(RecordingRunner):
|
|
def run(self, command, *, input_text=None, timeout=None):
|
|
result = super().run(command, input_text=input_text, timeout=timeout)
|
|
if list(command)[:2] == ["docker", "export"]:
|
|
output = Path(command[command.index("--output") + 1])
|
|
with tarfile.open(output, mode="w"):
|
|
pass
|
|
return result
|
|
|
|
saved = tmp_path / "image.tar"
|
|
with tarfile.open(saved, mode="w"):
|
|
pass
|
|
runner = ScanRunner()
|
|
|
|
image_release.scan_image("candidate", saved, runner)
|
|
|
|
commands = [call[0] for call in runner.commands]
|
|
assert ["docker", "history", "--no-trunc", "candidate"] in commands
|
|
assert ["docker", "image", "inspect", "candidate"] in commands
|
|
assert ["docker", "create", "--network", "none", "candidate"] in commands
|
|
assert any(command[:2] == ["docker", "export"] for command in commands)
|
|
assert ["docker", "rm", "--force", "container-id"] in commands
|
|
|
|
|
|
def test_canary_in_create_stderr_still_cleans_container(tmp_path):
|
|
class LeakingCreateRunner(RecordingRunner):
|
|
def run(self, command, *, input_text=None, timeout=None):
|
|
result = super().run(command, input_text=input_text, timeout=timeout)
|
|
if list(command)[:2] == ["docker", "create"]:
|
|
return image_release.Result("container-id\n", image_release.CANARY.decode())
|
|
return result
|
|
|
|
saved = tmp_path / "image.tar"
|
|
with tarfile.open(saved, mode="w"):
|
|
pass
|
|
runner = LeakingCreateRunner()
|
|
|
|
with pytest.raises(ValueError, match="canary marker found"):
|
|
image_release.scan_image("candidate", saved, runner)
|
|
assert runner.commands[-1][0] == ["docker", "rm", "--force", "container-id"]
|