platform-ci/platform_ci/image_packaging.py

"""Package OCI images from the outputs a component's own verification already produced, in the same job, without building the source again.

The image is part of the ONE verified bundle and manifest, not a second artifact protocol. A component that declares `images` runs, in the same
job and checkout: its native verification (once), `package-images` (this module), then the ordinary `package` step. Each image declares two
ordinary outputs (see `Component` in configuration.py): an artifact `image-<name>` of format `container` holding the single OCI archive, which is the
shape the existing OCI publisher primitive already consumes, and a report `image-<name>-evidence` holding the scan, SBOM and provenance files.
The existing `verified-<target>` upload, the build-input record, reuse and the publisher therefore need no change.

This module has no code path that starts a native build tool. Before BuildKit starts it:

1. runs the captured, receipted `image-boundary.sh` (rootless, no capabilities, no Docker socket, no credential, BuildKit reachable in sandbox mode);
2. hashes every file of the declared tested artifact in the checkout; that record (`tested-output.json`) is later required to equal the manifest the
   `package` step writes, so the image can only contain the tested bytes (`verify_image_binding`);
3. refuses a Dockerfile (or recipe) that would compile the source again, asks for an insecure BuildKit entitlement, or copies none of the tested output.
   A Dockerfile that recompiles does NOT satisfy build reuse and is rejected, never accepted as reuse.

It then builds with native BuildKit (`buildctl`), verifies the OCI layout, runs ONE Trivy image scan (HIGH and CRITICAL fail; every package is listed), the
unchanged captured `check-image-evidence.mjs` over that report, and converts the same report to the CycloneDX SBOM with `trivy convert`, so the image is
scanned once, not twice. Outputs are written to `ci-image-output/<name>/`
only after every check passed; a failure removes them. Nothing is pushed or signed and no credential is read.
"""

import fnmatch
import hashlib
import json
import re
import shutil
import tarfile
import tempfile
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from typing import Final, cast

from platform_ci.build_tools import BUILD_TOOLS, IMAGE_ARCHIVE_FILE, IMAGE_OUTPUT_DIRECTORY, IMAGE_RECIPE_RECORD, IMAGE_RECIPES
from platform_ci.caching import CacheError, trivy_cache_directory
from platform_ci.execution import CommandRunner
from platform_ci.packaging import tested_artifact_files
from platform_ci.plan import ImageTarget, Target
from platform_ci.workspace import PipelineError, resolve_component_directory

RECIPE_DIRECTORY: Final = Path(__file__).resolve().parent / "image_recipes"
HELPER_DIRECTORY: Final = Path(__file__).resolve().parent.parent / "image_checks"
EVIDENCE_PROGRAM: Final = "check-image-evidence.mjs"
BOUNDARY_PROGRAM: Final = "image-boundary.sh"
DIAGNOSTIC_DIRECTORY_NAME: Final = "ci-image-diagnostics"
TESTED_OUTPUT_FILE: Final = "tested-output.json"
BUILDKIT_PROGRAM: Final = BUILD_TOOLS["container"].executable
SCANNER_PROGRAM: Final = "trivy"
NODE_PROGRAM: Final = BUILD_TOOLS["source"].executable
SHELL_PROGRAM: Final = "bash"
PLATFORM: Final = "linux/amd64"

# Programs whose use in a Dockerfile RUN compiles, installs or tests source: the native tools of BUILD_TOOLS (minus the ones that only run
# the image build itself or a script) plus the wrappers and front ends the captured Dockerfiles used. A RUN of one of them is a rebuild.
_NOT_A_REBUILD: Final = frozenset({BUILDKIT_PROGRAM, NODE_PROGRAM, "python", ""})
RECOMPILE_PROGRAMS: Final = frozenset({PurePosixPath(tool.executable).name for tool in BUILD_TOOLS.values()} - _NOT_A_REBUILD) | {"mvnw", "gradle", "npx", "yarn", "pnpm", "javac", "tsc"}
_SAFE_PATH: Final = re.compile(r"[A-Za-z0-9_-][A-Za-z0-9._-]*(/[A-Za-z0-9_-][A-Za-z0-9._-]*)*")
_ENTITLEMENT: Final = re.compile(r"--(security=insecure|network=host)", re.IGNORECASE)
_COMMAND_SEPARATORS: Final = re.compile(r"&&|\|\||[;|\n(){}`]|\$\(")
_ASSIGNMENT: Final = re.compile(r"[A-Za-z_][A-Za-z0-9_]*=")
_PLACEHOLDER: Final = re.compile(r"@[A-Z_]+@")
_FULL_COMMIT: Final = re.compile(r"[0-9a-f]{40}")


