platform-ci/platform_ci/packaging.py

"""Collect the declared reports and artifacts of an already-built target into one bundle with a manifest.

This module only reads and copies files. It has no way to start a process, so packaging
can never compile or build anything; it records what the verification build produced.
The manifest is evidence produced in the same job as the build, not a signed attestation.
"""

import hashlib
import json
import shutil
from dataclasses import dataclass
from pathlib import Path

from platform_ci.build_inputs import BuildInputsRecord, compute_build_inputs
from platform_ci.build_tools import IMAGE_OUTPUT_DIRECTORY
from platform_ci.plan import Target
from platform_ci.versions import read_declared_version
from platform_ci.workspace import (
   GIT_METADATA_DIRECTORY_NAME,
   PLATFORM_CHECKOUT_DIRECTORY_NAME,
   PipelineError,
   resolve_component_directory,
)

MANIFEST_FILE_NAME = "manifest.json"
MANIFEST_SCHEMA_VERSION = 1


class PackagingError(PipelineError):
   """Raised when a declared output is missing, empty, or outside the checkout."""


@dataclass(frozen=True)
class RunIdentity:
   """Which source revision, platform revision and run produced the outputs."""

   source_revision: str
   platform_revision: str
   run_id: str
   run_attempt: str
   source_tree: str = ""
   """The git tree id of `source_revision`; a successful build cannot be packaged without it, because reuse proof needs it."""


def collect_outputs(target: Target, source_directory: Path, output_directory: Path, identity: RunIdentity, build_succeeded: bool) -> dict[str, object]:
   """Copy declared outputs into the bundle and write `manifest.json`; return the manifest.

   After a successful build every declared artifact and every required report must match at least one
   file and every matched file must be nonempty. An optional report (`required: false`) is retained when
   it matches and is never required to exist. After a failed build only the reports that exist are kept
   for diagnosis and no artifact is collected, so a failed build never yields a releasable bundle.
   """
   if output_directory.exists():
      raise PackagingError(f"output directory {output_directory} already exists")
   component_directory = resolve_component_directory(source_directory, target["directory"])
   build_inputs = _build_inputs(target, component_directory, identity) if build_succeeded else None
   excluded_directories = (
      source_directory.resolve() / PLATFORM_CHECKOUT_DIRECTORY_NAME,
      output_directory.resolve(),
   )

   declared = [("report", name, report["paths"], report["required"]) for name, report in sorted(target["reports"].items())]
   if build_succeeded:
      declared += [("artifact", name, patterns, True) for name, patterns in sorted(target["artifacts"].items())]

   problems: list[str] = []
   planned: list[tuple[str, str, Path, str]] = []
   for kind, name, patterns, required in declared:
      matches = _matching_files(component_directory, patterns, excluded_directories)
      if not matches and build_succeeded and required:
         problems.append(f"{kind} {name!r} matched no files")
      for relative_path, path in matches:
         if build_succeeded and required and path.stat().st_size == 0:
            problems.append(f"{kind} {name!r} file {relative_path} is empty")
         planned.append((kind, name, path, relative_path))
   if problems:
      raise PackagingError("; ".join(problems))

   entries: list[dict[str, object]] = []
   for kind, name, path, relative_path in planned:
      bundle_path = Path(kind) / name / relative_path
      destination = output_directory / bundle_path
      destination.parent.mkdir(parents=True, exist_ok=True)
      shutil.copyfile(path, destination)
      entries.append(
         {
            "kind": kind,
            "name": name,
            "path": bundle_path.as_posix(),
            "size": destination.stat().st_size,
            "sha256": _sha256_of_file(destination),
         }
      )

   output_directory.mkdir(parents=True, exist_ok=True)
   manifest: dict[str, object] = {
      "schema": MANIFEST_SCHEMA_VERSION,
      "source_revision": identity.source_revision,
      "platform_revision": identity.platform_revision,
      "run_id": identity.run_id,
      "run_attempt": identity.run_attempt,
      "target": target["name"],
      "component": target["component"],
      "platform": target["platform"],
      "build_succeeded": build_succeeded,
      "files": sorted(entries, key=lambda entry: str(entry["path"])),
   }
   if build_inputs is not None:
      manifest["build_inputs"] = dict(build_inputs)
   manifest["digest"] = manifest_digest(manifest)
   (output_directory / MANIFEST_FILE_NAME).write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8", newline="\n")
   return manifest


