diff --git a/.github/workflows/bake.yml b/.github/workflows/bake.yml index 770be422..2f789a81 100644 --- a/.github/workflows/bake.yml +++ b/.github/workflows/bake.yml @@ -5,6 +5,8 @@ on: paths-ignore: - '.github/workflows/pgrx.yml' - '.github/workflows/pgrx_targets.yml' + - '.github/workflows/sbom-generator.yml' + - 'sbom-generator/**' workflow_dispatch: inputs: extension_name: diff --git a/.github/workflows/bake_targets.yml b/.github/workflows/bake_targets.yml index 70f728e4..75f3b0dc 100644 --- a/.github/workflows/bake_targets.yml +++ b/.github/workflows/bake_targets.yml @@ -45,6 +45,17 @@ jobs: - name: Set up Docker Buildx uses: docker/setup-buildx-action@594f3bf4285d9ea8dc53c9a0c9c4092420091003 # v4 + with: + # Run SBOM generators sequentially to reduce peak runner memory use. + buildkitd-config-inline: | + [worker.oci] + max-parallelism = 1 + + - name: Expose builder stage to SBOM generator + env: + EXTENSION: ${{ inputs.extension_name }} + run: | + sed -i '2i ARG BUILDKIT_SBOM_SCAN_STAGE=builder' "$EXTENSION/Dockerfile" - name: Build and push uses: docker/bake-action@018cb6412ab401ebaa809aa5f85966b74628600f # v7 @@ -54,7 +65,11 @@ jobs: environment: testing registry: ghcr.io/${{ github.repository_owner }} revision: ${{ github.sha }} + # renovate: datasource=docker depName=ghcr.io/cnpg-extensions/cnpg-sbom-generator + sbom_generator: ghcr.io/cnpg-extensions/cnpg-sbom-generator:latest@sha256:38605482cdeb890e015a0d0e49edd473ecb483a2c0f075f3550e4470f247861e with: + # Use the checkout so Bake sees the injected Dockerfile declaration. + source: . files: ./docker-bake.hcl,./${{ inputs.extension_name }}/metadata.hcl push: true diff --git a/.github/workflows/sbom-generator.yml b/.github/workflows/sbom-generator.yml index 3723b873..7b42623f 100644 --- a/.github/workflows/sbom-generator.yml +++ b/.github/workflows/sbom-generator.yml @@ -1,34 +1,73 @@ -name: Build SBOM generator (stub) +name: Build and publish SBOM generator on: + push: + branches: [main] + paths: + - 'sbom-generator/**' + - '.github/workflows/sbom-generator.yml' + pull_request: + paths: + - 'sbom-generator/**' + - '.github/workflows/sbom-generator.yml' workflow_dispatch: + inputs: + publish: + description: "Publish the generator image to GHCR (manual testing only)" + required: false + default: false + type: boolean permissions: {} +concurrency: + group: sbom-generator-${{ github.ref }} + cancel-in-progress: false + jobs: build: - name: Build SBOM generator stub runs-on: ubuntu-24.04 permissions: contents: read + packages: write steps: - name: Checkout uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 with: persist-credentials: false + - name: Run tests + run: python3 -m unittest discover -s sbom-generator/tests -p 'test_*.py' + + - name: Set up QEMU + uses: docker/setup-qemu-action@99012661954931238ded8c8b007157a8430204e1 # v4 + with: + platforms: linux/arm64 + - name: Set up Docker Buildx uses: docker/setup-buildx-action@594f3bf4285d9ea8dc53c9a0c9c4092420091003 # v4 - - name: Build generator stub + - name: Log in to GHCR + if: ${{ github.event_name == 'push' || (github.event_name == 'workflow_dispatch' && inputs.publish == true) }} + uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f # v4 + with: + registry: ghcr.io + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Build generator env: - IMAGE: cnpg-sbom-generator:stub-${{ github.sha }} + EVENT_NAME: ${{ github.event_name }} + PUBLISH: ${{ inputs.publish }} run: | + image="ghcr.io/cnpg-extensions/cnpg-sbom-generator" + options=() + if [[ "$EVENT_NAME" == push || "$PUBLISH" == true ]]; then options+=(--push); fi docker buildx build \ - --load \ - --tag "$IMAGE" \ - -f- . <<'DOCKERFILE' - FROM scratch - LABEL org.opencontainers.image.title="cnpg-sbom-generator" - LABEL org.opencontainers.image.description="Workflow-dispatch stub build" - DOCKERFILE \ No newline at end of file + --platform linux/amd64,linux/arm64 \ + --build-arg "SBOM_GENERATOR_REVISION=${GITHUB_SHA}" \ + --label "org.opencontainers.image.source=https://github.com/${GITHUB_REPOSITORY}" \ + --tag "$image:latest" \ + --tag "$image:sha-${GITHUB_SHA}" \ + "${options[@]}" \ + sbom-generator diff --git a/.gitignore b/.gitignore index 51b8ae69..8c081895 100644 --- a/.gitignore +++ b/.gitignore @@ -8,6 +8,10 @@ *.so *.dylib +# Python test and bytecode caches +__pycache__/ +*.py[cod] + # Mac .DS_Store diff --git a/README.md b/README.md index 20fb46a5..a6d96748 100644 --- a/README.md +++ b/README.md @@ -211,6 +211,14 @@ docker buildx imagetools inspect --raw | jq '.annotations' skopeo inspect docker:// | jq '.Labels' ``` +## SBOMs and authenticity + +Published images carry platform-specific BuildKit SPDX and provenance +attestations. See the [SBOM and authenticity guide](./sbom-generator/README.md) +for digest-pinned Cosign verification, Buildx extraction, platform-specific +Trivy scanning, and the distinction between payload inventory and direct image +scans. + ## Image catalogs To simplify the deployment of PostgreSQL extensions, this project automatically diff --git a/docker-bake.hcl b/docker-bake.hcl index 382b7f55..24b66182 100644 --- a/docker-bake.hcl +++ b/docker-bake.hcl @@ -16,6 +16,9 @@ variable "revision" { } fullname = ( environment == "testing") ? "${registry}/${metadata.image_name}-testing" : "${registry}/${metadata.image_name}" +variable "sbom_generator" { + default = "" +} now = timestamp() authors = "The CNPG Extensions Contributors" url = "https://github.com/cnpg-extensions/postgres-extensions-containers" @@ -48,10 +51,9 @@ target "default" { output = [ "type=image,oci-mediatypes=true,oci-artifact=true", ] - attest = [ + attest = concat([ "type=provenance,mode=max", - "type=sbom" - ] + ], sbom_generator == "" ? ["type=sbom"] : ["type=sbom,generator=${sbom_generator}"]) annotations = [ "index,manifest:org.opencontainers.image.created=${now}", "index,manifest:org.opencontainers.image.url=${url}", diff --git a/examples/trivy-sbom-examples.txt b/examples/trivy-sbom-examples.txt new file mode 100644 index 00000000..c0c8ed25 --- /dev/null +++ b/examples/trivy-sbom-examples.txt @@ -0,0 +1,75 @@ +# Trivy SBOM example +# +# This report was generated from the SPDX document extracted from the local +# OCI attestation for pgagent bookworm. Long Debian copyright files were split +# at License: sections before ScanCode ran. The image was built locally with +# BuildKit and pushed to a local registry. GitHub OIDC/cosign attestation +# verification is omitted because it requires a hosted GitHub runner. + +$ docker buildx imagetools inspect 127.0.0.1:5000/pgagent-testing:4.2.3-18-bookworm@sha256:ea4735334c2f494b45b4ba50ac44f66bf55e6e4af56599862e6062341e5eb54e --format '{{ json .SBOM.SPDX }}' > /tmp/pgagent-bookworm-amd64.spdx.json +$ docker run --rm -v /tmp:/work aquasec/trivy:0.74.0 sbom --scanners vuln,license --no-progress --skip-version-check /work/pgagent-bookworm-amd64.spdx.json +2026-09-15T04:04:17Z INFO [vulndb] Need to update DB +2026-09-15T04:04:17Z INFO [vulndb] Downloading vulnerability DB... +2026-09-15T04:04:17Z INFO [vulndb] Downloading artifact... repo="mirror.gcr.io/aquasec/trivy-db:2" +2026-09-15T04:04:30Z INFO [vulndb] Artifact successfully downloaded repo="mirror.gcr.io/aquasec/trivy-db:2" +2026-09-15T04:04:30Z INFO [vuln] Vulnerability scanning is enabled +2026-09-15T04:04:30Z INFO [license] License scanning is enabled +2026-09-15T04:04:30Z INFO Detected SBOM format format="spdx-json" +2026-09-15T04:04:30Z INFO Detected OS family="debian" version="12.15" +2026-09-15T04:04:30Z INFO [debian] Detecting vulnerabilities... os_version="12" pkg_num=3 +2026-09-15T04:04:30Z INFO Number of language-specific files num=0 + +Report Summary + +┌───────────────────────────────────────────────────────┬────────┬─────────────────┬──────────┐ +│ Target │ Type │ Vulnerabilities │ Licenses │ +├───────────────────────────────────────────────────────┼────────┼─────────────────┼──────────┤ +│ /work/pgagent-bookworm-amd64.spdx.json (debian 12.15) │ debian │ 1 │ - │ +├───────────────────────────────────────────────────────┼────────┼─────────────────┼──────────┤ +│ OS Packages │ - │ - │ 3 │ +└───────────────────────────────────────────────────────┴────────┴─────────────────┴──────────┘ +Legend: +- '-': Not scanned +- '0': Clean (no security findings detected) + + +/work/pgagent-bookworm-amd64.spdx.json (debian 12.15) +===================================================== +Total: 1 (UNKNOWN: 0, LOW: 0, MEDIUM: 0, HIGH: 1, CRITICAL: 0) + +┌─────────┬───────────────┬──────────┬──────────┬───────────────────┬───────────────┬──────────────────────────────────────────────────────────────┐ +│ Library │ Vulnerability │ Severity │ Status │ Installed Version │ Fixed Version │ Title │ +├─────────┼───────────────┼──────────┼──────────┼───────────────────┼───────────────┼──────────────────────────────────────────────────────────────┤ +│ pgagent │ CVE-2025-0218 │ HIGH │ affected │ 4.2.3-5.pgdg12+1 │ │ When batch jobs are executed by pgAgent, a script is created │ +│ │ │ │ │ │ │ in... │ +│ │ │ │ │ │ │ https://avd.aquasec.com/nvd/cve-2025-0218 │ +└─────────┴───────────────┴──────────┴──────────┴───────────────────┴───────────────┴──────────────────────────────────────────────────────────────┘ + +OS Packages (license) +===================== +Total: 3 (UNKNOWN: 3, LOW: 0, MEDIUM: 0, HIGH: 0, CRITICAL: 0) + +┌───────────────────────────┬──────────────────────────────────────────────────────────────┬────────────────┬──────────┐ +│ Package │ License │ Classification │ Severity │ +├───────────────────────────┼──────────────────────────────────────────────────────────────┼────────────────┼──────────┤ +│ libboost-filesystem1.74.0 │ Apache-2.0 AND LicenseRef-BSD2 AND LicenseRef-BSD3-DEShaw │ unknown │ UNKNOWN │ +│ │ AND LicenseRef-BSD3-Google AND BSL-1.0 AND │ │ │ +│ │ LicenseRef-Caramel AND LicenseRef-CrystalClear AND │ │ │ +│ │ LicenseRef-HP AND Jam AND LicenseRef-Kempf AND MIT AND │ │ │ +│ │ LicenseRef-NIST AND LicenseRef-OldBoost1 AND │ │ │ +│ │ LicenseRef-OldBoost2 AND LicenseRef-OldBoost3 AND │ │ │ +│ │ LicenseRef-Python AND LicenseRef-SGI AND LicenseRef-Spencer │ │ │ +│ │ AND Zlib │ │ │ +├───────────────────────────┤ │ │ │ +│ libboost-thread1.74.0 │ │ │ │ +│ │ │ │ │ +│ │ │ │ │ +│ │ │ │ │ +│ │ │ │ │ +│ │ │ │ │ +│ │ │ │ │ +│ │ │ │ │ +│ │ │ │ │ +├───────────────────────────┼──────────────────────────────────────────────────────────────┤ │ │ +│ pgagent │ LicenseRef-scancode-unknown-license-reference AND PostgreSQL │ │ │ +└───────────────────────────┴──────────────────────────────────────────────────────────────┴────────────────┴──────────┘ diff --git a/renovate.json b/renovate.json index 4268ef27..3f9380ef 100644 --- a/renovate.json +++ b/renovate.json @@ -75,6 +75,39 @@ "matchStrings": [ "\\/\\/\\s+renovate: datasource=(?[a-z-.]+?) depName=(?[^\\s]+?)(?: (?:packageName)=(?[^\\s]+?))?(?: versioning=(?[^\\s]+?))\\s+\\/\\/\\s+\\+default=[\\\"']?[^:]+?:(?[^@]+?)(@(?sha256:[0-9a-f]+))?[\"']?\\s" ] + }, + { + "description": "updates the SBOM generator base image", + "customType": "regex", + "managerFilePatterns": [ + "sbom-generator/Dockerfile" + ], + "matchStrings": [ + "#\\s*renovate: datasource=(?[^\\s]+) depName=(?[^\\s]+)(?: packageName=(?[^\\s]+))?(?: versioning=(?[^\\s]+))?\\s+FROM\\s+(?:[a-zA-Z0-9._/-]+:)?(?[^@\\s]+)(?:@(?sha256:[0-9a-f]+))?" + ] + }, + { + "description": "updates the SBOM generator Dockerfile frontend", + "customType": "regex", + "managerFilePatterns": [ + "sbom-generator/Dockerfile" + ], + "matchStrings": [ + "# syntax=docker/dockerfile:(?[^\\s]+)" + ], + "depNameTemplate": "docker/dockerfile", + "datasourceTemplate": "docker", + "versioningTemplate": "docker" + }, + { + "description": "updates the SBOM generator tool versions", + "customType": "regex", + "managerFilePatterns": [ + "sbom-generator/Dockerfile" + ], + "matchStrings": [ + "#\\s*renovate: datasource=(?[^\\s]+) depName=(?[^\\s]+)(?: versioning=(?[^\\s]+))?\\s+ARG [A-Z0-9_]+_VERSION=(?[^\\s]+)" + ] } ], "packageRules": [ diff --git a/sbom-generator/Dockerfile b/sbom-generator/Dockerfile new file mode 100644 index 00000000..98913dae --- /dev/null +++ b/sbom-generator/Dockerfile @@ -0,0 +1,32 @@ +# syntax=docker/dockerfile:1.7 + +# Python 3.14 is the newest version supported by the pinned ScanCode 32.5.0. +# Move to a newer Python as soon as ScanCode supports it; keep this aligned with SCANCODE_VERSION. +# https://github.com/aboutcode-org/scancode-toolkit/releases/tag/v32.5.0 +# renovate: datasource=docker depName=python packageName=library/python versioning=docker +FROM python:3.14-slim-bookworm + +ARG TARGETARCH +# renovate: datasource=github-releases depName=anchore/syft versioning=semver +ARG SYFT_VERSION=1.51.1 +# renovate: datasource=pypi depName=scancode-toolkit versioning=pep440 +ARG SCANCODE_VERSION=32.5.0 + +RUN apt-get update \ + && apt-get install --no-install-recommends -y ca-certificates curl gcc g++ libxml2-dev libxslt1-dev pkg-config \ + && rm -rf /var/lib/apt/lists/* \ + && curl --fail --silent --show-error --location \ + "https://github.com/anchore/syft/releases/download/v${SYFT_VERSION}/syft_${SYFT_VERSION}_linux_${TARGETARCH}.tar.gz" \ + | tar -xz -C /usr/local/bin syft \ + && python -m pip install --no-cache-dir "scancode-toolkit==${SCANCODE_VERSION}" \ + && syft version + +WORKDIR /opt/cnpg-sbom-generator +COPY compose.py generator.py hooks.py ./ + +ARG SBOM_GENERATOR_REVISION=unknown +ENV SBOM_GENERATOR_REVISION=${SBOM_GENERATOR_REVISION} +LABEL org.opencontainers.image.revision=${SBOM_GENERATOR_REVISION} + +ENV PYTHONUNBUFFERED=1 +ENTRYPOINT ["python3", "/opt/cnpg-sbom-generator/generator.py"] diff --git a/sbom-generator/README.md b/sbom-generator/README.md new file mode 100644 index 00000000..4658b308 --- /dev/null +++ b/sbom-generator/README.md @@ -0,0 +1,253 @@ +# Extension image SBOMs + +BuildKit runs the custom generator as a separate container and mounts the +builder-stage and final scratch filesystems read-only. The generator replaces +the default SBOM scanner; Bake and BuildKit still handle building the image +and attaching its attestations. + +```mermaid +flowchart TD + A[1. Docker Bake submits build] --> B[2. BuildKit builds builder stage,
assembles final scratch filesystem,
and runs generator with both mounted read-only] + B --> C[Builder stage filesystem] + B --> D[Final scratch filesystem] + + subgraph G[Custom generator container] + E[3. Syft: packages, files, ownership] + F[4. Inventory and hash shipped files] + H[6. Match shipped files to builder files
Retain their owning packages] + I[5. ScanCode: shipped licenses] + P[7. Compose SPDX with license metadata] + X[8. Optional downstream augment_spdx hook:
add dependency version and license information] + J[9. Wrap SPDX in an in-toto statement] + E --> H + F --> H + H --> P + I --> P + P --> J + P -.-> X + E -. full builder_document .-> X + X -.-> J + end + + C -->|read-only mount| E + D -->|read-only mount| F + D -->|licenses| I + C -. additional evidence .-> X + J -->|output directory| K[10. BuildKit binds statement to image digest
and publishes SBOM attestation] +``` + +Numbers show the processing sequence for one target/platform; arrows show +data flow. Unnumbered filesystem nodes are inputs. Step 8 runs only when a +downstream hook file is present in the builder filesystem. + +The generator decides what is in the SBOM: shipped files and their identified +packages, plus license and distro metadata. It starts from its own Syft scan +of the builder stage, not Bake's default SBOM. BuildKit pulls the generator +image, supplies the mounts, and incorporates its output into the image index +alongside provenance. + +## Downstream extension hook + +The dashed path in the diagram is an optional extension point. Without a +hook file, the generator wraps the composed SPDX directly. With the +hook enabled, augmentation runs after final-payload +filtering and license composition, before in-toto statement wrapping. + +```python +from hooks import HookContext + +def augment_spdx(document: dict, context: HookContext) -> dict: + """Add additional dependency version and license information.""" + return document +``` + +The context exposes `api_version` (currently `1`), `extension_name`, `platform`, +and read-only filesystem mounts at `builder_path` and `final_path`, plus +`context.builder_document`: the complete, unfiltered SPDX JSON document +produced by the builder-stage Syft scan, before final-file matching or package +filtering. This is the existing parsed `builder_document`, not Syft's native +JSON format, and requires no additional scan. Hooks should treat it as +read-only evidence; the `document` argument is the composed final-payload SPDX +that the hook augments. +Final-file identifiers and checksums are available in `document["files"]`. + +A downstream hook could reuse Syft's full package, file, and relationship +data and read additional evidence from the builder stage to add dependency +version and license information for shipped artifacts to the SPDX document. +Running after filtering prevents these additions from being discarded by the existing +file-ownership selection logic. + +The upstream runner trusts the hook's returned document and wraps it +directly in an in-toto statement. Downstream code is responsible for +producing valid SPDX and for evidence that added components belong to shipped +artifacts. BuildKit receives one completed SPDX statement through the +existing output protocol. + +Place a Python file defining `augment_spdx(document, context)` at +`/usr/local/share/cnpg-sbom/augment_spdx.py` in the extension's builder stage. For +example, add this instruction to that stage, using a source path relative to +the build context: + +```dockerfile +COPY sbom/augment_spdx.py /usr/local/share/cnpg-sbom/augment_spdx.py +``` + +The generator loads that file from the read-only builder mount and calls its +function. No hook file means a no-op. The hook must return the complete SPDX +document; it can modify the supplied document or return a replacement. Load +and hook errors fail the build. There is no additional validation of the +hook's returned SPDX. + +Downstream builds use the same generator image. The hook and its evidence +remain in the builder stage and need not be copied into scratch. The hook +executes in the generator's Python environment, so dependencies installed +only in the builder are not automatically available. Standard-library +processing of precomputed JSON needs no additional generator dependencies. + +## Build locally + +Local builds use Docker Bake directly and do not require the SBOM generator: + +```bash +docker buildx bake -f docker-bake.hcl -f h3/metadata.hcl h3-4_2_3-18-trixie +``` + +The `sbom_generator` Bake variable defaults to empty, which uses BuildKit's +native `type=sbom` generator. Set it to a generator image reference to use the +custom generator. Use `--builder NAME` to select a configured builder, or +`--print` to inspect the Bake definition. + +The separate `sbom-generator.yml` workflow builds the generator when anything +under `sbom-generator/` or the publishing workflow changes. Pull requests validate +the build; changes on `main` publish multi-platform `latest` and `sha-` tags. + +Extension CI consumes the image reference in `bake_targets.yml`. Renovate's +existing Docker digest-pinning configuration tracks `latest`, adding its first +digest after publication and proposing digest updates for subsequent builds. +Generator source changes do not rebuild extensions directly; updating the image +reference in `bake_targets.yml` triggers those builds. When adopting this workflow +in another repository, update both the image reference and its Renovate +`depName` to the publishing owner's GHCR namespace. + +Immediately before Bake, CI edits the checked-out extension Dockerfile to insert +`ARG BUILDKIT_SBOM_SCAN_STAGE=builder` as line 2, after the syntax directive +where present. This exposes the builder +stage to the custom generator. The declaration is required by +[BuildKit stage scanning](https://docs.docker.com/build/metadata/attestations/sbom/#scan-stages); +passing only a build argument does not enable it. + +To test custom SBOM generation locally, publish a generator image accessible +to BuildKit, add the same declaration to your extension Dockerfile, and set +`sbom_generator` to the image reference when running Bake. Ordinary local +builds need neither preparation step. + +## Attestation layout + +Published extension images use BuildKit's SBOM generator protocol. The local +generator emits one in-toto Statement with an SPDX predicate for the platform +payload; BuildKit supplies the final image subject, attestation manifest, and +image index. A multi-platform index therefore contains one image and one +combined provenance/SPDX attestation for each platform. + +The SPDX creation metadata identifies the generator as +`Tool: cnpg-sbom-generator-`. The publishing workflow embeds its full +build Git SHA in the generator image through `SBOM_GENERATOR_REVISION`; this +also sets the image's `org.opencontainers.image.revision` label. Its document +annotation retains the generator name, Git SHA as `generatorVersion`, and source +repository URL; the immutable generator image digest +selected by the workflow provides the stronger reproducibility boundary. +Local image builds can pass `--build-arg SBOM_GENERATOR_REVISION=$(git rev-parse HEAD)` +when building from an unchanged checkout. Without a supplied revision, the +version is explicitly `unknown`, including when running the composer directly. + +During generation, BuildKit logs show phase start/completion and elapsed time. +Long Syft subprocesses emit a heartbeat every 10 seconds. +ScanCode's output and diagnostics go directly to stdout without filtering. +ScanCode controls its own progress display; non-terminal logs may omit per-file messages. +License-file preparation reports its file and chunk counts. + +The examples below use the H3 image produced by this repository. Replace +`INDEX_DIGEST` with the immutable index digest returned by +`docker buildx imagetools inspect`; the placeholder is intentional because a +tag is not a reproducible security reference. + +## Verify the image + +The retained release workflow is `.github/workflows/bake_targets.yml`. A +production image built from `main` is verified with the workflow identity and +GitHub's OIDC issuer: + +```bash +IMAGE='ghcr.io/cloudnative-pg/h3:4.2.3-18-trixie@sha256:INDEX_DIGEST' + +cosign verify "$IMAGE" \ + --certificate-identity-regexp='^https://github.com/cloudnative-pg/postgres-extensions-containers/.github/workflows/bake_targets\.yml@refs/heads/main$' \ + --certificate-oidc-issuer='https://token.actions.githubusercontent.com' +``` + +Substitute the actual GitHub owner in both the image and identity. A branch or +pull-request build has a different workflow identity. Local validation uses a +disposable Cosign key and `COSIGN_TLOG_UPLOAD=false`; it validates signature +storage and digest binding, not the future hosted OIDC identity. + +## Retrieve the platform SBOM + +Use Buildx's standard platform selector and SPDX template. The extension image +reference is the only difference from the base PostgreSQL extraction form: + +```bash +docker buildx imagetools inspect "$IMAGE" \ + --format '{{ json (index .SBOM "linux/amd64").SPDX }}' \ + > extension-amd64.spdx.json + +trivy sbom --scanners vuln,license extension-amd64.spdx.json +``` + +For the other platform, repeat the same command with `linux/arm64` and a +separate output file: + +```bash +docker buildx imagetools inspect "$IMAGE" \ + --format '{{ json (index .SBOM "linux/arm64").SPDX }}' \ + > extension-arm64.spdx.json + +trivy sbom --scanners vuln,license extension-arm64.spdx.json +``` + +Each report covers one platform. Native-only builds should be extracted and +scanned with `linux/amd64` or `linux/arm64` matching the built image; validate +the second architecture only after the final multi-platform phase. + +The composed document describes the shipped extension payload, including +copied system libraries and `/licenses` files. It does not describe the whole +PostgreSQL container or runtime dependencies supplied by the base image. Scan +the base PostgreSQL image and separately mounted extension images separately. + +## Direct image scans + +Debian-based PostgreSQL images retain installed-package metadata, so a direct +remote image scan can identify OS packages: + +```bash +trivy image --image-src remote --platform linux/amd64 --scanners vuln \ + ghcr.io/cloudnative-pg/postgresql:18-minimal-trixie +``` + +Scratch extension payloads generally do not retain that ownership/version +metadata. The extracted composed SPDX is the intended inventory-based path. +`trivy image` scans the image filesystem and does not consume the BuildKit SBOM +just because the index contains an SBOM attestation; `trivy sbom` consumes the +document extracted above. + +Extraction and vulnerability/license scanning are separate from authenticity. +Cosign verifies the signed index digest. Buildx extraction reads attestation +members whose OCI subject points to a platform image, and the in-toto subject +and descriptor digests bind each statement and blob to that graph. Trivy then +scans the extracted SPDX content; it does not verify the signature. + +For the first hosted rollout, publish the generator before running extension +CI, then let Renovate pin the published `latest` manifest digest. GHCR must be +readable by the consuming workflows and Renovate (through package visibility +or configured credentials). The initial reference has no digest until that +publication; subsequent digest updates select the generator used by extension +builds. Local validation does not exercise hosted OIDC/publication jobs. diff --git a/sbom-generator/compose.py b/sbom-generator/compose.py new file mode 100755 index 00000000..6d9b39d5 --- /dev/null +++ b/sbom-generator/compose.py @@ -0,0 +1,570 @@ +#!/usr/bin/env python3 +"""Compose a platform-specific SPDX predicate for the BuildKit SBOM protocol. + +The composer deliberately knows nothing about OCI indexes or in-toto +statements. The generator adds the protocol statement around this predicate; +BuildKit supplies the attestation manifest and binds it to the platform image. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import sys +from collections import defaultdict +from copy import deepcopy +from pathlib import Path +from typing import Any +from urllib.parse import parse_qs, urlsplit + + +EXTENSION_PACKAGE_ID = "SPDXRef-Package-extension-payload" +GENERATOR_NAME = "cnpg-sbom-generator" +GENERATOR_REPOSITORY = "https://github.com/cnpg-extensions/postgres-extensions-containers" +LICENSE_REF = re.compile(r"LicenseRef-[A-Za-z0-9][A-Za-z0-9.-]*") +LICENSE_OPERATOR = re.compile(r"\s+(?:AND|OR|WITH)\s+") + + +def read_json(path: Path) -> dict[str, Any]: + with path.open(encoding="utf-8") as stream: + return json.load(stream) + + +def builder_predicate(document: dict[str, Any], path: Path) -> dict[str, Any]: + """Return a raw SPDX document from either Syft or an old attestation. + + The old PR61 fixtures are accepted so the ownership algorithm can be + regression-tested without making an in-toto statement part of the + generator's input or output contract. + """ + + if document.get("predicateType") == "https://spdx.dev/Document": + predicate = document.get("predicate") + else: + predicate = document + if not isinstance(predicate, dict) or predicate.get("SPDXID") != "SPDXRef-DOCUMENT": + raise ValueError(f"{path}: builder evidence is not an SPDX document") + for field in ("packages", "files", "relationships"): + if not isinstance(predicate.get(field), list): + raise ValueError(f"{path}: SPDX {field} must be an array") + return predicate + + +def checksum_key(algorithm: str, value: str) -> tuple[str, str]: + return algorithm.lower(), value.lower() + + +def final_files(document: dict[str, Any], path: Path) -> list[dict[str, str]]: + """Return final file names and checksums from a BuildKit attestation.""" + + files: list[dict[str, str]] = [] + for subject in document["subject"]: + name = subject["name"] + if name.startswith("pkg:"): + raise ValueError( + f"{path}: subject {name!r} is an image subject; use a local-export SBOM" + ) + files.append({ + "name": name.lstrip("/"), + "algorithm": "sha256", + "value": subject["digest"]["sha256"], + }) + if not files: + raise ValueError(f"{path}: final image has no file subjects") + return files + + +def final_inventory_files(inventory: dict[str, Any] | list[dict[str, Any]], path: Path) -> list[dict[str, str]]: + """Validate and normalize the generator's direct final-files inventory.""" + + records = inventory.get("files") if isinstance(inventory, dict) else inventory + if not isinstance(records, list) or not records: + raise ValueError(f"{path}: final filesystem has no files") + + files: list[dict[str, str]] = [] + seen: set[tuple[str, str, str]] = set() + for record in records: + if not isinstance(record, dict): + raise ValueError(f"{path}: final inventory entry is not an object") + name = record.get("name", record.get("fileName")) + if not isinstance(name, str) or not name or name.startswith("pkg:"): + raise ValueError(f"{path}: final inventory has an invalid file name") + checksums = record.get("checksums") + if checksums is not None: + if not isinstance(checksums, list) or len(checksums) != 1: + raise ValueError(f"{path}: final inventory entries need one checksum") + checksum = checksums[0] + algorithm = checksum.get("algorithm") + value = checksum.get("checksumValue") + else: + algorithm = record.get("algorithm", "sha256") + value = record.get("value") + if not isinstance(algorithm, str) or not isinstance(value, str) or not value: + raise ValueError(f"{path}: final inventory entry has no checksum") + normalized = { + "name": name.lstrip("/"), + "algorithm": algorithm.lower(), + "value": value.lower(), + } + identity = (normalized["name"], normalized["algorithm"], normalized["value"]) + if identity not in seen: + files.append(normalized) + seen.add(identity) + files.sort(key=lambda record: (record["name"], record["algorithm"], record["value"])) + return files + + +def path_score(candidate: str, final_name: str) -> tuple[int, int]: + """Prefer an exact path, then the longest shared path suffix.""" + + candidate_parts = tuple(part for part in candidate.lstrip("/").split("/") if part) + final_parts = tuple(part for part in final_name.lstrip("/").split("/") if part) + common_suffix = 0 + for candidate_part, final_part in zip(reversed(candidate_parts), reversed(final_parts)): + if candidate_part != final_part: + break + common_suffix += 1 + return common_suffix, int(candidate_parts == final_parts) + + +def file_id(name: str, algorithm: str, value: str) -> str: + identity = f"{name}\0{algorithm.lower()}:{value.lower()}".encode() + return f"SPDXRef-File-final-{hashlib.sha256(identity).hexdigest()[:24]}" + + +def scancode_licenses( + document: dict[str, Any], +) -> tuple[dict[str, set[str]], dict[str, dict[str, str]]]: + """Return ScanCode SPDX expressions and custom license definitions.""" + + licenses_by_file: defaultdict[str, set[str]] = defaultdict(set) + custom_licenses: set[str] = set() + for record in document.get("files", []): + licenses = { + expression + for detection in record.get("license_detections", []) + if (expression := detection.get("license_expression_spdx")) + and expression not in {"NONE", "NOASSERTION"} + } + if not licenses: + continue + path = record["path"].lstrip("/") + licenses_by_file[path].update(licenses) + custom_licenses.update( + license_id for expression in licenses for license_id in LICENSE_REF.findall(expression) + ) + references = { + reference["spdx_license_key"]: { + "extractedText": reference.get("text") or "NOASSERTION", + "licenseId": reference["spdx_license_key"], + "name": reference.get("name") or reference["spdx_license_key"], + } + for reference in document.get("license_references", []) + if reference.get("spdx_license_key") in custom_licenses + } + for license_id in custom_licenses: + references.setdefault(license_id, { + "extractedText": "NOASSERTION", + "licenseId": license_id, + "name": license_id, + }) + return licenses_by_file, references + + +def debian_os_package(packages: list[dict[str, Any]], path: Path) -> dict[str, Any]: + distros = { + distro + for package in packages + for reference in package.get("externalRefs", []) + if reference.get("referenceType") == "purl" + for distro in parse_qs(urlsplit(reference["referenceLocator"]).query).get("distro", []) + } + if len(distros) != 1 or not next(iter(distros), "").startswith("debian-"): + raise ValueError(f"{path}: packages must identify one Debian distro") + version = next(iter(distros)).removeprefix("debian-") + return { + "SPDXID": f"SPDXRef-OperatingSystem-debian-{version}", + "copyrightText": "NOASSERTION", + "downloadLocation": "NONE", + "filesAnalyzed": False, + "licenseConcluded": "NOASSERTION", + "licenseDeclared": "NOASSERTION", + "name": "debian", + "primaryPackagePurpose": "OPERATING-SYSTEM", + "versionInfo": version, + } + + +def set_document_namespace(document: dict[str, Any], extension_name: str, platform: str) -> None: + # SPDX element IDs are scoped by namespace; hash the content to prevent + # distinct SBOMs from sharing identities when consumers combine them. + content = {key: value for key, value in document.items() if key != "documentNamespace"} + digest = hashlib.sha256( + json.dumps(content, sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest() + document["documentNamespace"] = ( + "https://github.com/cnpg-extensions/postgres-extensions-containers/" + f"sbom-generator/v1/documents/{extension_name}/{platform.replace('/', '-')}-{digest}" + ) + + +def compose(builder_document: dict[str, Any], *, + extension_name: str, + builder_path: Path = Path("builder"), + scancode_report: dict[str, Any] | None = None, + final_inventory: dict[str, Any] | list[dict[str, Any]] | None = None, + platform: str | None = None, + evidence: dict[str, Any] | None = None, + ) -> dict[str, Any]: + """Return one raw, platform-specific SPDX predicate. + + ``final_inventory`` is the normal path. ``subject`` handling remains as a + compatibility fixture for the original composer tests, but the generator + never needs a final image/index digest before it writes its predicate. + """ + + builder = builder_predicate(builder_document, builder_path) + licenses_by_file, custom_licenses = scancode_licenses(scancode_report or {}) + final = ( + final_inventory_files(final_inventory, builder_path) + if final_inventory is not None + else final_files(builder_document, builder_path) + ) + builder_records = builder["files"] + relationships = builder["relationships"] + packages = builder["packages"] + + builder_packages = { + package["SPDXID"]: package + for package in packages + if package.get("primaryPackagePurpose") != "FILE" + } + + all_package_ids = {package["SPDXID"] for package in packages} + package_ids = set(builder_packages) + package_ids_by_name: defaultdict[str, set[str]] = defaultdict(set) + for package_id, package in builder_packages.items(): + package_ids_by_name[package["name"]].add(package_id) + retained_package_ids: set[str] = set() + owners_by_source_file: defaultdict[str, set[str]] = defaultdict(set) + for relationship in relationships: + if relationship["relationshipType"] != "CONTAINS": + continue + package_id = relationship["spdxElementId"] + source_file_id = relationship["relatedSpdxElement"] + if package_id in package_ids: + owners_by_source_file[source_file_id].add(package_id) + + by_checksum: defaultdict[tuple[str, str], list[dict[str, Any]]] = defaultdict(list) + all_file_ids = {record["SPDXID"] for record in builder_records} + for record in builder_records: + for checksum in record["checksums"]: + by_checksum[checksum_key( + checksum["algorithm"], checksum["checksumValue"] + )].append(record) + + composed_files: list[dict[str, Any]] = [] + source_to_final: defaultdict[str, set[str]] = defaultdict(set) + direct_final_owners: defaultdict[str, set[str]] = defaultdict(set) + final_ids: set[str] = set() + + def add_synthetic_file(record: dict[str, str], owner: str | None = None) -> None: + output_record = { + "SPDXID": file_id(record["name"], record["algorithm"], record["value"]), + "checksums": [{ + "algorithm": record["algorithm"].upper(), + "checksumValue": record["value"], + }], + "copyrightText": "NOASSERTION", + "fileName": record["name"], + "licenseConcluded": "NOASSERTION", + "licenseInfoInFiles": ["NOASSERTION"], + } + composed_files.append(output_record) + final_ids.add(output_record["SPDXID"]) + if owner is not None: + retained_package_ids.add(owner) + direct_final_owners[output_record["SPDXID"]].add(owner) + + for final_record in final: + final_name = final_record["name"].lstrip("/") + license_parts = final_name.split("/", 2) + if license_parts[0] == "licenses" and len(license_parts) > 1: + owners = package_ids_by_name.get(license_parts[1], set()) + add_synthetic_file(final_record, next(iter(owners)) if len(owners) == 1 else None) + continue + + candidates = by_checksum.get( + checksum_key(final_record["algorithm"], final_record["value"]), [] + ) + if not candidates: + add_synthetic_file(final_record) + continue + + owned_candidates = [ + record for record in candidates if record["SPDXID"] in owners_by_source_file + ] + candidates = owned_candidates or candidates + best_score = max(path_score(record["fileName"], final_name) for record in candidates) + selected = [ + record for record in candidates + if path_score(record["fileName"], final_name) == best_score + ] + selected.sort(key=lambda record: record["SPDXID"]) + source_names = {record["fileName"].lstrip("/") for record in selected} + if len(source_names) > 1: + add_synthetic_file(final_record) + continue + + source = selected[0] + new_id = file_id(final_record["name"], final_record["algorithm"], final_record["value"]) + output_record = source.copy() + output_record["SPDXID"] = new_id + output_record["fileName"] = final_record["name"] + composed_files.append(output_record) + final_ids.add(new_id) + for record in selected: + source_to_final[record["SPDXID"]].add(new_id) + + for record in composed_files: + licenses = licenses_by_file.get(record["fileName"].lstrip("/")) + if not licenses: + continue + record["licenseInfoInFiles"] = sorted( + set(record.get("licenseInfoInFiles", [])) + .union(licenses) + - {"NONE", "NOASSERTION"} + ) + + owned_final_ids: set[str] = set(direct_final_owners) + for source_file_id, package_ids_for_file in owners_by_source_file.items(): + final_ids_for_source = source_to_final.get(source_file_id) + if not final_ids_for_source: + continue + retained_package_ids.update(package_ids_for_file) + owned_final_ids.update(final_ids_for_source) + + extension_file_ids = final_ids - owned_final_ids + + if extension_file_ids: + retained_package_ids.add(EXTENSION_PACKAGE_ID) + + composed_relationships: list[dict[str, Any]] = [] + seen_relationships: set[str] = set() + for relationship in relationships: + element_id = relationship["spdxElementId"] + related_id = relationship["relatedSpdxElement"] + if ( + (element_id in all_file_ids and element_id not in source_to_final) + or (related_id in all_file_ids and related_id not in source_to_final) + ): + continue + if ( + (element_id in all_package_ids and element_id not in retained_package_ids) + or (related_id in all_package_ids and related_id not in retained_package_ids) + ): + continue + if element_id not in source_to_final and related_id not in source_to_final: + composed_relationships.append(relationship.copy()) + continue + + element_ids = source_to_final.get(element_id, {element_id}) + related_ids = source_to_final.get(related_id, {related_id}) + for new_element_id in element_ids: + for new_related_id in related_ids: + replacement = relationship.copy() + replacement["spdxElementId"] = new_element_id + replacement["relatedSpdxElement"] = new_related_id + identity = json.dumps(replacement, sort_keys=True, separators=(",", ":")) + if identity not in seen_relationships: + seen_relationships.add(identity) + composed_relationships.append(replacement) + + composed_relationships.extend( + { + "spdxElementId": package_id, + "relationshipType": "CONTAINS", + "relatedSpdxElement": file_id_value, + } + for file_id_value, package_ids_for_file in direct_final_owners.items() + for package_id in sorted(package_ids_for_file) + ) + + output = deepcopy(builder) + os_package = debian_os_package(packages, builder_path) + output["packages"] = [ + deepcopy(package) for package in packages + if package["SPDXID"] in retained_package_ids + ] + if extension_file_ids: + output["packages"].append({ + "SPDXID": EXTENSION_PACKAGE_ID, + "copyrightText": "NOASSERTION", + "downloadLocation": "NOASSERTION", + "filesAnalyzed": True, + "licenseConcluded": "NOASSERTION", + "licenseDeclared": "NOASSERTION", + "name": f"{extension_name}-extension-artifacts", + "supplier": "NOASSERTION", + "versionInfo": "NOASSERTION", + }) + output["packages"].append(os_package) + output["files"] = composed_files + output["relationships"] = composed_relationships + if extension_file_ids: + composed_relationships.extend( + { + "spdxElementId": EXTENSION_PACKAGE_ID, + "relationshipType": "CONTAINS", + "relatedSpdxElement": file_id_value, + } + for file_id_value in sorted(extension_file_ids) + ) + package_by_id = {package["SPDXID"]: package for package in output["packages"]} + file_by_id = {record["SPDXID"]: record for record in composed_files} + licenses_by_package: defaultdict[str, set[str]] = defaultdict(set) + for relationship in composed_relationships: + package = package_by_id.get(relationship.get("spdxElementId")) + file = file_by_id.get(relationship.get("relatedSpdxElement")) + if ( + relationship.get("relationshipType") == "CONTAINS" + and package is not None + and file is not None + and package.get("filesAnalyzed", True) is not False + ): + licenses_by_package[package["SPDXID"]].update( + set(file.get("licenseInfoInFiles", [])) - {"NONE", "NOASSERTION"} + ) + for package_id, licenses in licenses_by_package.items(): + if licenses: + package = package_by_id[package_id] + package["licenseInfoFromFiles"] = sorted( + set(package.get("licenseInfoFromFiles", [])) + .union(licenses) + - {"NONE", "NOASSERTION"} + ) + for package in output["packages"]: + if package.get("licenseDeclared") not in {None, "NONE", "NOASSERTION"}: + continue + file_licenses = set(package.get("licenseInfoFromFiles", [])) - {"NONE", "NOASSERTION"} + if not file_licenses: + continue + package["licenseDeclared"] = " AND ".join( + f"({term})" if LICENSE_OPERATOR.search(term) else term + for term in sorted(file_licenses) + ) + extracted = { + item["licenseId"]: item + for item in output.get("hasExtractedLicensingInfos", []) + } + for license_id in custom_licenses: + extracted.setdefault(license_id, custom_licenses[license_id]) + output["hasExtractedLicensingInfos"] = [ + extracted[key] for key in sorted(extracted) + ] + described_ids = { + relationship["relatedSpdxElement"] + for relationship in composed_relationships + if relationship.get("spdxElementId") == output["SPDXID"] + and relationship.get("relationshipType") == "DESCRIBES" + } + for package_id in sorted(retained_package_ids): + if package_id not in described_ids: + output["relationships"].append({ + "spdxElementId": output["SPDXID"], + "relationshipType": "DESCRIBES", + "relatedSpdxElement": package_id, + }) + output["relationships"].append({ + "spdxElementId": output["SPDXID"], + "relationshipType": "DESCRIBES", + "relatedSpdxElement": os_package["SPDXID"], + }) + output["relationships"].extend( + { + "spdxElementId": os_package["SPDXID"], + "relationshipType": "CONTAINS", + "relatedSpdxElement": package["SPDXID"], + } + for package in output["packages"] + if package["SPDXID"] != os_package["SPDXID"] + and any( + reference.get("referenceType") == "purl" + and reference.get("referenceLocator", "").startswith("pkg:deb/debian/") + for reference in package.get("externalRefs", []) + ) + ) + output["name"] = f"{extension_name}-sbom" + if platform is not None: + if platform not in {"linux/amd64", "linux/arm64"}: + raise ValueError(f"unsupported target platform: {platform!r}") + creation_info = output.setdefault("creationInfo", {}) + creators = list(creation_info.get("creators", [])) + generator_version = os.getenv("SBOM_GENERATOR_REVISION") or "unknown" + generator_creator = f"Tool: {GENERATOR_NAME}-{generator_version}" + if generator_creator not in creators: + creators.append(generator_creator) + creation_info["creators"] = creators + metadata = { + "generator": GENERATOR_NAME, + "generatorVersion": generator_version, + "generatorRepository": GENERATOR_REPOSITORY, + "platform": platform, + "evidence": evidence or {}, + } + output["annotations"] = list(output.get("annotations", [])) + [{ + "annotationDate": creation_info.get("created", "1970-01-01T00:00:00Z"), + "annotationType": "OTHER", + "annotator": generator_creator, + "comment": json.dumps(metadata, sort_keys=True, separators=(",", ":")), + "spdxElementId": "SPDXRef-DOCUMENT", + }] + if platform is not None: + set_document_namespace(output, extension_name, platform) + return output + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Compose one platform-specific final-payload SPDX predicate" + ) + parser.add_argument("--builder-sbom", type=Path, required=True) + parser.add_argument("--final-inventory", type=Path, required=True) + parser.add_argument("--platform", required=True) + parser.add_argument( + "--scancode-report", + type=Path, + help="Optional ScanCode JSON report for shipped license files", + ) + parser.add_argument("--evidence", type=Path, help="Optional reproducibility evidence JSON") + parser.add_argument("--extension-name", required=True) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + + output = compose( + read_json(args.builder_sbom), + extension_name=args.extension_name, + builder_path=args.builder_sbom, + final_inventory=read_json(args.final_inventory), + platform=args.platform, + scancode_report=read_json(args.scancode_report) if args.scancode_report else {}, + evidence=read_json(args.evidence) if args.evidence else {}, + ) + args.output.parent.mkdir(parents=True, exist_ok=True) + with args.output.open("w", encoding="utf-8") as stream: + json.dump(output, stream, indent=2) + stream.write("\n") + print( + f"composed {len(output['packages'])} packages, " + f"{len(output['files'])} final files, " + f"{len(output['relationships'])} relationships", + file=sys.stderr, + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/sbom-generator/generator.py b/sbom-generator/generator.py new file mode 100644 index 00000000..06a3f151 --- /dev/null +++ b/sbom-generator/generator.py @@ -0,0 +1,467 @@ +#!/usr/bin/env python3 +"""BuildKit SBOM generator for the final extension payload. + +The scanner consumes only the files BuildKit mounts for this invocation. The +builder stage is evidence for package ownership; the final stage is inventoried +directly. The output is one BuildKit SBOM-bundle entry: an in-toto Statement +whose predicate is the composed SPDX document. BuildKit owns the OCI +attestation and image/index binding around that statement. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import platform as host_platform +import re +import shutil +import subprocess +import sys +import tempfile +import time +from pathlib import Path +from typing import Any, Sequence + +from compose import compose, set_document_namespace +from hooks import HookContext, run_augmentation_hook + + +PLATFORM_ARCHITECTURES = { + "amd64": "linux/amd64", + "x86_64": "linux/amd64", + "arm64": "linux/arm64", + "aarch64": "linux/arm64", +} + +INTOTO_STATEMENT_TYPE = "https://in-toto.io/Statement/v1" +SPDX_PREDICATE_TYPE = "https://spdx.dev/Document" +PROGRESS_INTERVAL_SECONDS = 10 + + +def require_directory(value: str | None, variable: str) -> Path: + if not value: + raise RuntimeError(f"{variable} is required") + path = Path(value) + if not path.is_dir(): + raise RuntimeError(f"{variable} does not name a directory: {path}") + return path + + +def find_builder(extra_root: Path) -> Path: + """Find the explicitly requested ``builder`` stage in BuildKit extras.""" + + candidates = [path for path in extra_root.iterdir() if path.name == "sbom-builder"] + if len(candidates) != 1 or not candidates[0].is_dir(): + names = ", ".join(sorted(path.name for path in extra_root.iterdir())) + raise RuntimeError( + "BUILDKIT_SCAN_SOURCE_EXTRAS must contain exactly one sbom-builder " + f"mount; found [{names}]" + ) + return candidates[0] + + +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def final_inventory(root: Path) -> dict[str, Any]: + """Inventory regular files and symlinks without leaving ``root``.""" + + records: list[dict[str, Any]] = [] + + def add_symlink(path: Path) -> None: + relative = path.relative_to(root).as_posix() + stat = path.lstat() + records.append({ + "name": relative, + "algorithm": "sha256", + "value": hashlib.sha256(os.readlink(path).encode()).hexdigest(), + "kind": "symlink", + "mode": stat.st_mode & 0o7777, + }) + + for directory, directory_names, file_names in os.walk(root, followlinks=False): + directory_names.sort() + file_names.sort() + for name in list(directory_names): + path = Path(directory) / name + if path.is_symlink(): + # os.walk lists symlinked directories separately. Record the + # link itself but never recurse through a target outside root. + add_symlink(path) + directory_names.remove(name) + for name in file_names: + path = Path(directory) / name + relative = path.relative_to(root).as_posix() + stat = path.lstat() + if path.is_symlink(): + # Hash the link payload, never its target. This keeps links + # outside the mounted filesystem from being followed. + add_symlink(path) + continue + elif path.is_file(): + value = sha256_file(path) + kind = "file" + else: + raise RuntimeError(f"unsupported final filesystem entry: {path}") + records.append({ + "name": relative, + "algorithm": "sha256", + "value": value, + "kind": kind, + "mode": stat.st_mode & 0o7777, + }) + if not records: + raise RuntimeError(f"final filesystem is empty: {root}") + return {"files": records} + + +def progress(message: str) -> None: + print(f"sbom-generator: {message}", file=sys.stderr, flush=True) + + +def run_command_with_progress( + command: Sequence[str], label: str +) -> subprocess.CompletedProcess: + """Run a scanner while keeping long-running phases visible in BuildKit logs.""" + + progress(f"{label} started") + started = time.monotonic() + try: + process = subprocess.Popen( + list(command), + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + except FileNotFoundError: + raise + + while True: + try: + stdout, stderr = process.communicate(timeout=PROGRESS_INTERVAL_SECONDS) + break + except subprocess.TimeoutExpired: + progress( + f"{label} still running " + f"({time.monotonic() - started:.0f}s elapsed)" + ) + + result = subprocess.CompletedProcess( + list(command), process.returncode, stdout, stderr + ) + elapsed = time.monotonic() - started + if result.returncode: + detail = (result.stderr or result.stdout or "scanner failed").strip() + raise RuntimeError(f"{' '.join(command)} failed: {detail}") + progress(f"{label} complete in {elapsed:.1f}s") + return result + + +def run_json_command( + command: Sequence[str], output: Path, label: str +) -> dict[str, Any]: + try: + run_command_with_progress( + [*command, f"spdx-json={output}"], + label, + ) + except FileNotFoundError as error: + raise RuntimeError(f"required scanner is unavailable: {command[0]}") from error + try: + with output.open(encoding="utf-8") as stream: + document = json.load(stream) + except (OSError, json.JSONDecodeError) as error: + raise RuntimeError(f"scanner did not produce valid SPDX JSON at {output}") from error + if not isinstance(document, dict): + raise RuntimeError(f"scanner output is not a JSON object: {output}") + return document + + +def scan_builder(builder: Path, temporary: Path) -> dict[str, Any]: + fixture = os.getenv("BUILDKIT_BUILDER_SPDX") + if fixture: + path = Path(fixture) + if not path.is_file(): + raise RuntimeError(f"BUILDKIT_BUILDER_SPDX does not exist: {path}") + with path.open(encoding="utf-8") as stream: + document = json.load(stream) + if not isinstance(document, dict): + raise RuntimeError("BUILDKIT_BUILDER_SPDX is not a JSON object") + progress("using supplied builder SPDX fixture") + return document + + syft = shutil.which("syft") + if not syft: + raise RuntimeError("syft is required to scan the mounted builder stage") + output = temporary / "builder.spdx.json" + document = run_json_command( + [syft, f"dir:{builder}", "--scope", "all-layers", "--quiet", "--output"], + output, + "Syft builder scan", + ) + progress(f"Syft found {len(document.get('packages', []))} builder packages") + return document + + +def scan_licenses(final_root: Path, temporary: Path) -> dict[str, Any]: + licenses = final_root / "licenses" + if not licenses.exists(): + return {"files": []} + scancode = shutil.which("scancode") + if not scancode: + raise RuntimeError("scancode is required when the final payload has /licenses") + scan_root = prepare_license_scan_root(final_root, temporary) + output = temporary / "scancode.json" + try: + subprocess.run( + [ + scancode, + "--verbose", + "--license", + "--license-references", + "--json", + str(output), + str(scan_root), + ], + stderr=subprocess.STDOUT, + check=True, + ) + except FileNotFoundError as error: + raise RuntimeError("scancode is required when the final payload has /licenses") from error + except subprocess.CalledProcessError as error: + raise RuntimeError(f"ScanCode failed with exit code {error.returncode}; see output above") from error + try: + with output.open(encoding="utf-8") as stream: + report = json.load(stream) + except (OSError, json.JSONDecodeError) as error: + raise RuntimeError(f"scancode did not produce valid JSON at {output}") from error + if not isinstance(report, dict): + raise RuntimeError("scancode output is not a JSON object") + normalize_scancode_report_paths(report, scan_root, final_root) + return report + + +def prepare_license_scan_root(final_root: Path, temporary: Path) -> Path: + """Split shipped copyright files before ScanCode scans them. + + Debian copyright files can contain thousands of ``License:`` sections. + ScanCode's per-file timeout applies to the combined file, so scan a + directory of chunks instead and map the chunk paths back afterward. + """ + + licenses = final_root / "licenses" + scan_root = temporary / "license-chunks" / "licenses" + started = time.monotonic() + license_file_count = 0 + chunk_count = 0 + for license_file in sorted(licenses.rglob("*")): + if license_file.is_symlink() or not license_file.is_file(): + continue + license_file_count += 1 + relative = license_file.relative_to(licenses) + chunk_directory = scan_root / relative + chunk_directory.mkdir(parents=True, exist_ok=True) + prefix = chunk_directory / "license-" + try: + result = subprocess.run( + [ + "csplit", + "-s", + "-z", + "-f", + str(prefix), + str(license_file), + "/^License:/", + "{*}", + ], + check=False, + capture_output=True, + text=True, + ) + except FileNotFoundError as error: + raise RuntimeError("csplit is required to prepare license files") from error + if result.returncode: + detail = (result.stderr or result.stdout or "csplit failed").strip() + raise RuntimeError(f"csplit failed for {license_file}: {detail}") + chunk_count += sum( + 1 for path in chunk_directory.glob("license-*") if path.is_file() + ) + if license_file_count % 100 == 0: + progress( + f"prepared {license_file_count} license files " + f"({chunk_count} ScanCode chunks, " + f"{time.monotonic() - started:.1f}s elapsed)" + ) + progress( + f"prepared {license_file_count} license files into {chunk_count} " + f"ScanCode chunks in {time.monotonic() - started:.1f}s" + ) + return scan_root + + +def normalize_scancode_report_paths( + report: dict[str, Any], scan_root: Path, final_root: Path +) -> None: + """Collapse split-file paths to the original final-image paths.""" + + scan_prefix = str(scan_root).rstrip("/") + final_prefix = str(final_root).rstrip("/") + chunk_suffix = re.compile(r"/license-[0-9]+$") + + def normalize(path: str) -> str: + if path == scan_prefix: + path = scan_root.name + elif path.startswith(f"{scan_prefix}/"): + path = f"{scan_root.name}/{path[len(scan_prefix) + 1:]}" + elif path == final_prefix: + path = final_root.name + elif path.startswith(f"{final_prefix}/"): + path = f"{final_root.name}/{path[len(final_prefix) + 1:]}" + return chunk_suffix.sub("", path.lstrip("/")) + + for record in report.get("files", []): + path = record.get("path") + if isinstance(path, str): + record["path"] = normalize(path) + for detection in record.get("license_detections", []): + for match in detection.get("matches", []): + from_file = match.get("from_file") + if isinstance(from_file, str): + match["from_file"] = normalize(from_file) + + +def infer_platform(builder_document: dict[str, Any]) -> str: + explicit = os.getenv("SBOM_TARGET_PLATFORM") or os.getenv("BUILDKIT_SCAN_PLATFORM") + if explicit: + if explicit not in {"linux/amd64", "linux/arm64"}: + raise RuntimeError(f"unsupported target platform: {explicit}") + return explicit + + architectures: set[str] = set() + for package in builder_document.get("packages", []): + for reference in package.get("externalRefs", []): + if reference.get("referenceType") != "purl": + continue + locator = reference.get("referenceLocator", "") + for part in locator.split("?")[-1].split("&"): + key, _, value = part.partition("=") + if key == "arch" and value in PLATFORM_ARCHITECTURES: + architectures.add(value) + platforms = {PLATFORM_ARCHITECTURES[architecture] for architecture in architectures} + if len(platforms) != 1: + raise RuntimeError( + "cannot determine target platform from builder package evidence; " + "set SBOM_TARGET_PLATFORM explicitly" + ) + return next(iter(platforms)) + + +def tool_version(command: str) -> str: + executable = shutil.which(command) + if not executable: + return "unavailable" + try: + result = subprocess.run( + [executable, "--version"], check=False, capture_output=True, text=True + ) + except OSError: + return "unavailable" + return (result.stdout or result.stderr).splitlines()[0] if (result.stdout or result.stderr) else "unknown" + + +def statement_for(predicate: dict[str, Any]) -> dict[str, Any]: + """Wrap one SPDX predicate in the statement format BuildKit unbundles. + + BuildKit v0.32's bundle exporter parses scanner output as an in-toto + Statement. It replaces the empty subject with the final image subject when + it attaches the statement, so the generator does not guess an image digest. + """ + + return { + "_type": INTOTO_STATEMENT_TYPE, + "predicateType": SPDX_PREDICATE_TYPE, + "predicate": predicate, + "subject": [], + } + + +def generate() -> Path: + source = require_directory(os.getenv("BUILDKIT_SCAN_SOURCE"), "BUILDKIT_SCAN_SOURCE") + extras = require_directory( + os.getenv("BUILDKIT_SCAN_SOURCE_EXTRAS"), "BUILDKIT_SCAN_SOURCE_EXTRAS" + ) + destination = require_directory( + os.getenv("BUILDKIT_SCAN_DESTINATION"), "BUILDKIT_SCAN_DESTINATION" + ) + builder = find_builder(extras) + if any(destination.iterdir()): + raise RuntimeError(f"scanner output directory must be empty: {destination}") + + extension_name = os.getenv("SBOM_EXTENSION_NAME", "extension") + progress(f"starting SBOM for {extension_name}") + with tempfile.TemporaryDirectory(prefix="cnpg-sbom-") as temporary_name: + temporary = Path(temporary_name) + builder_document = scan_builder(builder, temporary) + platform = infer_platform(builder_document) + progress(f"target platform: {platform}") + inventory = final_inventory(source) + progress(f"final payload inventory contains {len(inventory['files'])} files") + report = scan_licenses(source, temporary) + progress("composing SPDX document") + evidence = { + "builderSha256": sha256_file(Path(os.getenv("BUILDKIT_BUILDER_SPDX"))) + if os.getenv("BUILDKIT_BUILDER_SPDX") + else "generated-by-syft", + "platformEvidence": "builder package purl architecture", + "hostArchitecture": host_platform.machine(), + "tools": { + "syft": tool_version("syft"), + "scancode": tool_version("scancode"), + "python": sys.version.split()[0], + }, + } + predicate = compose( + builder_document, + extension_name=extension_name, + builder_path=builder, + final_inventory=inventory, + platform=platform, + scancode_report=report, + evidence=evidence, + ) + predicate = run_augmentation_hook(predicate, HookContext( + extension_name=extension_name, + platform=platform, + builder_path=builder, + final_path=source, + builder_document=builder_document, + )) + # Include any downstream augmentation in the final document identity. + set_document_namespace(predicate, extension_name, platform) + statement = statement_for(predicate) + output = destination / "final-payload.spdx.json" + progress("writing SPDX attestation") + with output.open("w", encoding="utf-8") as stream: + json.dump(statement, stream, indent=2, sort_keys=True) + stream.write("\n") + return output + + +def main() -> int: + try: + output = generate() + except (OSError, RuntimeError, ValueError, json.JSONDecodeError) as error: + print(f"sbom-generator: {error}", file=sys.stderr) + return 1 + print(f"wrote {output}", file=sys.stderr) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/sbom-generator/hooks.py b/sbom-generator/hooks.py new file mode 100644 index 00000000..cf82e18b --- /dev/null +++ b/sbom-generator/hooks.py @@ -0,0 +1,35 @@ +"""Versioned downstream extension API for the composed payload SPDX.""" + +from dataclasses import dataclass +from pathlib import Path +from runpy import run_path +from typing import Any + + +@dataclass(frozen=True) +class HookContext: + """Filesystem paths and unfiltered scan evidence available to a hook. + + Treat builder_document as read-only evidence. Filesystem paths are valid + only during this invocation. The frozen context does not freeze nested JSON. + """ + + extension_name: str + platform: str + builder_path: Path + final_path: Path + builder_document: dict[str, Any] + api_version: int = 1 + + +def run_augmentation_hook(document: dict[str, Any], context: HookContext) -> dict[str, Any]: + """Run the builder's optional hook file, or return the document unchanged.""" + + hook_path = context.builder_path / "usr/local/share/cnpg-sbom/augment_spdx.py" + if not hook_path.exists(): + return document + try: + namespace = run_path(str(hook_path)) + return namespace["augment_spdx"](document, context) + except Exception as error: + raise RuntimeError(f"SBOM hook {hook_path} failed: {error}") from error diff --git a/sbom-generator/tests/test_compose.py b/sbom-generator/tests/test_compose.py new file mode 100644 index 00000000..94e6362f --- /dev/null +++ b/sbom-generator/tests/test_compose.py @@ -0,0 +1,228 @@ +#!/usr/bin/env python3 +import copy +import hashlib +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).parents[1])) + +from compose import compose, final_inventory_files, set_document_namespace # noqa: E402 + + +def checksum(value): + return [{"algorithm": "SHA256", "checksumValue": value}] + + +def package(spdxid, name, version, purl): + return { + "SPDXID": spdxid, + "copyrightText": "NOASSERTION", + "downloadLocation": "NOASSERTION", + "externalRefs": [{ + "referenceCategory": "PACKAGE-MANAGER", + "referenceLocator": purl, + "referenceType": "purl", + }], + "filesAnalyzed": True, + "licenseConcluded": "NOASSERTION", + "licenseDeclared": "NOASSERTION", + "name": name, + "supplier": "NOASSERTION", + "versionInfo": version, + } + + +def builder_document(): + return { + "spdxVersion": "SPDX-2.3", + "SPDXID": "SPDXRef-DOCUMENT", + "name": "builder", + "creationInfo": {"created": "2026-01-01T00:00:00Z", "creators": ["Tool: syft"]}, + "packages": [ + package("SPDXRef-Package-base", "base", "1", "pkg:deb/debian/base@1?arch=amd64&distro=debian-12.15"), + package("SPDXRef-Package-extension", "extension", "2", "pkg:deb/debian/extension@2?arch=amd64&distro=debian-12.15"), + package("SPDXRef-Package-build-only", "build-only", "3", "pkg:deb/debian/build-only@3?arch=amd64&distro=debian-12.15"), + ], + "files": [ + {"SPDXID": "SPDXRef-File-base", "fileName": "usr/lib/base.so", "checksums": checksum("base")}, + {"SPDXID": "SPDXRef-File-extension", "fileName": "usr/lib/postgresql/ext.so", "checksums": checksum("extension")}, + {"SPDXID": "SPDXRef-File-build-only", "fileName": "usr/bin/cc", "checksums": checksum("build-only")}, + ], + "relationships": [ + {"spdxElementId": "SPDXRef-Package-base", "relationshipType": "CONTAINS", "relatedSpdxElement": "SPDXRef-File-base"}, + {"spdxElementId": "SPDXRef-Package-extension", "relationshipType": "CONTAINS", "relatedSpdxElement": "SPDXRef-File-extension"}, + {"spdxElementId": "SPDXRef-Package-build-only", "relationshipType": "CONTAINS", "relatedSpdxElement": "SPDXRef-File-build-only"}, + {"spdxElementId": "SPDXRef-Package-base", "relationshipType": "DEPENDENCY_OF", "relatedSpdxElement": "SPDXRef-Package-extension"}, + ], + } + + +def inventory(*entries): + return {"files": [{"name": name, "algorithm": "sha256", "value": digest} for name, digest in entries]} + + +class ComposeTest(unittest.TestCase): + def test_composes_only_shipped_files_and_owned_packages(self): + output = compose( + builder_document(), + extension_name="plr", + final_inventory=inventory( + ("lib/ext.so", "extension"), + ("generated/artifact", "generated"), + ), + platform="linux/amd64", + ) + self.assertEqual(output["name"], "plr-sbom") + self.assertEqual( + {record["name"] for record in output["packages"]}, + {"extension", "debian", "plr-extension-artifacts"}, + ) + self.assertEqual( + [record["fileName"] for record in output["files"]], + ["generated/artifact", "lib/ext.so"], + ) + self.assertNotIn("build-only", json.dumps(output)) + self.assertFalse("subject" in output) + + @patch.dict(os.environ, {"SBOM_GENERATOR_REVISION": "abc123" * 6 + "abcd"}) + def test_generator_metadata_identifies_version_and_repository(self): + revision = os.environ["SBOM_GENERATOR_REVISION"] + output = compose( + builder_document(), + extension_name="plr", + final_inventory=inventory(("lib/ext.so", "extension")), + platform="linux/amd64", + ) + generator_annotation = next( + annotation for annotation in output["annotations"] + if annotation["annotator"] == f"Tool: cnpg-sbom-generator-{revision}" + ) + metadata = json.loads(generator_annotation["comment"]) + self.assertIn(f"Tool: cnpg-sbom-generator-{revision}", output["creationInfo"]["creators"]) + self.assertEqual(metadata["generator"], "cnpg-sbom-generator") + self.assertEqual(metadata["generatorVersion"], revision) + self.assertEqual( + metadata["generatorRepository"], + "https://github.com/cnpg-extensions/postgres-extensions-containers", + ) + + def test_license_files_are_directly_mapped_to_the_named_package(self): + document = builder_document() + document["packages"].append( + package("SPDXRef-Package-copyright", "libgomp1", "1", "pkg:deb/debian/libgomp1@1?arch=amd64&distro=debian-12.15") + ) + document["files"].append({ + "SPDXID": "SPDXRef-File-copyright", + "fileName": "usr/share/doc/libgomp1/copyright", + "checksums": checksum("copyright"), + }) + document["relationships"].append({ + "spdxElementId": "SPDXRef-Package-copyright", + "relationshipType": "CONTAINS", + "relatedSpdxElement": "SPDXRef-File-copyright", + }) + output = compose( + document, + extension_name="plr", + final_inventory=inventory(("licenses/libgomp1/copyright", "copyright")), + platform="linux/amd64", + scancode_report={"files": [{ + "path": "licenses/libgomp1/copyright", + "license_detections": [{"license_expression_spdx": "GPL-2.0-only"}], + }]}, + ) + file_record = output["files"][0] + package_record = next(item for item in output["packages"] if item["name"] == "libgomp1") + self.assertEqual(file_record["licenseInfoInFiles"], ["GPL-2.0-only"]) + self.assertEqual(package_record["licenseDeclared"], "GPL-2.0-only") + + def test_unmatched_file_licenses_are_combined_on_synthetic_package(self): + output = compose( + builder_document(), + extension_name="demo", + final_inventory=inventory( + ("licenses/vendor/copyright", "vendor-license"), + ("licenses/other/copyright", "other-license"), + ), + platform="linux/amd64", + scancode_report={"files": [ + { + "path": "licenses/vendor/copyright", + "license_detections": [{"license_expression_spdx": "MIT"}], + }, + { + "path": "licenses/other/copyright", + "license_detections": [{ + "license_expression_spdx": "Apache-2.0 OR BSD-2-Clause", + }], + }, + ]}, + ) + synthetic = next( + item for item in output["packages"] + if item["name"] == "demo-extension-artifacts" + ) + self.assertEqual(synthetic["licenseInfoFromFiles"], [ + "Apache-2.0 OR BSD-2-Clause", "MIT", + ]) + self.assertEqual( + synthetic["licenseDeclared"], + "(Apache-2.0 OR BSD-2-Clause) AND MIT", + ) + + def test_platform_documents_are_deterministic_and_isolated(self): + first = compose( + builder_document(), extension_name="plr", + final_inventory=inventory(("lib/ext.so", "extension")), platform="linux/amd64", + evidence={"builderSha256": "abc"}, + ) + second = compose( + builder_document(), extension_name="plr", + final_inventory=inventory(("lib/ext.so", "extension")), platform="linux/arm64", + evidence={"builderSha256": "abc"}, + ) + self.assertNotEqual(first["documentNamespace"], second["documentNamespace"]) + self.assertEqual(first, compose( + builder_document(), extension_name="plr", + final_inventory=inventory(("lib/ext.so", "extension")), platform="linux/amd64", + evidence={"builderSha256": "abc"}, + )) + + def test_namespace_changes_with_content_and_is_stable_when_reapplied(self): + first = compose( + builder_document(), extension_name="demo", + final_inventory=inventory(("lib/ext.so", "extension")), platform="linux/amd64", + ) + changed = compose( + builder_document(), extension_name="demo", + final_inventory=inventory(("lib/ext.so", "different")), platform="linux/amd64", + ) + self.assertNotEqual(first["documentNamespace"], changed["documentNamespace"]) + original_namespace = first["documentNamespace"] + set_document_namespace(first, "demo", "linux/amd64") + self.assertEqual(first["documentNamespace"], original_namespace) + first["comment"] = "Downstream augmentation" + set_document_namespace(first, "demo", "linux/amd64") + self.assertNotEqual(first["documentNamespace"], original_namespace) + + def test_malformed_final_inventory_fails(self): + with self.assertRaises(ValueError): + final_inventory_files({"files": [{"name": "lib/ext.so", "checksums": []}]}, Path("inventory")) + + def test_builder_wrapper_is_accepted_only_as_legacy_input(self): + wrapped = { + "predicateType": "https://spdx.dev/Document", + "predicate": builder_document(), + "subject": [{"name": "lib/ext.so", "digest": {"sha256": "extension"}}], + } + output = compose(wrapped, extension_name="plr") + self.assertEqual(output["name"], "plr-sbom") + + +if __name__ == "__main__": + unittest.main() diff --git a/sbom-generator/tests/test_generator.py b/sbom-generator/tests/test_generator.py new file mode 100644 index 00000000..1205b2e2 --- /dev/null +++ b/sbom-generator/tests/test_generator.py @@ -0,0 +1,138 @@ +#!/usr/bin/env python3 +import hashlib +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).parents[1])) + +from generator import ( # noqa: E402 + final_inventory, + infer_platform, + run_command_with_progress, + scan_licenses, + statement_for, +) + + +class GeneratorTest(unittest.TestCase): + def test_inventory_hashes_regular_and_symlink_files_without_following_links(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "lib").mkdir() + payload = b"payload" + (root / "lib" / "ext.so").write_bytes(payload) + (root / "lib" / "alias.so").symlink_to("ext.so") + (root / "outside").symlink_to("/tmp", target_is_directory=True) + output = final_inventory(root) + records = {record["name"]: record for record in output["files"]} + self.assertEqual(records["lib/ext.so"]["value"], hashlib.sha256(payload).hexdigest()) + self.assertEqual(records["lib/alias.so"]["kind"], "symlink") + self.assertNotEqual(records["lib/alias.so"]["value"], records["lib/ext.so"]["value"]) + self.assertEqual(records["outside"]["kind"], "symlink") + + def test_platform_comes_from_builder_package_architecture(self): + document = {"packages": [{"externalRefs": [{ + "referenceType": "purl", + "referenceLocator": "pkg:deb/debian/base@1?arch=amd64&distro=debian-12.15", + }]}]} + self.assertEqual(infer_platform(document), "linux/amd64") + + def test_platform_ambiguity_is_rejected(self): + document = {"packages": [{"externalRefs": [ + {"referenceType": "purl", "referenceLocator": "pkg:deb/debian/a@1?arch=amd64"}, + {"referenceType": "purl", "referenceLocator": "pkg:deb/debian/b@1?arch=arm64"}, + ]}]} + with self.assertRaises(RuntimeError): + infer_platform(document) + + def test_output_is_an_spdx_intoto_statement(self): + predicate = {"SPDXID": "SPDXRef-DOCUMENT", "name": "demo"} + statement = statement_for(predicate) + self.assertEqual(statement["_type"], "https://in-toto.io/Statement/v1") + self.assertEqual(statement["predicateType"], "https://spdx.dev/Document") + self.assertEqual(statement["predicate"], predicate) + self.assertEqual(statement["subject"], []) + + def test_long_running_scanner_reports_a_heartbeat(self): + messages = [] + + class FakeProcess: + returncode = 0 + + def __init__(self): + self.calls = 0 + + def communicate(self, timeout=None): + self.calls += 1 + if self.calls == 1: + raise subprocess.TimeoutExpired(["scanner"], timeout) + return "", "" + + with patch("generator.subprocess.Popen", return_value=FakeProcess()), patch( + "generator.progress", side_effect=messages.append + ): + result = run_command_with_progress(["scanner"], "test scanner") + + self.assertEqual(result.returncode, 0) + self.assertEqual(messages[0], "test scanner started") + self.assertTrue(any("test scanner still running" in message for message in messages)) + self.assertTrue(any("test scanner complete" in message for message in messages)) + + def test_license_files_are_split_before_scancode_and_paths_are_collapsed(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "licenses").mkdir() + (root / "licenses" / "copyright").write_text("license text") + temporary = root / "temporary" + temporary.mkdir() + commands = [] + + def fake_csplit(command, **_kwargs): + if command[0] == "scancode": + return fake_scancode(command, **_kwargs) + commands.append(command) + prefix = Path(command[command.index("-f") + 1]) + prefix.with_name("license-00").write_text("license text") + return subprocess.CompletedProcess(command, 0, stdout="", stderr="") + + def fake_scancode(command, **_kwargs): + self.assertIn("--verbose", command) + self.assertNotIn("stdout", _kwargs) + self.assertEqual(_kwargs["stderr"], subprocess.STDOUT) + self.assertTrue(_kwargs["check"]) + output = Path(command[command.index("--json") + 1]) + output.write_text(json.dumps({ + "files": [{ + "type": "file", + "path": "licenses/copyright/license-00", + "license_detections": [{ + "matches": [{"from_file": "licenses/copyright/license-00"}], + "license_expression_spdx": "MIT", + }], + }], + })) + return subprocess.CompletedProcess(command, 0) + + with patch("generator.shutil.which", return_value="scancode"), patch( + "generator.subprocess.run", side_effect=fake_csplit + ): + report = scan_licenses(root, temporary) + + self.assertEqual(report["files"][0]["path"], "licenses/copyright") + self.assertEqual( + report["files"][0]["license_detections"][0]["matches"][0]["from_file"], + "licenses/copyright", + ) + split_command = next(command for command in commands if command[0] == "csplit") + self.assertIn("/^License:/", split_command) + self.assertIn("{*}", split_command) + + +if __name__ == "__main__": + unittest.main() diff --git a/sbom-generator/tests/test_hooks.py b/sbom-generator/tests/test_hooks.py new file mode 100644 index 00000000..6f5381f9 --- /dev/null +++ b/sbom-generator/tests/test_hooks.py @@ -0,0 +1,78 @@ +import json +import os +from pathlib import Path +import sys +import tempfile +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).parents[1])) + +import generator +from hooks import HookContext, run_augmentation_hook +from test_compose import builder_document + + +def write_hook(builder, source): + path = builder / "usr/local/share/cnpg-sbom/augment_spdx.py" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(source) + + +class HookTest(unittest.TestCase): + def test_absent_hook_preserves_document(self): + document = {"packages": []} + with tempfile.TemporaryDirectory() as directory: + context = HookContext("demo", "linux/amd64", Path(directory), Path(directory), {}) + self.assertIs(run_augmentation_hook(document, context), document) + + def test_hook_receives_full_builder_evidence_and_output_is_wrapped(self): + evidence = builder_document() + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source, extras, destination = root / "source", root / "extras", root / "output" + source.mkdir() + builder = extras / "sbom-builder" + builder.mkdir(parents=True) + destination.mkdir() + (source / "artifact").write_text("payload") + write_hook(builder, ''' +from hooks import HookContext + +def augment_spdx(document, context: HookContext): + assert context.api_version == 1 + assert context.platform == "linux/amd64" + assert context.builder_path.is_dir() + assert context.final_path.is_dir() + assert "build-only" not in {p["name"] for p in document["packages"]} + assert "build-only" in {p["name"] for p in context.builder_document["packages"]} + assert document["files"][0]["SPDXID"] + assert document["files"][0]["checksums"] + return {**document, "comment": "downstream dependency version and license information"} +''') + with patch.dict(os.environ, { + "BUILDKIT_SCAN_SOURCE": str(source), + "BUILDKIT_SCAN_SOURCE_EXTRAS": str(extras), + "BUILDKIT_SCAN_DESTINATION": str(destination), + "BUILDKIT_BUILDER_SPDX": "", + "SBOM_TARGET_PLATFORM": "linux/amd64", + }), patch.object(generator, "scan_builder", return_value=evidence), \ + patch.object(generator, "tool_version", return_value="test"): + statement = json.loads(generator.generate().read_text()) + self.assertEqual(statement["predicate"]["comment"], + "downstream dependency version and license information") + self.assertEqual(statement["subject"], []) + self.assertEqual([f["fileName"] for f in statement["predicate"]["files"]], ["artifact"]) + + def test_hook_errors_are_not_silently_ignored(self): + sources = ( + "def augment_spdx(document, context):\n raise ValueError('evidence missing')\n", + "# Missing entry point\n", + "invalid python syntax!\n", + ) + for source in sources: + with self.subTest(source=source), tempfile.TemporaryDirectory() as directory: + builder = Path(directory) + write_hook(builder, source) + with self.assertRaisesRegex(RuntimeError, "SBOM hook"): + run_augmentation_hook({}, HookContext("demo", "linux/amd64", builder, builder, {}))