class ImagePackagingError(PipelineError):
   """Raised when an image cannot be packaged from the verified output: nothing is left behind and nothing is built again."""


@dataclass(frozen=True)
class ImageRun:
   """The facts and processes of one packaging run: the revisions it is bound to and how it starts BuildKit, Trivy and Node."""

   source_revision: str
   platform_revision: str
   run_command: CommandRunner
   environment: Mapping[str, str]
   """The already allow-listed environment of the target's tool run."""


@dataclass(frozen=True)
class PackagedImage:
   """One packaged image: where its files are and the SHA-256 of its OCI archive."""

   name: str
   directory: Path
   archive_sha256: str


@dataclass(frozen=True)
class _Outputs:
   """The files one image's build, scan and SBOM produced, staged only once every check passed."""

   archive: Path
   trivy_report: Path
   sbom: Path
   source_revision: str


def package_images(target: Target, source_directory: Path, run: ImageRun) -> list[PackagedImage]:
   """Package every image of the target from the tested outputs in the checkout; the commands go through `run_command`, never a native build.

   The first failure removes everything written so far.
   """
   if not target["images"]:
      return []
   if _FULL_COMMIT.fullmatch(run.source_revision) is None:
      raise ImagePackagingError("the source revision must be a full commit id")
   component_directory = resolve_component_directory(source_directory, target["directory"])
   output_root = component_directory / IMAGE_OUTPUT_DIRECTORY
   if output_root.exists():
      raise ImagePackagingError(f"{IMAGE_OUTPUT_DIRECTORY} already exists, so the images would not be freshly packaged")
   _verify_helpers()
   if run.run_command([SHELL_PROGRAM, str(HELPER_DIRECTORY / BOUNDARY_PROGRAM)], component_directory, run.environment) != 0:
      raise ImagePackagingError("the image execution boundary check failed, so nothing was built")
   packaged: list[PackagedImage] = []
   try:
      for name, image in sorted(target["images"].items()):
         tested = _tested(target, name, image, source_directory, component_directory)
         packaged.append(_package_one(name, image, tested, output_root / name, (run, component_directory)))
   except BaseException:
      shutil.rmtree(output_root, ignore_errors=True)
      raise
   return packaged


def verify_image_binding(target: Target, source_directory: Path, manifest: Mapping[str, object]) -> None:
   """Require every image's recorded tested files and archive to equal what the `package` step just wrote into the manifest.

   This is the proof that the image came from the bytes the one native verification produced and that the bundle retains exactly that image.
   """
   component_directory = resolve_component_directory(source_directory, target["directory"])
   retained = {str(entry["path"]): str(entry["sha256"]) for entry in cast("list[dict[str, object]]", manifest["files"])}
   for name, image in sorted(target["images"].items()):
      directory = component_directory / IMAGE_OUTPUT_DIRECTORY / name
      try:
         record = cast("dict[str, object]", json.loads((directory / TESTED_OUTPUT_FILE).read_text(encoding="utf-8")))
         recorded = {str(item["path"]): str(item["sha256"]) for item in cast("list[dict[str, object]]", record["files"])}
      except (OSError, ValueError, KeyError, TypeError) as error:
         raise ImagePackagingError(f"image {name}: {TESTED_OUTPUT_FILE} is unreadable: {type(error).__name__}") from error
      prefix = f"artifact/{image['artifact']}/"
      in_bundle = {path.removeprefix(prefix): digest for path, digest in retained.items() if path.startswith(prefix)}
      if in_bundle != recorded:
         raise ImagePackagingError(f"image {name}: the bundle's tested files differ from the files the image was built from")
      archive_path = f"artifact/image-{name}/{IMAGE_OUTPUT_DIRECTORY}/{name}/{IMAGE_ARCHIVE_FILE}"
      if retained.get(archive_path) != _sha256(directory / IMAGE_ARCHIVE_FILE):
         raise ImagePackagingError(f"image {name}: the bundle does not retain the image archive that was scanned")