def _build_inputs(target: Target, component_directory: Path, identity: RunIdentity) -> BuildInputsRecord:
   """Record what the successful build consumed; a missing source tree is an error, never a manifest nobody can reuse or prove."""
   if not identity.source_tree:
      raise PackagingError("the source tree id is required to record the build inputs of a successful build")
   try:
      return compute_build_inputs(target, identity.source_tree, read_declared_version(target["build_tool"], component_directory, target["commands"], target.get("version_source")))
   except ValueError as error:
      raise PackagingError(str(error)) from error


def tested_artifact_files(target: Target, source_directory: Path, artifact_name: str) -> list[tuple[str, Path]]:
   """Return the (component-relative path, file) pairs the verification left for one declared artifact; each must exist and be non-empty.

   Image packaging reads exactly the files the bundle will later record for that artifact, so what an image contains is what the manifest lists.
   Git metadata, the platform checkout and the image output directory are never part of it. This only reads files; it cannot start a process.
   """
   component_directory = resolve_component_directory(source_directory, target["directory"])
   excluded = (source_directory.resolve() / PLATFORM_CHECKOUT_DIRECTORY_NAME, component_directory / IMAGE_OUTPUT_DIRECTORY)
   matches = _matching_files(component_directory, target["artifacts"][artifact_name], excluded)
   if not matches:
      raise PackagingError(f"artifact {artifact_name!r} matched no files")
   empty = [relative for relative, path in matches if path.stat().st_size == 0]
   if empty:
      raise PackagingError(f"artifact {artifact_name!r} has empty files: {empty}")
   return matches


def manifest_digest(manifest: dict[str, object]) -> str:
   """Digest every manifest field except the digest itself, so it binds the files to the source revision and run."""
   content = {key: value for key, value in manifest.items() if key != "digest"}
   return hashlib.sha256(json.dumps(content, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest()


def _matching_files(component_directory: Path, patterns: list[str], excluded_directories: tuple[Path, ...]) -> list[tuple[str, Path]]:
   """Return the sorted unique files matched by the globs.

   Git metadata, the platform checkout and the bundle's own output directory are never collected,
   however broad the glob. A match, including a symbolic link to a directory, that resolves outside
   the component directory or into an excluded location is an error, not a silent skip.
   """
   matches: dict[str, Path] = {}
   for pattern in patterns:
      for candidate in component_directory.glob(pattern):
         relative_path = candidate.relative_to(component_directory)
         if _is_excluded(candidate, relative_path, excluded_directories):
            continue
         resolved = candidate.resolve()
         if not resolved.is_relative_to(component_directory):
            raise PackagingError(f"{relative_path.as_posix()} resolves outside the component directory")
         if _is_excluded(resolved, resolved.relative_to(component_directory), excluded_directories):
            raise PackagingError(f"{relative_path.as_posix()} resolves into Git metadata, the platform checkout or the output directory")
         if resolved.is_file():
            matches[relative_path.as_posix()] = candidate
   return sorted(matches.items())


def _is_excluded(path: Path, relative_path: Path, excluded_directories: tuple[Path, ...]) -> bool:
   """Return whether the path is, or is inside, Git metadata, the platform checkout or the output directory."""
   return GIT_METADATA_DIRECTORY_NAME in relative_path.parts or any(path.is_relative_to(directory) for directory in excluded_directories)


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