fix: parse literal amd64 pinned journey image bases

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)

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._:/-]*@(?P<digest>sha256:[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))

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

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,