def _tested(target: Target, name: str, image: ImageTarget, source_directory: Path, component_directory: Path) -> list[tuple[str, str]]:
   """Return (component-relative path, SHA-256) of the tested artifact's files, each of which the build context must contain."""
   artifact = image["artifact"]
   if artifact not in target["artifacts"]:
      raise ImagePackagingError(f"image {name}: artifact {artifact!r} is not declared by the component")
   files = tested_artifact_files(target, source_directory, artifact)
   context = (component_directory / image["context"]).resolve()
   tested: list[tuple[str, str]] = []
   for relative, path in files:
      if path.is_symlink() or not path.resolve().is_relative_to(context):
         raise ImagePackagingError(f"image {name}: {relative} is a link or lies outside the build context {image['context']!r}")
      tested.append((relative, _sha256(path)))
   return tested


def _package_one(name: str, image: ImageTarget, tested: list[tuple[str, str]], destination: Path, context_run: tuple[ImageRun, Path]) -> PackagedImage:
   """Build, verify, scan and stage one image; the destination exists only if every step passed."""
   run, component_directory = context_run
   source_revision, platform_revision, run_command, environment = run.source_revision, run.platform_revision, run.run_command, run.environment
   context = _safe_directory(component_directory, image["context"], "context", name)
   with tempfile.TemporaryDirectory() as scratch_name:
      scratch = Path(scratch_name)
      dockerfile, record = _dockerfile(name, image, component_directory, scratch, platform_revision)
      _check_dockerfile(name, dockerfile.read_text(encoding="utf-8"), _context_relative([relative for relative, _digest in tested], component_directory, context))
      archive = scratch / IMAGE_ARCHIVE_FILE
      arguments = [BUILDKIT_PROGRAM, "build", "--frontend", "dockerfile.v0", "--local", f"context={context}", "--local", f"dockerfile={dockerfile.parent}", "--opt", f"filename={dockerfile.name}", "--opt", f"platform={PLATFORM}", "--progress", "plain", "--output", f"type=oci,dest={archive}"]
      if image["target"]:
         arguments += ["--opt", f"target={image['target']}"]
      if image["recipe"]:
         arguments += ["--opt", f"label:org.kof22.ci.recipe={image['recipe']}", "--opt", f"label:org.kof22.ci.recipe-sha256={record['rendered_dockerfile_sha256']}", "--opt", f"label:org.kof22.ci.engine={platform_revision}", "--opt", f"build-arg:GIT_COMMIT_SHA={source_revision}"]
      if run_command(arguments, component_directory, environment) != 0:
         raise ImagePackagingError(f"image {name}: buildctl build failed")
      oci = scratch / "oci"
      _verify_oci_archive(name, archive, oci)
      trivy_report, sbom = scratch / "trivy.json", scratch / "sbom.cdx.json"
      # ONE image scan: the blocking HIGH/CRITICAL verdict and, with every package listed, the report the SBOM is converted from. `--skip-db-update`
      # is never passed, so the scanner validates the vulnerability database's own freshness and updates it when stale, cache or not.
      scan = [
         SCANNER_PROGRAM,
         "image",
         *_scanner_cache_arguments(environment, component_directory),
         "--input",
         str(oci),
         "--scanners",
         "vuln",
         "--list-all-pkgs",
         "--severity",
         "HIGH,CRITICAL",
         *(["--ignore-unfixed"] if image["ignore_unfixed"] else []),
         "--exit-code",
         "1",
         "--format",
         "json",
         "--output",
         str(trivy_report),
      ]
      # `ignore_unfixed` is the one closed relaxation a repository can declare (findings with no available fix); there is no way to skip the scan
      if run_command(scan, component_directory, environment) != 0:
         _retain_diagnostics(component_directory / DIAGNOSTIC_DIRECTORY_NAME / name, trivy_report, source_revision)
         raise ImagePackagingError(f"image {name}: Trivy found blocking HIGH/CRITICAL findings or could not scan")
      if run_command([NODE_PROGRAM, str(HELPER_DIRECTORY / EVIDENCE_PROGRAM), str(oci), str(trivy_report)], component_directory, environment) != 0:
         raise ImagePackagingError(f"image {name}: the scan evidence does not bind to the image")
      # the SBOM is the native conversion of the scan's own report (the same image bytes, already bound to the image by the evidence check above)
      if run_command([SCANNER_PROGRAM, "convert", "--format", "cyclonedx", "--output", str(sbom), str(trivy_report)], component_directory, environment) != 0:
         raise ImagePackagingError(f"image {name}: SBOM conversion failed")
      _require_cyclonedx(name, sbom)
      provenance = {"image": name, "artifact": image["artifact"], "files": [{"path": relative, "sha256": digest} for relative, digest in tested]}
      return _stage(name, destination, _Outputs(archive, trivy_report, sbom, source_revision), (record, provenance))


