diff --git a/src/echo_certification_forge/journey_image_admission.py b/src/echo_certification_forge/journey_image_admission.py index c4b99e1..c18db9f 100644 --- a/src/echo_certification_forge/journey_image_admission.py +++ b/src/echo_certification_forge/journey_image_admission.py @@ -23,6 +23,7 @@ ImageIdentity, ImageRole, evaluate_image_admission, + pinned_dockerfile_base_digests, scan_dockerfile, verify_spdx_document, ) @@ -154,8 +155,7 @@ def prepare_admission_request( scan = scan_dockerfile(dockerfile.decode("utf-8")) if not scan.valid: raise ImageAdmissionDenied("dockerfile_policy_failed") - bases = [line.split()[1].split("@", 1)[1] for line in dockerfile.decode("utf-8").splitlines() - if line.strip().upper().startswith("FROM ")] + bases = pinned_dockerfile_base_digests(dockerfile.decode("utf-8")) if identity.base_image_digest != bases[-1]: raise ImageAdmissionDenied("final_base_mismatch") lock = json.loads(lockfile) diff --git a/src/echo_certification_forge/supply_chain.py b/src/echo_certification_forge/supply_chain.py index 573a7ce..35ee8a9 100644 --- a/src/echo_certification_forge/supply_chain.py +++ b/src/echo_certification_forge/supply_chain.py @@ -19,7 +19,11 @@ _IMAGE_DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$") _COMMIT = re.compile(r"^[0-9a-f]{40}$") -_PINNED_FROM = re.compile(r"^FROM\s+[^\s]+@sha256:[0-9a-f]{64}(?:\s+AS\s+[A-Za-z0-9_.-]+)?$", re.IGNORECASE) +_PINNED_FROM = re.compile( + r"^(?i:FROM)\s+(?:--platform=linux/amd64\s+)?" + r"[A-Za-z0-9][A-Za-z0-9._:/-]*@(?Psha256:[0-9a-f]{64})" + r"(?:\s+(?i:AS)\s+[A-Za-z0-9_.-]+)?$" +) class ImageRole(StrEnum): @@ -273,8 +277,8 @@ class DockerfileScanResult(BaseModel): sha256: str -def scan_dockerfile(text: str) -> DockerfileScanResult: - findings: list[DockerfileFinding] = [] +def _dockerfile_instructions(text: str) -> list[tuple[int, str]]: + """Retain original line numbers while joining Dockerfile continuations.""" logical_lines: list[tuple[int, str]] = [] pending = "" pending_line = 0 @@ -290,13 +294,39 @@ def scan_dockerfile(text: str) -> DockerfileScanResult: pending = "" if pending: logical_lines.append((pending_line, pending.strip())) + return logical_lines + + +def pinned_dockerfile_base_digests(text: str) -> tuple[str, ...]: + """Parse immutable FROM pins; only the literal linux/amd64 option is admitted.""" + bases: list[str] = [] + for _, line in _dockerfile_instructions(text): + if not line: + continue + if line.split(maxsplit=1)[0].upper() != "FROM": + continue + match = _PINNED_FROM.fullmatch(line) + if match is None: + raise ImageAdmissionDenied("dockerfile_base_not_pinned") + bases.append(match.group("digest")) + if not bases: + raise ImageAdmissionDenied("dockerfile_base_missing") + return tuple(bases) + + +def scan_dockerfile(text: str) -> DockerfileScanResult: + findings: list[DockerfileFinding] = [] from_count = 0 - for line_number, line in logical_lines: + for line_number, line in _dockerfile_instructions(text): + if not line: + continue upper = line.upper() - if upper.startswith("FROM "): + if line.split(maxsplit=1)[0].upper() == "FROM": from_count += 1 - if not _PINNED_FROM.fullmatch(line): + try: + pinned_dockerfile_base_digests(line) + except ImageAdmissionDenied: findings.append(DockerfileFinding(line=line_number, rule="unpinned_base", detail=line)) if upper.startswith("ADD "): findings.append(DockerfileFinding(line=line_number, rule="add_forbidden", detail=line)) diff --git a/tests/test_journey_image_admission.py b/tests/test_journey_image_admission.py index 323cecc..94d3997 100644 --- a/tests/test_journey_image_admission.py +++ b/tests/test_journey_image_admission.py @@ -86,6 +86,23 @@ def test_owner_seal_retains_trust_and_does_not_mutate_old_policy(prepared): assert kwargs['prior_policy'].model_dump_json() == before +def test_admission_extracts_final_amd64_multistage_pin_without_recipe_rewriting(prepared): + kwargs, _ = prepared + kwargs['dockerfile'] = ( + 'FROM --platform=linux/amd64 node@sha256:' + '6' * 64 + ' AS node_runtime\n' + 'FROM --platform=linux/amd64 python@sha256:' + '4' * 64 + '\n' + 'USER 65534:65534\n' + ).encode() + kwargs['identity'] = kwargs['identity'].model_copy(update={ + 'dockerfile_sha256': hashlib.sha256(kwargs['dockerfile']).hexdigest()}) + kwargs['qualification'] = kwargs['qualification'].model_copy(update={ + 'identity_digest': kwargs['identity'].digest}) + request = prepare_admission_request(**kwargs) + assert request.identity.base_image_digest == 'sha256:' + '4' * 64 + assert request.identity.dockerfile_sha256 == hashlib.sha256(kwargs['dockerfile']).hexdigest() + assert request.state == 'UNSIGNED_NOT_ADMITTED' + + @pytest.mark.parametrize('field',['dockerfile','lockfile','sbom','provenance']) def test_material_drift_rejected(prepared,field): kwargs,_=prepared diff --git a/tests/test_supply_chain.py b/tests/test_supply_chain.py index 00c3c42..b868158 100644 --- a/tests/test_supply_chain.py +++ b/tests/test_supply_chain.py @@ -13,12 +13,14 @@ ImageAttestationAuthority, ImageIdentity, ImageRole, + ImageAdmissionDenied, build_spdx_document, container_link_escapes, contains_private_key_material, evaluate_image_admission, scan_build_context, scan_dockerfile, + pinned_dockerfile_base_digests, verify_image_attestation, verify_spdx_document, ) @@ -27,12 +29,69 @@ IMAGE = "sha256:" + "1" * 64 BASE = "sha256:" + "2" * 64 SOURCE = "3" * 40 +PAIRED_DOCKERFILE = ( + "FROM --platform=linux/amd64 node:24.22.1-alpine3.23@sha256:" + "47019ffab12a48f49795a8279603f87e91a64de017fb9aa874e64f489b686079 AS node_runtime\n" + "FROM --platform=linux/amd64 python:3.12.16-alpine3.23@sha256:" + "fe32c943235f1c5748d1106dbdfa25d206355c1be1ae1e167444a7ff9dfa9422\n" + "USER 65534:65534\n" +) DOCKERFILE = "4" * 64 LOCK = "5" * 64 SBOM = "6" * 64 PROVENANCE = "7" * 64 +def test_exact_amd64_multistage_bases_are_pinned_without_recipe_rewriting(): + result = scan_dockerfile(PAIRED_DOCKERFILE) + assert result.valid, result.findings + assert result.sha256 == sha256_bytes(PAIRED_DOCKERFILE.encode()) + assert pinned_dockerfile_base_digests(PAIRED_DOCKERFILE) == ( + "sha256:47019ffab12a48f49795a8279603f87e91a64de017fb9aa874e64f489b686079", + "sha256:fe32c943235f1c5748d1106dbdfa25d206355c1be1ae1e167444a7ff9dfa9422", + ) + + +@pytest.mark.parametrize("prefix", [ + "--platform=linux/arm64 ", "--platform=$TARGETPLATFORM ", + "--platform=${TARGETPLATFORM} ", "--platform=linux/AMD64 ", + "--PLATFORM=linux/amd64 ", "--platform linux/amd64 ", + "--platform=linux/amd64 --platform=linux/amd64 ", "--pull ", +]) +def test_unsupported_or_variable_from_options_fail_both_parsers(prefix): + text = f"FROM {prefix}python@{BASE}\nUSER 65534\n" + assert not scan_dockerfile(text).valid + with pytest.raises(ImageAdmissionDenied): + pinned_dockerfile_base_digests(text) + + +@pytest.mark.parametrize("reference", [ + "python:latest", "${BASE}", "${BASE}@" + BASE, "$BASE@" + BASE, + "python@sha256:" + "A" * 64, "python@" + BASE + " unexpected", +]) +def test_variable_unpinned_or_extra_from_material_fails_both_parsers(reference): + text = f"FROM --platform=linux/amd64 {reference}\nUSER 65534\n" + assert not scan_dockerfile(text).valid + with pytest.raises(ImageAdmissionDenied): + pinned_dockerfile_base_digests(text) + + +def test_base_parser_shares_continuation_and_case_handling(): + text = ("from --platform=linux/amd64 " + "\\" + "\n" + f" registry.example:5000/team/python:3.12@{BASE} aS runtime\nUSER 65534\n") + assert scan_dockerfile(text).valid + assert pinned_dockerfile_base_digests(text) == (BASE,) + + +@pytest.mark.parametrize("text", [ + "\\" + "\n", " \\" + "\n \\" + "\n", "\\" + "\n# comment only\n", +]) +def test_empty_continuation_instruction_is_missing_base_not_index_error(text): + assert not scan_dockerfile(text).valid + with pytest.raises(ImageAdmissionDenied, match="dockerfile_base_missing"): + pinned_dockerfile_base_digests(text) + + def identity(role: ImageRole = ImageRole.RUNNER, image_digest: str = IMAGE) -> ImageIdentity: return ImageIdentity( role=role,