retire GHCR image publication
This commit is contained in:
Hermes Agent
@@ -1,216 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
WORKFLOW_PATH = REPO / ".github" / "workflows" / "publish-image.yml"
|
||||
|
||||
|
||||
def _workflow():
|
||||
return yaml.load(WORKFLOW_PATH.read_text(), Loader=yaml.BaseLoader)
|
||||
|
||||
|
||||
def _step_index(steps, name):
|
||||
return next(index for index, step in enumerate(steps) if step.get("name") == name)
|
||||
|
||||
|
||||
def test_workflow_triggers_only_v_tag_push_candidates():
|
||||
workflow = _workflow()
|
||||
assert set(workflow["on"]) == {"push"}
|
||||
assert workflow["on"]["push"] == {"tags": ["v*"]}
|
||||
|
||||
|
||||
def test_workflow_rejects_non_strict_version_events_before_checkout_and_build():
|
||||
steps = _workflow()["jobs"]["verify"]["steps"]
|
||||
guard = steps[0]
|
||||
assert guard["name"] == "Reject non-version publication events"
|
||||
assert guard["env"] == {
|
||||
"PUBLICATION_EVENT": "${{ github.event_name }}",
|
||||
"PUBLICATION_REF": "${{ github.ref }}",
|
||||
}
|
||||
assert '"${PUBLICATION_EVENT}" != "push"' in guard["run"]
|
||||
strict_ref = (
|
||||
r"^refs/tags/v(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)$"
|
||||
)
|
||||
assert strict_ref in guard["run"]
|
||||
assert _step_index(steps, guard["name"]) < _step_index(steps, "Check out exact source commit")
|
||||
assert _step_index(steps, guard["name"]) < _step_index(steps, "Build source image once")
|
||||
|
||||
|
||||
def test_workflow_reverifies_exact_checked_out_commit_before_repository_or_registry_work():
|
||||
workflow = _workflow()
|
||||
verify_steps = workflow["jobs"]["verify"]["steps"]
|
||||
publish_steps = workflow["jobs"]["publish"]["steps"]
|
||||
checkout_index = _step_index(verify_steps, "Check out exact source commit")
|
||||
reverify_index = _step_index(verify_steps, "Reverify exact checked out commit")
|
||||
|
||||
assert checkout_index < reverify_index < _step_index(verify_steps, "Install repository checks")
|
||||
assert verify_steps[reverify_index]["run"] == (
|
||||
'python scripts/image_release.py verify-checkout --commit "${GITHUB_SHA}"'
|
||||
)
|
||||
assert reverify_index < _step_index(verify_steps, "Build source image once")
|
||||
assert all("REGISTRY_TOKEN" not in str(step) for step in verify_steps[: reverify_index + 1])
|
||||
assert _step_index(publish_steps, "Verify, load, and publish without rebuilding") > 0
|
||||
|
||||
|
||||
def test_publish_is_gated_and_has_the_only_package_write_permission():
|
||||
workflow = _workflow()
|
||||
verify = workflow["jobs"]["verify"]
|
||||
publish = workflow["jobs"]["publish"]
|
||||
|
||||
assert publish["needs"] == "verify"
|
||||
assert publish["permissions"]["packages"] == "write"
|
||||
assert "packages" not in workflow["permissions"]
|
||||
assert "packages" not in verify["permissions"]
|
||||
|
||||
|
||||
def test_publish_job_exposes_selected_version_tag_with_other_release_evidence():
|
||||
outputs = _workflow()["jobs"]["publish"]["outputs"]
|
||||
assert outputs == {
|
||||
"image": "${{ steps.publish.outputs.image }}",
|
||||
"digest": "${{ steps.publish.outputs.digest }}",
|
||||
"source_commit": "${{ steps.publish.outputs.source_commit }}",
|
||||
"version_tag": "${{ steps.publish.outputs.version_tag }}",
|
||||
}
|
||||
|
||||
|
||||
def test_verify_canary_build_smoke_export_scan_checksum_upload_contract():
|
||||
steps = _workflow()["jobs"]["verify"]["steps"]
|
||||
names = [step["name"] for step in steps]
|
||||
assert names.count("Create excluded synthetic canary") == 1
|
||||
assert names.count("Build source image once") == 1
|
||||
assert names.count("Scan all canary surfaces") == 1
|
||||
|
||||
canary = steps[_step_index(steps, "Create excluded synthetic canary")]["run"]
|
||||
assert 'Path(".env.ci-canary")' in canary
|
||||
assert "TREX_SECRET_CANARY_DO_NOT_SHIP" in canary
|
||||
|
||||
scan = steps[_step_index(steps, "Scan all canary surfaces")]["run"]
|
||||
assert "image_release.py scan" in scan
|
||||
status_contract = '''>canary-scan.stdout 2>canary-scan.stderr
|
||||
scan_status=$?
|
||||
set -e
|
||||
python scripts/image_release.py assert-clean-output \\
|
||||
canary-scan.stdout canary-scan.stderr
|
||||
exit "${scan_status}"'''
|
||||
assert status_contract in scan
|
||||
assert scan.count("scan_status=$?") == 1
|
||||
assert scan.rstrip().endswith('exit "${scan_status}"')
|
||||
|
||||
assert _step_index(steps, "Run repository lint") < _step_index(steps, "Build source image once")
|
||||
assert _step_index(steps, "Run repository tests") < _step_index(steps, "Build source image once")
|
||||
ordered = [
|
||||
"Create excluded synthetic canary",
|
||||
"Build source image once",
|
||||
"Smoke built image",
|
||||
"Export tested image",
|
||||
"Scan all canary surfaces",
|
||||
"Record tested archive checksum",
|
||||
"Upload immutable tested image artifact",
|
||||
]
|
||||
assert [_step_index(steps, name) for name in ordered] == sorted(
|
||||
_step_index(steps, name) for name in ordered
|
||||
)
|
||||
|
||||
|
||||
def test_repository_checks_use_the_locked_uv_environment():
|
||||
steps = _workflow()["jobs"]["verify"]["steps"]
|
||||
install = steps[_step_index(steps, "Install repository checks")]["run"]
|
||||
lint = steps[_step_index(steps, "Run repository lint")]["run"]
|
||||
tests = steps[_step_index(steps, "Run repository tests")]["run"]
|
||||
|
||||
assert 'python -m pip install "uv==0.11.14"' in install
|
||||
assert "uv sync --extra dev --locked" in install
|
||||
assert lint == "uv run ruff check ."
|
||||
assert tests == "uv run pytest"
|
||||
assert 'pip install ".[dev]"' not in install
|
||||
|
||||
|
||||
def test_tested_image_id_is_recorded_after_smoke_and_transferred_to_publish():
|
||||
workflow = _workflow()
|
||||
verify_steps = workflow["jobs"]["verify"]["steps"]
|
||||
publish_steps = workflow["jobs"]["publish"]["steps"]
|
||||
|
||||
smoke_index = _step_index(verify_steps, "Smoke built image")
|
||||
identity_index = _step_index(verify_steps, "Record tested image ID")
|
||||
export_index = _step_index(verify_steps, "Export tested image")
|
||||
assert smoke_index < identity_index < export_index
|
||||
|
||||
identity = verify_steps[identity_index]["run"]
|
||||
assert "image_release.py record-image-id" in identity
|
||||
assert "--output trex-core-image.id" in identity
|
||||
|
||||
upload = verify_steps[_step_index(verify_steps, "Upload immutable tested image artifact")]["with"]
|
||||
assert "trex-core-image.id" in upload["path"].splitlines()
|
||||
|
||||
publish = publish_steps[_step_index(publish_steps, "Verify, load, and publish without rebuilding")][
|
||||
"run"
|
||||
]
|
||||
assert "--image-id tested-image/trex-core-image.id" in publish
|
||||
|
||||
|
||||
def test_publish_job_has_only_download_then_verified_publish_helper():
|
||||
steps = _workflow()["jobs"]["publish"]["steps"]
|
||||
assert len(steps) == 2
|
||||
assert "run" not in steps[0]
|
||||
assert steps[0]["uses"].startswith("actions/download-artifact@")
|
||||
|
||||
run_steps = [step for step in steps if "run" in step]
|
||||
assert run_steps == [steps[1]]
|
||||
expected_run = '''python tested-image/scripts/image_release.py publish \\
|
||||
--archive tested-image/trex-core-image.tar \\
|
||||
--checksum tested-image/trex-core-image.tar.sha256 \\
|
||||
--image-id tested-image/trex-core-image.id \\
|
||||
--local-image "trex-core-ci:${GITHUB_SHA}" \\
|
||||
--commit "${GITHUB_SHA}" \\
|
||||
--event-name "${GITHUB_EVENT_NAME}" \\
|
||||
--ref "${GITHUB_REF}"'''
|
||||
assert steps[1]["run"].strip() == expected_run
|
||||
|
||||
rendered = "\n".join(str(step) for step in steps).lower()
|
||||
for forbidden in (
|
||||
"actions/checkout",
|
||||
"docker login",
|
||||
"docker push",
|
||||
"docker build",
|
||||
"buildx",
|
||||
):
|
||||
assert forbidden not in rendered
|
||||
|
||||
|
||||
def test_workflow_uses_full_sha_and_contains_no_latest_tag():
|
||||
text = WORKFLOW_PATH.read_text()
|
||||
assert "${GITHUB_SHA}" in text
|
||||
assert "github.sha" in text
|
||||
assert "--provenance=false" in text
|
||||
assert ":latest" not in text.lower()
|
||||
|
||||
|
||||
def test_docker_context_contract():
|
||||
dockerfile = (REPO / "Dockerfile").read_text()
|
||||
copy_lines = [line.strip() for line in dockerfile.splitlines() if line.strip().startswith("COPY ")]
|
||||
assert copy_lines == ["COPY pyproject.toml uv.lock ./", "COPY trex_core/ ./trex_core/"]
|
||||
assert 'python -m pip install --no-cache-dir "uv==0.11.14"' in dockerfile
|
||||
assert "uv sync --locked --no-dev --no-editable" in dockerfile
|
||||
assert "pip install --no-cache-dir ." not in dockerfile
|
||||
assert "org.opencontainers.image.revision" in dockerfile
|
||||
assert "python:3.12" in dockerfile
|
||||
|
||||
ignored = set((REPO / ".dockerignore").read_text().splitlines())
|
||||
assert {".git", ".env", ".env.*", "config/", ".proofline/"} <= ignored
|
||||
|
||||
|
||||
def test_operator_documentation_preserves_rollback_boundaries():
|
||||
operations = (REPO / "docs" / "operations.md").read_text()
|
||||
assert "strict `vX.Y.Z` Git tag" in operations
|
||||
assert "사용자 승인 외부 쓰기" in operations
|
||||
assert "Git tag version은 `pyproject.toml` package version과 독립" in operations
|
||||
assert "read-only volume" in operations
|
||||
assert "runtime environment" in operations
|
||||
assert "sha256:<64-hex>" in operations
|
||||
assert "이전" in operations and "digest" in operations
|
||||
assert "`trex-core-reset`" in operations
|
||||
assert "자동 rollback 절차에 절대로 포함하지 않는다" in operations
|
||||
@@ -1,687 +0,0 @@
|
||||
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 == ["git", "rev-parse", "HEAD"]:
|
||||
return image_release.Result(self.revision + "\n")
|
||||
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():
|
||||
release = image_release.release_tags(COMMIT, "push", "refs/tags/v1.2.3")
|
||||
|
||||
assert release == [
|
||||
f"{image_release.IMAGE}:{COMMIT[:12]}",
|
||||
f"{image_release.IMAGE}:v1.2.3",
|
||||
]
|
||||
assert all(not tag.endswith(":latest") for tag in release)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("ref", ["refs/tags/v0.0.0", "refs/tags/v10.20.30"])
|
||||
def test_strict_version_tags_accept_zero_and_nonzero_decimal_components(ref):
|
||||
assert image_release.validate_publication_trigger("push", ref) == ref.removeprefix(
|
||||
"refs/tags/"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("event_name", "ref"),
|
||||
[
|
||||
("workflow_dispatch", "refs/tags/v1.2.3"),
|
||||
("pull_request", "refs/tags/v1.2.3"),
|
||||
("push", "refs/heads/main"),
|
||||
("push", "refs/heads/release"),
|
||||
("push", "refs/tags/1.2.3"),
|
||||
("push", "refs/tags/v1.2"),
|
||||
("push", "refs/tags/v1.2.3.4"),
|
||||
("push", "refs/tags/v01.2.3"),
|
||||
("push", "refs/tags/v1.02.3"),
|
||||
("push", "refs/tags/v1.2.03"),
|
||||
("push", "refs/tags/v1.2.3-rc.1"),
|
||||
("push", "refs/tags/v1.2.3+build.1"),
|
||||
("push", "refs/tags/v-1.2.3"),
|
||||
("push", "refs/tags/v1.a.3"),
|
||||
],
|
||||
)
|
||||
def test_non_push_branch_and_malformed_version_triggers_are_rejected(event_name, ref):
|
||||
with pytest.raises(ValueError):
|
||||
image_release.validate_publication_trigger(event_name, ref)
|
||||
|
||||
|
||||
@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, "push", "refs/tags/v1.2.3")
|
||||
|
||||
|
||||
def test_verify_checkout_requires_exact_git_head_match():
|
||||
runner = RecordingRunner()
|
||||
|
||||
assert image_release.verify_checkout(COMMIT, runner) == COMMIT
|
||||
assert runner.commands == [(["git", "rev-parse", "HEAD"], None, None)]
|
||||
|
||||
|
||||
def test_verify_checkout_mismatch_fails_closed():
|
||||
runner = RecordingRunner(revision="c" * 40)
|
||||
|
||||
with pytest.raises(ValueError, match="checked out HEAD does not match source commit"):
|
||||
image_release.verify_checkout(COMMIT, runner)
|
||||
assert runner.commands == [(["git", "rev-parse", "HEAD"], None, None)]
|
||||
|
||||
|
||||
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,
|
||||
event_name="push",
|
||||
ref="refs/tags/v1.2.3",
|
||||
actor="actor",
|
||||
token="fake-token",
|
||||
runner=runner,
|
||||
)
|
||||
assert runner.commands == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("event_name", "ref"),
|
||||
[
|
||||
("workflow_dispatch", "refs/tags/v1.2.3"),
|
||||
("push", "refs/heads/main"),
|
||||
("push", "refs/tags/v01.2.3"),
|
||||
("push", "refs/tags/v1.2.3-rc.1"),
|
||||
],
|
||||
)
|
||||
def test_invalid_publication_trigger_stops_before_artifact_load_or_registry(
|
||||
tmp_path, event_name, ref
|
||||
):
|
||||
runner = RecordingRunner()
|
||||
|
||||
with pytest.raises(ValueError, match="publication"):
|
||||
image_release.publish_image(
|
||||
archive=tmp_path / "missing.tar",
|
||||
checksum=tmp_path / "missing.sha256",
|
||||
image_id=tmp_path / "missing.id",
|
||||
local_image=f"trex-core-ci:{COMMIT}",
|
||||
commit=COMMIT,
|
||||
event_name=event_name,
|
||||
ref=ref,
|
||||
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,
|
||||
event_name="push",
|
||||
ref="refs/tags/v1.2.3",
|
||||
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,
|
||||
event_name="push",
|
||||
ref="refs/tags/v1.2.3",
|
||||
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,
|
||||
event_name="push",
|
||||
ref="refs/tags/v1.2.3",
|
||||
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,
|
||||
event_name="push",
|
||||
ref="refs/tags/v1.2.3",
|
||||
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,
|
||||
event_name="push",
|
||||
ref="refs/tags/v1.2.3",
|
||||
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_exact_release_evidence_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,
|
||||
event_name="push",
|
||||
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 output.read_text() == (
|
||||
f"image={image_release.IMAGE}@{DIGEST}\n"
|
||||
f"digest={DIGEST}\n"
|
||||
f"source_commit={COMMIT}\n"
|
||||
"version_tag=v2.0.0\n"
|
||||
)
|
||||
assert summary.read_text() == (
|
||||
"## Published trex-core image\n\n"
|
||||
f"- Image: `{image_release.IMAGE}@{DIGEST}`\n"
|
||||
f"- Digest: `{DIGEST}`\n"
|
||||
f"- Source commit: `{COMMIT}`\n"
|
||||
"- Version tag: `v2.0.0`\n"
|
||||
)
|
||||
|
||||
|
||||
@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,
|
||||
event_name="push",
|
||||
ref="refs/tags/v1.2.3",
|
||||
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,
|
||||
event_name="push",
|
||||
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"]
|
||||
Reference in New Issue
Block a user