def _scanner_cache_arguments(environment: Mapping[str, str], component_directory: Path) -> list[str]:
   """Return `--cache-dir` for the scanner's cache directory the workflow restores and saves, or nothing when it cannot be proven safe."""
   home = environment.get("HOME", "")
   if not home:
      return []
   try:
      return ["--cache-dir", trivy_cache_directory(Path(home), component_directory).as_posix()]
   except CacheError:
      return []


def _dockerfile(name: str, image: ImageTarget, component_directory: Path, scratch: Path, platform_revision: str) -> tuple[Path, dict[str, str]]:
   """Return the Dockerfile to build and the recipe record: the repository's own file, or a central recipe rendered outside the context."""
   if not image["recipe"]:
      path = component_directory / image["dockerfile"]
      _safe_directory(component_directory, image["dockerfile"], "dockerfile", name, is_file=True)
      return path, {}
   if image["recipe"] not in IMAGE_RECIPES:
      raise ImagePackagingError(f"image {name}: unknown recipe {image['recipe']!r}")
   template = (RECIPE_DIRECTORY / f"{image['recipe']}.Dockerfile").read_bytes()
   rendered = template.decode("utf-8").replace("@NODE_VERSION@", image["node_version"]).replace("@HEALTH_PORT@", image["health_port"])
   if _PLACEHOLDER.search(rendered):
      raise ImagePackagingError(f"image {name}: unresolved recipe placeholder")
   rendered_directory = scratch / "recipe"
   rendered_directory.mkdir()
   (rendered_directory / "Dockerfile").write_text(rendered, encoding="utf-8", newline="\n")
   record = {"recipe": image["recipe"], "template_sha256": hashlib.sha256(template).hexdigest(), "rendered_dockerfile_sha256": hashlib.sha256(rendered.encode("utf-8")).hexdigest(), "engine": platform_revision, "node_version": image["node_version"], "health_port": image["health_port"]}
   return rendered_directory / "Dockerfile", record


def _safe_directory(component_directory: Path, relative: str, kind: str, name: str, *, is_file: bool = False) -> Path:
   """Return a path that is a plain, link-free, hidden-segment-free directory (or file) inside the component, as the captured build requires."""
   if relative != "." and _SAFE_PATH.fullmatch(relative) is None:
      raise ImagePackagingError(f"image {name}: unsafe {kind} path {relative!r}")
   walked = component_directory
   for part in () if relative == "." else PurePosixPath(relative).parts:
      walked /= part
      if walked.is_symlink():
         raise ImagePackagingError(f"image {name}: symlink in the {kind} path {relative!r}")
   resolved = walked.resolve()
   if not resolved.is_relative_to(component_directory.resolve()) or not (resolved.is_file() if is_file else resolved.is_dir()):
      raise ImagePackagingError(f"image {name}: the {kind} {relative!r} is not a {'file' if is_file else 'directory'} inside the component")
   return walked


def _context_relative(relative_sources: list[str], component_directory: Path, context: Path) -> list[str]:
   """Return the tested artifact paths relative to the build context, the form a Dockerfile COPY names them in."""
   return [(component_directory / source).resolve().relative_to(context.resolve()).as_posix() for source in relative_sources]


def _check_dockerfile(name: str, text: str, tested_paths: list[str]) -> None:
   """Refuse a Dockerfile that asks for an insecure entitlement, rebuilds the source, or copies none of the tested output."""
   if _ENTITLEMENT.search(text):
      raise ImagePackagingError(f"image {name}: the Dockerfile requests an insecure entitlement")
   copies_tested_output = False
   for instruction, arguments in _instructions(text):
      if instruction == "RUN":
         for program in _run_programs(arguments):
            if program in RECOMPILE_PROGRAMS:
               raise ImagePackagingError(f"image {name}: the Dockerfile runs {program!r}, which builds the source again; it does not reuse the tested output, so it is not accepted")
      elif instruction in ("COPY", "ADD") and not any(token.startswith("--from=") for token in arguments.split()):
         sources = _copy_tokens(arguments)[:-1]
         copies_tested_output = copies_tested_output or any(_covers(source, tested_paths) for source in sources)
   if not copies_tested_output:
      raise ImagePackagingError(f"image {name}: the Dockerfile copies none of the tested artifact files, so the image would not contain the tested output")


def _instructions(text: str) -> list[tuple[str, str]]:
   """Split a Dockerfile into (instruction, arguments) pairs, joining backslash continuations and dropping comments."""
   lines: list[str] = []
   pending = ""
   for raw in text.splitlines():
      stripped = raw.strip()
      if stripped.startswith("#") or (not pending and not stripped):
         continue
      pending += stripped.removesuffix("\\") + " "
      if not stripped.endswith("\\"):
         lines.append(pending.strip())
         pending = ""
   if pending:
      lines.append(pending.strip())
   return [(line.split(None, 1)[0].upper(), line.split(None, 1)[1] if " " in line else "") for line in lines]


def _run_programs(arguments: str) -> list[str]:
   """Return the program that starts each command of a RUN line, shell or exec form, without a leading path or `./`."""
   if arguments.startswith("["):
      tokens = [token.strip(" \"'") for token in arguments.strip("[] ").split(",")]
      segments = [" ".join(tokens)]
   else:
      segments = _COMMAND_SEPARATORS.split(arguments)
   programs: list[str] = []
   for segment in segments:
      words = [word for word in segment.split() if _ASSIGNMENT.match(word) is None]
      if words:
         programs.append(PurePosixPath(words[0].strip("\"'")).name)
   return programs


def _copy_tokens(arguments: str) -> list[str]:
   """Return the non-flag tokens of a COPY or ADD line, shell or JSON form."""
   if arguments.startswith("["):
      return [token.strip(" \"'") for token in arguments.strip("[] ").split(",")]
   return [token for token in arguments.split() if not token.startswith("--")]


def _covers(source: str, tested_paths: list[str]) -> bool:
   """Return whether a COPY source names a tested file (by path or glob), a directory holding one, or the whole context."""
   normalized = PurePosixPath(source.removeprefix("./") or ".").as_posix().rstrip("/") or "."
   return normalized == "." or any(path == normalized or path.startswith(f"{normalized}/") or fnmatch.fnmatchcase(path, normalized) for path in tested_paths)


def _verify_oci_archive(name: str, archive: Path, oci: Path) -> None:
   """Require a non-empty, readable OCI layout (`oci-layout`, `index.json`, `blobs/`) and unpack it for the scanner."""
   if not archive.is_file() or archive.stat().st_size == 0:
      raise ImagePackagingError(f"image {name}: no OCI archive was produced")
   try:
      with tarfile.open(archive) as bundle:
         names = bundle.getnames()
         if "oci-layout" not in names or "index.json" not in names or not any(member.startswith("blobs/") for member in names):
            raise ImagePackagingError(f"image {name}: the archive is not an OCI layout (oci-layout, index.json and blobs/ are required)")
         oci.mkdir()
         bundle.extractall(oci, filter="data")
   except tarfile.TarError as error:
      raise ImagePackagingError(f"image {name}: the OCI archive is unreadable: {type(error).__name__}") from error


def _require_cyclonedx(name: str, sbom: Path) -> None:
   """Require a CycloneDX document, as the captured scan helper does."""
   try:
      bom = cast("dict[str, object]", json.loads(sbom.read_text(encoding="utf-8")))
   except (OSError, ValueError) as error:
      raise ImagePackagingError(f"image {name}: the SBOM is not valid JSON") from error
   if bom.get("bomFormat") != "CycloneDX" or ("components" in bom and not isinstance(bom["components"], list)):
      raise ImagePackagingError(f"image {name}: the SBOM is not a valid CycloneDX document")


def _stage(name: str, destination: Path, outputs: _Outputs, evidence: tuple[dict[str, str], Mapping[str, object]]) -> PackagedImage:
   """Write the image archive and its evidence beside each other, after every check passed."""
   record, provenance = evidence
   destination.mkdir(parents=True)
   shutil.copyfile(outputs.archive, destination / IMAGE_ARCHIVE_FILE)
   shutil.copyfile(outputs.trivy_report, destination / "trivy.json")
   shutil.copyfile(outputs.sbom, destination / "sbom.cdx.json")
   (destination / "SOURCE_COMMIT").write_text(f"{outputs.source_revision}\n", encoding="utf-8", newline="\n")
   (destination / TESTED_OUTPUT_FILE).write_text(json.dumps(provenance, indent=2, sort_keys=True) + "\n", encoding="utf-8", newline="\n")
   written = [IMAGE_ARCHIVE_FILE, "trivy.json", "sbom.cdx.json", "SOURCE_COMMIT", TESTED_OUTPUT_FILE]
   if record:
      (destination / IMAGE_RECIPE_RECORD).write_text(json.dumps(record, sort_keys=True) + "\n", encoding="utf-8", newline="\n")
      written.append(IMAGE_RECIPE_RECORD)
   (destination / "SHA256SUMS").write_text("".join(f"{_sha256(destination / file_name)} *{file_name}\n" for file_name in sorted(written)), encoding="utf-8", newline="\n")
   return PackagedImage(name=name, directory=destination, archive_sha256=_sha256(destination / IMAGE_ARCHIVE_FILE))


def _retain_diagnostics(directory: Path, trivy_report: Path, source_revision: str) -> None:
   """Keep only the scanner's own report and the source commit, never image.tar, so a failed scan's findings survive the cleanup."""
   directory.mkdir(parents=True, exist_ok=True)
   if trivy_report.is_file() and not trivy_report.is_symlink():
      shutil.copyfile(trivy_report, directory / "trivy.json")
   (directory / "SOURCE_COMMIT").write_text(f"{source_revision}\n", encoding="utf-8", newline="\n")


def _verify_helpers() -> None:
   """Require each vendored captured helper (`check-image-evidence.mjs`, `image-boundary.sh`) to be the receipted file before it runs."""
   try:
      receipt = cast("dict[str, dict[str, str]]", json.loads((HELPER_DIRECTORY / "receipt.json").read_text(encoding="utf-8")))
      expected = receipt["files"]
   except (OSError, ValueError, KeyError) as error:
      raise ImagePackagingError(f"the image helper receipt cannot be read: {type(error).__name__}") from error
   for program in (EVIDENCE_PROGRAM, BOUNDARY_PROGRAM):
      path = HELPER_DIRECTORY / program
      if path.is_symlink() or not path.is_file() or _sha256(path) != expected.get(program):
         raise ImagePackagingError(f"{program} is missing or differs from its receipted revision")


def _sha256(path: Path) -> str:
   """Return the SHA-256 of a file."""
   with path.open("rb") as stream:
      return hashlib.file_digest(stream, "sha256").hexdigest()