From d139d66c3dbac0ff46ff0a0de9357205a9f56c9c Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Thu, 22 Oct 2026 16:23:00 +0200 Subject: [PATCH] feat: judge, sign and verify netOrk's images in one place (NetOrk/netork#1390) The CIS gate, the SBOM, the cosign signature and the matcher's verdict lived in NetOrk/base-os, the head of the chain. Every link below it -- base-python, engine, ui, satellite, the service images -- needs the same, so the tools move here, and every image repository pins them by commit. - image_tools: cis.py, cis_os.sh, sbom.py and verify.py from base-os. The image's own repository holds its cis.toml and packages.allow; both default to the working directory. - bin/sign.sh signs whatever platforms a reference holds: base-os's index of amd64 and arm64 (EXPECT_PLATFORMS) or a dev image's single manifest (image_tools/platforms.py). - bin/verify-base.sh checks every FROM image of a Dockerfile against cosign.pub before a build; one without a digest is refused (image_tools/dockerfile.py). - bin/install-cosign.sh installs cosign v3.1.3 for amd64 and arm64 runners, checked against the release checksums. - actions/verify-base, judge, sign wrap them for the workflows. - cosign.pub is the public half of the COSIGN_PRIVATE_KEY_B64 org secret, the key rotated on 2026-10-22. --- .gitea/workflows/ci.yml | 26 +++ .gitignore | 5 + LICENSE | 21 +++ README.md | 77 +++++++++ actions/judge/action.yml | 45 +++++ actions/sign/action.yml | 28 ++++ actions/verify-base/action.yml | 20 +++ bin/install-cosign.sh | 25 +++ bin/sign.sh | 60 +++++++ bin/verify-base.sh | 14 ++ bin/verify-signature.sh | 18 ++ cosign-signing.json | 9 + cosign.pub | 4 + image_tools/__init__.py | 0 image_tools/cis.py | 297 +++++++++++++++++++++++++++++++++ image_tools/cis_os.sh | 171 +++++++++++++++++++ image_tools/dockerfile.py | 59 +++++++ image_tools/platforms.py | 64 +++++++ image_tools/sbom.py | 60 +++++++ image_tools/verify.py | 221 ++++++++++++++++++++++++ pyproject.toml | 23 +++ tests/test_actions.py | 38 +++++ tests/test_cis.py | 251 ++++++++++++++++++++++++++++ tests/test_dockerfile.py | 60 +++++++ tests/test_platforms.py | 82 +++++++++ tests/test_signing.py | 228 +++++++++++++++++++++++++ tests/test_verify.py | 179 ++++++++++++++++++++ 27 files changed, 2085 insertions(+) create mode 100644 .gitea/workflows/ci.yml create mode 100644 .gitignore create mode 100644 LICENSE create mode 100644 README.md create mode 100644 actions/judge/action.yml create mode 100644 actions/sign/action.yml create mode 100644 actions/verify-base/action.yml create mode 100755 bin/install-cosign.sh create mode 100755 bin/sign.sh create mode 100755 bin/verify-base.sh create mode 100755 bin/verify-signature.sh create mode 100644 cosign-signing.json create mode 100644 cosign.pub create mode 100644 image_tools/__init__.py create mode 100644 image_tools/cis.py create mode 100644 image_tools/cis_os.sh create mode 100644 image_tools/dockerfile.py create mode 100644 image_tools/platforms.py create mode 100644 image_tools/sbom.py create mode 100644 image_tools/verify.py create mode 100644 pyproject.toml create mode 100644 tests/test_actions.py create mode 100644 tests/test_cis.py create mode 100644 tests/test_dockerfile.py create mode 100644 tests/test_platforms.py create mode 100644 tests/test_signing.py create mode 100644 tests/test_verify.py diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml new file mode 100644 index 0000000..b5820d0 --- /dev/null +++ b/.gitea/workflows/ci.yml @@ -0,0 +1,26 @@ +name: CI + +on: + push: + branches: ['main'] + pull_request: + branches: ['**'] + +jobs: + test: + name: Tests, lint + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.12' + - name: Install + # Pinned: a newer ruff or shellcheck can turn an untouched tree red. + run: pip install pytest==8.4.2 ruff==0.14.0 pyyaml==6.0.2 shellcheck-py==0.11.0.1 + - name: ruff + run: ruff format --check . && ruff check . + - name: shellcheck + run: shellcheck bin/*.sh image_tools/cis_os.sh + - name: pytest + run: pytest -q diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..dc94c12 --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +__pycache__/ +*.egg-info/ +.pytest_cache/ +.ruff_cache/ +build/ diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..ac8e1dd --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Christian Manivong + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..e277672 --- /dev/null +++ b/README.md @@ -0,0 +1,77 @@ +# image-tools + +Judges, signs and verifies the container images of [netOrk](https://git.netork.io/NetOrk/netork). +Every image repository of NetOrk uses these tools at a pinned commit, so the same rules apply +to every link of the chain: + +``` +netork/base-os ──► netork/base-python ──► netork/engine, netork/satellite + └─► netork/ui + └─► postgres, valkey, signal-api, apt-cacher-ng, unbound +``` + +Each link checks the signature of the image it builds on, is judged against CIS, and is signed +with its CIS result and SBOM attached (NetOrk/netork#1390). + +## What it does + +| | | +|---|---| +| **CIS gate** (`image_tools/cis.py`) | CIS Docker Benchmark 1.6.0 §4 and the CIS Debian Linux 12 controls that make sense in a container. Trivy measures what it automates; `image_tools/cis_os.sh` measures from inside the image. The image's repository decides the rest in its `cis.toml` and names the packages it may hold in `packages.allow`. | +| **SBOM** (`image_tools/sbom.py`) | The image's CycloneDX SBOM, as Trivy reads it. | +| **Signature** (`bin/sign.sh`) | A cosign signature on the reference and on every platform image behind it, with the CIS result and the SBOM as in-toto attestations. Nothing goes to a public transparency log. | +| **Verification** (`bin/verify-base.sh`, `bin/verify-signature.sh`) | Every `FROM` image of a Dockerfile is pinned by digest and signed with the key in `cosign.pub`. | +| **Verdict** (`image_tools/verify.py`) | Whether a rebuild improves on the pinned image, by netOrk's matcher (kb-server), for a base's nightly. | + +## Using it in a workflow + +```yaml +- name: Verify the base images + uses: https://git.netork.io/NetOrk/image-tools/actions/verify-base@ + with: + dockerfiles: docker/engine/Dockerfile + +# build the image, then: +- name: Judge it + uses: https://git.netork.io/NetOrk/image-tools/actions/judge@ + with: + image: registry.netork.io/netork/engine:main-abc1234 + arch: amd64 + +# push it, then: +- name: Sign it + uses: https://git.netork.io/NetOrk/image-tools/actions/sign@ + env: + COSIGN_PRIVATE_KEY_B64: ${{ secrets.COSIGN_PRIVATE_KEY_B64 }} + COSIGN_PASSWORD: ${{ secrets.COSIGN_PASSWORD }} + with: + ref: registry.netork.io/netork/engine@sha256:… +``` + +The judge action reads `cis.toml` and `packages.allow` from the working directory by default. +The runner needs Docker, Python 3.11 or newer, and `curl`; cosign is installed by the actions, +pinned and checked against its release checksum (`bin/install-cosign.sh`). + +## Verifying an image yourself + +```bash +cosign verify --key cosign.pub --insecure-ignore-tlog registry.netork.io/netork/base-os@sha256:… +cosign verify-attestation --key cosign.pub --insecure-ignore-tlog \ + --type https://netork.io/attestations/cis/v1 registry.netork.io/netork/base-os@sha256:… +``` + +`cosign.pub` is the public half of the `COSIGN_PRIVATE_KEY_B64` organization secret. The private +key comes base64 in one line: the runner masks a secret only where it appears whole. + +## Development + +```bash +pip install pytest ruff pyyaml shellcheck-py +pytest -q +ruff format --check . && ruff check . +shellcheck bin/*.sh image_tools/cis_os.sh +``` + +## License + +MIT, see [LICENSE](LICENSE). diff --git a/actions/judge/action.yml b/actions/judge/action.yml new file mode 100644 index 0000000..9fa1a1a --- /dev/null +++ b/actions/judge/action.yml @@ -0,0 +1,45 @@ +name: Judge an image +description: >- + The CIS gate (image_tools/cis.py) and the CycloneDX SBOM (image_tools/sbom.py) + of one platform image, kept as /cis-.json, cis-.md and + sbom-.cdx.json for bin/sign.sh. The step fails when CIS is red. +inputs: + image: + description: The image to judge, as Docker can run it (local tag or reference) + required: true + arch: + description: Its platform, amd64 or arm64 + required: true + decisions: + description: The image repository's CIS decisions + default: cis.toml + packages: + description: The packages the image may hold + default: packages.allow + out: + description: Where the results go + default: attest +runs: + using: composite + steps: + - name: CIS + shell: bash + env: + IMAGE: ${{ inputs.image }} + ARCH: ${{ inputs.arch }} + DECISIONS: ${{ inputs.decisions }} + PACKAGES: ${{ inputs.packages }} + OUT: ${{ inputs.out }} + PYTHONPATH: ${{ github.action_path }}/../.. + run: | + mkdir -p "$OUT" + python3 -m image_tools.cis "$IMAGE" --decisions "$DECISIONS" --packages "$PACKAGES" \ + --json "$OUT/cis-$ARCH.json" --report "$OUT/cis-$ARCH.md" + - name: SBOM + shell: bash + env: + IMAGE: ${{ inputs.image }} + ARCH: ${{ inputs.arch }} + OUT: ${{ inputs.out }} + PYTHONPATH: ${{ github.action_path }}/../.. + run: python3 -m image_tools.sbom "$IMAGE" "$OUT/sbom-$ARCH.cdx.json" diff --git a/actions/sign/action.yml b/actions/sign/action.yml new file mode 100644 index 0000000..a38cc3a --- /dev/null +++ b/actions/sign/action.yml @@ -0,0 +1,28 @@ +name: Sign an image +description: >- + Sign a pushed image and every platform image behind it with netOrk's key, and + attach what each was judged by (bin/sign.sh). The caller passes the secrets + COSIGN_PRIVATE_KEY_B64 and COSIGN_PASSWORD as env of this step. +inputs: + ref: + description: The pushed image, @ + required: true + dir: + description: The judge action's results, cis-.json and sbom-.cdx.json + default: attest + expect-platforms: + description: The platforms the reference must hold, e.g. "amd64 arm64"; empty takes what is there + default: "" +runs: + using: composite + steps: + - name: Install cosign + shell: bash + run: bash "${{ github.action_path }}/../../bin/install-cosign.sh" + - name: Sign and attest + shell: bash + env: + REF: ${{ inputs.ref }} + DIR: ${{ inputs.dir }} + EXPECT_PLATFORMS: ${{ inputs.expect-platforms }} + run: bash "${{ github.action_path }}/../../bin/sign.sh" "$REF" "$DIR" diff --git a/actions/verify-base/action.yml b/actions/verify-base/action.yml new file mode 100644 index 0000000..c154450 --- /dev/null +++ b/actions/verify-base/action.yml @@ -0,0 +1,20 @@ +name: Verify the base images +description: >- + Every image the Dockerfiles build on is signed with netOrk's key, checked + before the build (NetOrk/netork#1390, bin/verify-base.sh). +inputs: + dockerfiles: + description: The Dockerfiles, separated by spaces, relative to the workspace + required: true +runs: + using: composite + steps: + - name: Install cosign + shell: bash + run: bash "${{ github.action_path }}/../../bin/install-cosign.sh" + - name: Verify + shell: bash + env: + DOCKERFILES: ${{ inputs.dockerfiles }} + # shellcheck disable=SC2086 # the list splits on purpose + run: bash "${{ github.action_path }}/../../bin/verify-base.sh" $DOCKERFILES diff --git a/bin/install-cosign.sh b/bin/install-cosign.sh new file mode 100755 index 0000000..df3c715 --- /dev/null +++ b/bin/install-cosign.sh @@ -0,0 +1,25 @@ +#!/bin/bash +# install-cosign.sh [dir]: cosign, pinned, checked against its release checksum. +# +# Bump the version and both checksums together, from the release's +# cosign_checksums.txt. +set -euo pipefail +VERSION=v3.1.3 +SHA256_AMD64=4629c757b7618056f8ddd7e2625ae9fdd94c0372a65049520bc7d9df9efc7f71 +SHA256_ARM64=c5d324e091826b0d7a78eb16fef316450b4eb9aaec045611c08ba06f5e73220a +dir=${1:-/usr/local/bin} + +case $(uname -m) in + x86_64) arch=amd64 sha=$SHA256_AMD64 ;; + aarch64) arch=arm64 sha=$SHA256_ARM64 ;; + *) echo "no cosign checksum for $(uname -m)" >&2; exit 1 ;; +esac +if [ -x "$dir/cosign" ] && echo "$sha $dir/cosign" | sha256sum -c - >/dev/null 2>&1; then + exit 0 +fi +tmp=$(mktemp) +curl -fsSLo "$tmp" "https://github.com/sigstore/cosign/releases/download/$VERSION/cosign-linux-$arch" +echo "$sha $tmp" | sha256sum -c - >/dev/null +install -m 0755 "$tmp" "$dir/cosign" +rm -f "$tmp" +echo "cosign $VERSION ($arch) in $dir" diff --git a/bin/sign.sh b/bin/sign.sh new file mode 100755 index 0000000..b72ec63 --- /dev/null +++ b/bin/sign.sh @@ -0,0 +1,60 @@ +#!/bin/bash +# sign.sh @ : sign a published image and attach what it was judged by. +# +# The reference and every platform image behind it get a cosign signature made +# with the key in COSIGN_PRIVATE_KEY_B64 (its password in COSIGN_PASSWORD). Each +# platform image gets its CIS result (/cis-.json, image_tools/cis.py) +# and its CycloneDX SBOM (/sbom-.cdx.json, image_tools/sbom.py) as +# in-toto attestations. This takes the place of Docker Content Trust (CIS Docker +# 4.5), as cosign does for Docker's Hardened Images. +# +# The platforms are whatever the reference holds (image_tools/platforms.py): an +# index of amd64 and arm64, or one manifest. EXPECT_PLATFORMS ("amd64 arm64") +# makes any other set an error. +# +# Nothing goes to a public transparency log: cosign-signing.json names none, so +# the names in this registry stay inside. Everything is verified against +# cosign.pub before it returns. +set -euo pipefail +: "${COSIGN_PRIVATE_KEY_B64:?is empty: the secret is missing}" "${COSIGN_PASSWORD:?is empty: the secret is missing}" +# The key comes base64 in one line: the runner prints a step's env and masks a +# secret only where it appears whole, and a PEM key's lines did not (2026-10-22). +COSIGN_PRIVATE_KEY=$(printf '%s' "$COSIGN_PRIVATE_KEY_B64" | base64 -d) +export COSIGN_PRIVATE_KEY +ref=$1 +dir=$(cd "$2" && pwd) +cd "$(dirname "$0")/.." + +CIS_TYPE=https://netork.io/attestations/cis/v1 +# quietly : cosign's chatter only when it fails. +quietly() { local out; out=$("$@" 2>&1) || { echo "$out" >&2; return 1; }; } + +case $ref in *@sha256:*) ;; *) echo "$ref is not pinned by digest" >&2; exit 1 ;; esac +image=${ref%@*} +case ${image##*/} in *:*) image=${image%:*} ;; esac # the tag, not a registry port + +quietly cosign sign --yes --recursive --key env://COSIGN_PRIVATE_KEY \ + --signing-config cosign-signing.json "$ref" +quietly cosign verify --key cosign.pub --insecure-ignore-tlog "$ref" + +raw=$(mktemp) config=$(mktemp) +trap 'rm -f "$raw" "$config"' EXIT +docker buildx imagetools inspect "$ref" --raw >"$raw" +docker buildx imagetools inspect "$ref" --format '{{json .Image}}' >"$config" +platforms=$(python3 -m image_tools.platforms "${ref#*@}" "$raw" "$config") +if [ -n "${EXPECT_PLATFORMS:-}" ]; then + got=$(echo "$platforms" | cut -d' ' -f1 | sort | xargs) + want=$(echo "$EXPECT_PLATFORMS" | tr ' ' '\n' | sort | xargs) + [ "$got" = "$want" ] || { echo "expected $want in $ref, found $got" >&2; exit 1; } +fi + +while read -r arch digest; do + for attestation in "$CIS_TYPE cis-$arch.json" "cyclonedx sbom-$arch.cdx.json"; do + read -r type file <<<"$attestation" + quietly cosign attest --yes --key env://COSIGN_PRIVATE_KEY --signing-config cosign-signing.json \ + --type "$type" --predicate "$dir/$file" "$image@$digest" + quietly cosign verify-attestation --key cosign.pub --insecure-ignore-tlog \ + --type "$type" "$image@$digest" + done + echo "signed and attested: $image@$digest ($arch)" +done <<<"$platforms" diff --git a/bin/verify-base.sh b/bin/verify-base.sh new file mode 100755 index 0000000..fc8e1c3 --- /dev/null +++ b/bin/verify-base.sh @@ -0,0 +1,14 @@ +#!/bin/bash +# verify-base.sh ...: every image the files build on is signed with +# netOrk's key, checked before the build (NetOrk/netork#1390). +# +# The images are the FROM lines' references (image_tools/dockerfile.py): stages +# of the same file and scratch are none, and one without a digest is refused. +set -euo pipefail +tools=$(cd "$(dirname "$0")/.." && pwd) +[ $# -gt 0 ] || { echo "usage: verify-base.sh ..." >&2; exit 2; } +files=() +for f in "$@"; do files+=("$(cd "$(dirname "$f")" && pwd)/$(basename "$f")"); done +mapfile -t bases < <(cd "$tools" && python3 -m image_tools.dockerfile "${files[@]}") +[ ${#bases[@]} -gt 0 ] || { echo "no image to verify in $*" >&2; exit 1; } +"$tools/bin/verify-signature.sh" "${bases[@]}" diff --git a/bin/verify-signature.sh b/bin/verify-signature.sh new file mode 100755 index 0000000..db8a6e2 --- /dev/null +++ b/bin/verify-signature.sh @@ -0,0 +1,18 @@ +#!/bin/bash +# verify-signature.sh @...: each one is signed with netOrk's key. +# +# The key is cosign.pub beside this script's repository, the one public half of +# the COSIGN_PRIVATE_KEY_B64 secret every image of NetOrk is signed with. A +# reference without a digest is refused: a tag can move after the check. +set -euo pipefail +cd "$(dirname "$0")/.." +[ $# -gt 0 ] || { echo "usage: verify-signature.sh @..." >&2; exit 2; } +for ref in "$@"; do + case $ref in *@sha256:*) ;; *) echo "$ref is not pinned by digest" >&2; exit 1 ;; esac + if ! out=$(cosign verify --key cosign.pub --insecure-ignore-tlog "$ref" 2>&1); then + echo "$out" >&2 + echo "not signed with netOrk's key: $ref" >&2 + exit 1 + fi + echo "verified: $ref" +done diff --git a/cosign-signing.json b/cosign-signing.json new file mode 100644 index 0000000..da91d87 --- /dev/null +++ b/cosign-signing.json @@ -0,0 +1,9 @@ +{ + "mediaType": "application/vnd.dev.sigstore.signingconfig.v0.2+json", + "caUrls": [], + "oidcUrls": [], + "rekorTlogUrls": [], + "tsaUrls": [], + "rekorTlogConfig": {"selector": "ANY"}, + "tsaConfig": {"selector": "ANY"} +} diff --git a/cosign.pub b/cosign.pub new file mode 100644 index 0000000..9a80d55 --- /dev/null +++ b/cosign.pub @@ -0,0 +1,4 @@ +-----BEGIN PUBLIC KEY----- +MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE6CqcdpadZWHM2pmd2QgmXAUSPX6u +ih4JMspCDo+hye3cccnVZx1fNlyyP0NRQhLlEk+czT5eX9HqPnNFD84i4Q== +-----END PUBLIC KEY----- diff --git a/image_tools/__init__.py b/image_tools/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/image_tools/cis.py b/image_tools/cis.py new file mode 100644 index 0000000..d31a0c0 --- /dev/null +++ b/image_tools/cis.py @@ -0,0 +1,297 @@ +"""Is the image what CIS asks of it? (CIS Docker Benchmark 1.6.0 §4, CIS Debian Linux 12) + +Two measurements, one verdict: + +- Trivy's ``docker-cis-1.6.0`` spec, for the controls it automates (4.1, 4.4, + 4.6, 4.7, 4.9, 4.10); the rest it calls "Manual"; +- ``image_tools/cis_os.sh``, run inside the image as root, for what of that can be + seen from inside (4.3, 4.8, 4.11) and the CIS Debian controls that make sense + in a container. + +The image's own repository decides the rest in ``cis.toml``: ``[[manual]]`` is +met by how the image is made, ``[[exception]]`` fails on purpose, with a reason, +until it expires. ``packages.allow`` there names the packages it may hold. The gate is red when a control fails without an exception, an +exception has expired, a control is neither measured nor decided, a decision +names a control nothing knows, or a measurement could not be taken. + + python -m image_tools.cis [--decisions cis.toml] [--packages packages.allow] \ + --json cis.json --report cis.md + +Both files default to the working directory: the image's repository. + +Standard library only; Trivy runs as the container image_tools/verify.py pins. +""" + +from __future__ import annotations + +import argparse +import datetime +import json +import subprocess +import tomllib +from collections.abc import Callable +from dataclasses import asdict, dataclass, field +from pathlib import Path + +from image_tools.verify import TIMEOUT, TRIVY_IMAGE + +OS_CHECKS = Path(__file__).with_name("cis_os.sh") +SPEC = "docker-cis-1.6.0" + +PASS, FAIL, UNMEASURED = "pass", "fail", "unmeasured" +#: What the gate makes of a control; the first three keep it green. +GREEN = frozenset({"pass", "manual", "excepted"}) + + +@dataclass +class Control: + id: str + title: str + result: str + detail: str = "" + state: str = "" + reason: str = "" + + +@dataclass +class Decision: + kind: str + reason: str + expires: datetime.date | None = None + + +@dataclass +class Report: + ok: bool + controls: list[Control] = field(default_factory=list) + stale: list[str] = field(default_factory=list) + unknown: list[str] = field(default_factory=list) + error: str | None = None + + def to_json(self) -> str: + return json.dumps(asdict(self), indent=2, default=str) + + def failing(self) -> set[str]: + return {c.id for c in self.controls if c.result == FAIL} + + +def from_trivy(summary: dict) -> list[Control]: + """Trivy's compliance summary: no ``TotalFail`` means it did not measure.""" + controls = [] + for c in summary.get("SummaryControls") or []: + fails = c.get("TotalFail") + result = UNMEASURED if fails is None else (FAIL if fails else PASS) + detail = f"{fails} failed (Trivy)" if fails else "" + controls.append(Control(f"docker:{c['ID']}", c.get("Name", ""), result, detail)) + return controls + + +def from_os(text: str) -> list[Control]: + """cis_os.sh's lines: id, PASS|FAIL, title, what fails.""" + controls = [] + for line in text.splitlines(): + if not line.strip(): + continue + cid, result, title, detail = (line.split("\t") + ["", "", ""])[:4] + if result not in ("PASS", "FAIL"): + raise ValueError(f"cis_os.sh said: {line!r}") + controls.append(Control(cid, title, result.lower(), detail)) + return controls + + +def merge(*sources: list[Control]) -> list[Control]: + """One control per id, in order of first appearance; a failure wins, then a pass.""" + rank = {FAIL: 2, PASS: 1, UNMEASURED: 0} + merged: dict[str, Control] = {} + for controls in sources: + for c in controls: + known = merged.get(c.id) + if known is None or rank[c.result] > rank[known.result]: + if known is not None and not c.title: + c.title = known.title + merged[c.id] = c + return list(merged.values()) + + +def load_decisions(text: str) -> dict[str, Decision]: + data = tomllib.loads(text) + decisions = {} + for kind in ("manual", "exception"): + for entry in data.get(kind) or []: + if entry["id"] in decisions: + raise ValueError(f"{entry['id']} is decided twice") + if not entry.get("reason", "").strip(): + raise ValueError(f"{entry['id']} has no reason") + if not isinstance(entry.get("expires", datetime.date.min), datetime.date): + raise ValueError(f"{entry['id']}: expires is a date, not {entry['expires']!r}") + decisions[entry["id"]] = Decision(kind, entry["reason"], entry.get("expires")) + return decisions + + +def judge(controls: list[Control], decisions: dict[str, Decision], today: datetime.date) -> Report: + """The gate on the measured *controls* and what ``cis.toml`` decided.""" + stale = [] + for c in controls: + d = decisions.get(c.id) + c.reason = d.reason if d else "" + if d and d.kind == "manual": + c.state = "manual" + elif c.result == PASS: + c.state = "pass" + if d: + stale.append(c.id) + elif d and d.expires and d.expires < today: + c.state = "expired" + elif d: + c.state = "excepted" + else: + c.state = "fail" if c.result == FAIL else "undecided" + unknown = sorted(set(decisions) - {c.id for c in controls}) + ok = bool(controls) and not unknown and all(c.state in GREEN for c in controls) + return Report(ok, controls, stale, unknown) + + +def measure( + ref: str, + packages: list[str], + *, + trivy: Callable[[str], dict] | None = None, + os_checks: Callable[[str, list[str]], str] | None = None, +) -> list[Control]: + trivy = trivy or trivy_compliance + os_checks = os_checks or run_os_checks + return merge(from_trivy(trivy(ref)), from_os(os_checks(ref, packages))) + + +def check( + ref: str, + decisions: dict[str, Decision], + packages: list[str], + today: datetime.date, + **seams, +) -> Report: + """Measure *ref* and judge it; never raises.""" + try: + return judge(measure(ref, packages, **seams), decisions, today) + except (OSError, ValueError, KeyError, subprocess.SubprocessError) as exc: + return Report(False, error=f"{type(exc).__name__}: {exc}") + + +STATE_MARK = { + "pass": "✅ pass", + "manual": "📝 manual", + "excepted": "⚠️ exception", + "fail": "❌ fail", + "expired": "❌ exception expired", + "undecided": "❌ not measured, not decided", +} + + +def render(report: Report, ref: str) -> str: + head = "**CIS: green**" if report.ok else "**CIS: red**" + lines = [f"{head} for `{ref}` (CIS Docker 1.6.0 §4, CIS Debian Linux 12).", ""] + if report.error: + lines += [f"The measurement could not be taken: {report.error}", ""] + if report.unknown: + lines += [f"Decided in cis.toml but measured by nothing: {', '.join(report.unknown)}", ""] + if report.stale: + lines += [f"Exceptions for controls that pass now: {', '.join(report.stale)}", ""] + if report.controls: + lines += ["| Control | | Title | Detail |", "|---|---|---|---|"] + for c in report.controls: + detail = c.reason if c.state in ("manual", "excepted") else c.detail + detail = detail.replace("|", "\\|").replace("\n", " ") + lines.append(f"| {c.id} | {STATE_MARK.get(c.state, c.state)} | {c.title} | {detail} |") + return "\n".join(lines).rstrip() + "\n" + + +def load_packages(text: str) -> list[str]: + return [ln.strip() for ln in text.splitlines() if ln.strip() and not ln.startswith("#")] + + +def trivy_compliance(ref: str) -> dict: # pragma: no cover - container seam + """Trivy's summary of *ref* against the spec, with the checks it was released with.""" + proc = subprocess.run( + [ + "docker", + "run", + "--rm", + "-v", + "/var/run/docker.sock:/var/run/docker.sock", + "-e", + "TRIVY_USERNAME", + "-e", + "TRIVY_PASSWORD", + TRIVY_IMAGE, + "image", + "--quiet", + "--disable-telemetry", + "--skip-version-check", + "--skip-check-update", + # 4.4 is the matcher's (cis.toml); no vulnerability DB to fetch + "--scanners", + "misconfig,secret", + "--compliance", + SPEC, + "--format", + "json", + ref, + ], + capture_output=True, + text=True, + timeout=TIMEOUT, + check=True, + ) + return json.loads(proc.stdout) + + +def run_os_checks(ref: str, packages: list[str]) -> str: # pragma: no cover - container seam + """cis_os.sh inside *ref*, as root, fed on stdin: nothing to mount.""" + proc = subprocess.run( + [ + "docker", + "run", + "--rm", + "-i", + "--user", + "0", + "--network", + "none", + "-e", + "CIS_PACKAGES=" + " ".join(packages), + "--entrypoint", + "sh", + ref, + "-s", + ], + input=OS_CHECKS.read_text(), + capture_output=True, + text=True, + timeout=TIMEOUT, + check=True, + ) + return proc.stdout + + +def main() -> int: # pragma: no cover - CLI seam + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("ref") + parser.add_argument("--decisions", default="cis.toml") + parser.add_argument("--packages", default="packages.allow") + parser.add_argument("--json") + parser.add_argument("--report") + args = parser.parse_args() + decisions = load_decisions(Path(args.decisions).read_text()) + packages = load_packages(Path(args.packages).read_text()) + report = check(args.ref, decisions, packages, datetime.date.today()) + text = render(report, args.ref) + if args.json: + Path(args.json).write_text(report.to_json()) + if args.report: + Path(args.report).write_text(text) + print(text) + return 0 if report.ok else 1 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/image_tools/cis_os.sh b/image_tools/cis_os.sh new file mode 100644 index 0000000..25e6335 --- /dev/null +++ b/image_tools/cis_os.sh @@ -0,0 +1,171 @@ +#!/bin/sh +# cis_os.sh: the CIS controls image_tools/cis.py measures from inside the image. +# +# Runs as root in the image, fed on stdin (docker run -i --user 0 ... sh -s), so +# nothing has to be mounted from a runner whose paths the host daemon does not +# know. Reads nothing from outside but CIS_PACKAGES, the package names the image +# may hold. Prints one line per control: +# +# TAB PASS|FAIL TAB TAB <what fails> +# +# docker:* are the CIS Docker Benchmark 1.6.0 controls Trivy leaves to "Manual" +# and that can be seen from inside; debian:* are the CIS Debian Linux 12 v1.1.0 +# controls that make sense in a container. Boot, partitions, kernel parameters, +# services, logging, audit and sshd belong to the host (infrastructure +# BAUSTANDARD §F) and are not measured here. +set -u + +# check <id> <title> <function>: PASS when the function names no offender. +check() { + out=$("$3" 2>/dev/null | head -n 20 | tr '\n' ' ') + if [ -z "$out" ]; then + printf '%s\tPASS\t%s\t\n' "$1" "$2" + else + printf '%s\tFAIL\t%s\t%s\n' "$1" "$2" "${out% }" + fi +} + +login_defs() { awk -v k="$1" '$1 == k { v = $2 } END { print v }' /etc/login.defs; } + +# Shells one can log in with: /etc/shells without the nologin ones. +SHELLS=$(grep '^/' /etc/shells | grep -v nologin) + +# --- CIS Docker Benchmark 1.6.0, section 4 ----------------------------------- + +extra_packages() { + dpkg-query -W -f '${Package}\n' | while read -r p; do + case " ${CIS_PACKAGES:-} " in *" $p "*) ;; *) echo "$p" ;; esac + done +} + +setid_files() { find / -xdev -type f -perm /6000; } + +unverified_sources() { + grep -rEil 'trusted=yes|Trusted: *yes|AllowUnauthenticated|AllowInsecureRepositories|AllowDowngradeToInsecureRepositories' /etc/apt + for f in /etc/apt/sources.list.d/*.sources; do + [ -e "$f" ] || continue + awk -v f="$f" 'BEGIN { RS = "" } index($0, "Signed-By:") == 0 { print f }' "$f" + done + grep -sE '^[[:space:]]*deb[[:space:]]' /etc/apt/sources.list /etc/apt/sources.list.d/*.list \ + | grep -v 'signed-by=' +} + +check docker:4.3 "Unnecessary packages are not installed (packages.allow)" extra_packages +check docker:4.8 "setuid and setgid permissions are removed" setid_files +check docker:4.11 "Only verified packages are installed" unverified_sources + +# --- CIS Debian Linux 12 v1.1.0 ---------------------------------------------- + +# in_range <key> <min> [max]: names the key unless min <= its value <= max. +in_range() { + v=$(login_defs "$1") + case $v in '' | *[!0-9]*) echo "$1=${v:-unset}"; return ;; esac + if [ "$v" -lt "$2" ] || [ "$v" -gt "${3:-$v}" ]; then echo "$1=$v"; fi +} +pass_max_days() { in_range PASS_MAX_DAYS 1 365; } +pass_min_days() { in_range PASS_MIN_DAYS 1; } +pass_warn_age() { in_range PASS_WARN_AGE 7; } +weak_hashing() { + v=$(login_defs ENCRYPT_METHOD) + case $v in SHA512 | YESCRYPT) ;; *) echo "ENCRYPT_METHOD=${v:-unset}" ;; esac +} +weak_umask() { + v=$(login_defs UMASK) + case $v in '' | *[!0-7]*) echo "UMASK=${v:-unset}"; return ;; esac + # 027 or stricter: group may not write, others get nothing + if [ $(((0$v >> 3) & 2)) -eq 0 ] || [ $((0$v & 7)) -ne 7 ]; then echo "UMASK=$v"; fi +} + +other_uid0() { awk -F: '$3 == "0" && $1 != "root" { print $1 }' /etc/passwd; } +other_gid0_accounts() { + awk -F: '$4 == "0" && $1 !~ /^(root|sync|shutdown|halt|operator)$/ { print $1 }' /etc/passwd +} +other_gid0_groups() { awk -F: '$3 == "0" && $1 != "root" { print $1 }' /etc/group; } +root_unlocked() { awk -F: '$1 == "root" && $2 !~ /^[!*]/ { print "root" }' /etc/shadow; } +system_login_shells() { + min=$(login_defs UID_MIN) + echo "$SHELLS" | awk -F: -v min="${min:-1000}" ' + NR == FNR { if ($0 != "") shell[$0] = 1; next } + $1 != "root" && ($3 + 0 < min + 0 || $3 == "65534") && ($7 in shell) { print $1 } + ' - /etc/passwd +} +unlocked_without_shell() { + echo "$SHELLS" | awk -F: ' + FILENAME == "-" { if ($0 != "") shell[$0] = 1; next } + FILENAME == "/etc/passwd" { if ($1 != "root" && !($7 in shell)) nologin[$1] = 1; next } + ($1 in nologin) && $2 !~ /^[!*]/ { print $1 } + ' - /etc/passwd /etc/shadow +} + +# too_open <file> <max mode> <owner> <group regex>: names the file when it is +# more open than the mode allows or owned by someone else; a missing file is fine. +too_open() { + [ -e "$1" ] || return 0 + # shellcheck disable=SC2046 + set -- "$1" "$2" "$3" "$4" $(stat -Lc '%a %U %G' "$1") + if [ $((0$5 & ~0$2 & 07777)) -ne 0 ] || [ "$6" != "$3" ] || ! echo "$7" | grep -qxE "$4"; then + echo "$1=$5:$6:$7" + fi +} +perm_passwd() { too_open /etc/passwd 644 root root; } +perm_passwd_() { too_open /etc/passwd- 644 root root; } +perm_group() { too_open /etc/group 644 root root; } +perm_group_() { too_open /etc/group- 644 root root; } +perm_shadow() { too_open /etc/shadow 640 root 'root|shadow'; } +perm_shadow_() { too_open /etc/shadow- 640 root 'root|shadow'; } +perm_gshadow() { too_open /etc/gshadow 640 root 'root|shadow'; } +perm_gshadow_() { too_open /etc/gshadow- 640 root 'root|shadow'; } +perm_shells() { too_open /etc/shells 644 root root; } +perm_opasswd() { too_open /etc/security/opasswd 600 root root; } + +world_writable() { + find / -xdev -type f -perm -0002 + find / -xdev -type d -perm -0002 ! -perm -1000 +} +unowned() { find / -xdev \( -nouser -o -nogroup \); } + +not_shadowed() { awk -F: '$2 != "x" { print $1 }' /etc/passwd; } +empty_passwords() { awk -F: '$2 == "" { print $1 }' /etc/shadow; } +missing_groups() { awk -F: 'NR == FNR { g[$3] = 1; next } !($4 in g) { print $1 ":" $4 }' /etc/group /etc/passwd; } +shadow_members() { + awk -F: ' + NR == FNR { if ($1 == "shadow") { gid = $3; if ($4 != "") print "shadow:" $4 } next } + gid != "" && $4 == gid { print $1 } + ' /etc/group /etc/passwd +} +dup_uids() { cut -d: -f3 /etc/passwd | sort | uniq -d; } +dup_gids() { cut -d: -f3 /etc/group | sort | uniq -d; } +dup_users() { cut -d: -f1 /etc/passwd | sort | uniq -d; } +dup_groups() { cut -d: -f1 /etc/group | sort | uniq -d; } + +check debian:5.4.1.1 "Password expiration is configured" pass_max_days +check debian:5.4.1.2 "Minimum password days are configured" pass_min_days +check debian:5.4.1.3 "Password expiration warning days are configured" pass_warn_age +check debian:5.4.1.4 "Strong password hashing algorithm is configured" weak_hashing +check debian:5.4.2.1 "root is the only UID 0 account" other_uid0 +check debian:5.4.2.2 "root is the only GID 0 account" other_gid0_accounts +check debian:5.4.2.3 "group root is the only GID 0 group" other_gid0_groups +check debian:5.4.2.4 "root account access is controlled (locked)" root_unlocked +check debian:5.4.2.7 "System accounts do not have a valid login shell" system_login_shells +check debian:5.4.2.8 "Accounts without a valid login shell are locked" unlocked_without_shell +check debian:5.4.3.3 "Default user umask is configured" weak_umask +check debian:7.1.1 "Permissions on /etc/passwd are configured" perm_passwd +check debian:7.1.2 "Permissions on /etc/passwd- are configured" perm_passwd_ +check debian:7.1.3 "Permissions on /etc/group are configured" perm_group +check debian:7.1.4 "Permissions on /etc/group- are configured" perm_group_ +check debian:7.1.5 "Permissions on /etc/shadow are configured" perm_shadow +check debian:7.1.6 "Permissions on /etc/shadow- are configured" perm_shadow_ +check debian:7.1.7 "Permissions on /etc/gshadow are configured" perm_gshadow +check debian:7.1.8 "Permissions on /etc/gshadow- are configured" perm_gshadow_ +check debian:7.1.9 "Permissions on /etc/shells are configured" perm_shells +check debian:7.1.10 "Permissions on /etc/security/opasswd are configured" perm_opasswd +check debian:7.1.11 "World writable files and directories are secured" world_writable +check debian:7.1.12 "No files or directories without an owner and a group exist" unowned +check debian:7.2.1 "Accounts in /etc/passwd use shadowed passwords" not_shadowed +check debian:7.2.2 "/etc/shadow password fields are not empty" empty_passwords +check debian:7.2.3 "All groups in /etc/passwd exist in /etc/group" missing_groups +check debian:7.2.4 "The shadow group is empty" shadow_members +check debian:7.2.5 "No duplicate UIDs exist" dup_uids +check debian:7.2.6 "No duplicate GIDs exist" dup_gids +check debian:7.2.7 "No duplicate user names exist" dup_users +check debian:7.2.8 "No duplicate group names exist" dup_groups diff --git a/image_tools/dockerfile.py b/image_tools/dockerfile.py new file mode 100644 index 0000000..ec30f96 --- /dev/null +++ b/image_tools/dockerfile.py @@ -0,0 +1,59 @@ +"""The images a Dockerfile builds on, each pinned by digest. + +bin/verify-base.sh checks their signatures before the build. A FROM naming a +stage of the same file is no image, ``scratch`` is none either, and an image +without ``@sha256:`` cannot be verified: a tag can move after the check. + + python -m image_tools.dockerfile <Dockerfile>... +""" + +from __future__ import annotations + +import argparse +import re +from pathlib import Path + +_FROM = re.compile(r"^\s*FROM\s+(.*)$", re.IGNORECASE) +_DIGEST = re.compile(r"@sha256:[0-9a-f]{64}$") + + +def _instructions(text: str) -> list[str]: + return re.sub(r"\\\s*\n", " ", text).splitlines() + + +def external_bases(text: str) -> list[str]: + """Every image the file's FROM lines name, once, in order.""" + stages: set[str] = set() + bases: list[str] = [] + for line in _instructions(text): + match = _FROM.match(line) + if not match: + continue + words = [w for w in match.group(1).split() if not w.startswith("--")] + ref = words[0] + if len(words) >= 3 and words[1].lower() == "as": + stages.add(words[2].lower()) + if ref.lower() in stages or ref == "scratch": + continue + if not _DIGEST.search(ref): + raise ValueError(f"FROM {ref} is not pinned by digest") + if ref not in bases: + bases.append(ref) + return bases + + +def main() -> int: # pragma: no cover - CLI seam + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("dockerfiles", nargs="+") + args = parser.parse_args() + seen: list[str] = [] + for path in args.dockerfiles: + for ref in external_bases(Path(path).read_text()): + if ref not in seen: + seen.append(ref) + print(ref) + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/image_tools/platforms.py b/image_tools/platforms.py new file mode 100644 index 0000000..72d7af0 --- /dev/null +++ b/image_tools/platforms.py @@ -0,0 +1,64 @@ +"""The platform images of a published reference, for bin/sign.sh. + +An index (base-os, a released netork image) lists them; a single manifest (a +netork dev image) is one itself, its architecture in its config. Each line out +is ``<arch> <digest>``. Only amd64 and arm64 are built, so anything else, or +nothing at all, is an error rather than a quiet skip. + + python -m image_tools.platforms <digest> <raw.json> <image.json> + +``raw.json`` is ``docker buildx imagetools inspect --raw``, ``image.json`` its +``--format '{{json .Image}}'``, and ``<digest>`` the reference's own. +""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +BUILT = ("amd64", "arm64") +INDEXES = ( + "application/vnd.oci.image.index.v1+json", + "application/vnd.docker.distribution.manifest.list.v2+json", +) + + +def _checked(arch: str) -> str: + if arch not in BUILT: + raise ValueError(f"{arch} is not a platform built here ({', '.join(BUILT)})") + return arch + + +def platforms(raw: str, image: str, digest: str) -> list[tuple[str, str]]: + """``(arch, digest)`` of every platform image behind the reference.""" + manifest = json.loads(raw) + if manifest.get("mediaType") in INDEXES or "manifests" in manifest: + found = [] + for entry in manifest.get("manifests") or []: + platform = entry.get("platform") or {} + if platform.get("os") != "linux": + continue # buildx's attestation manifests say "unknown" + found.append((_checked(platform.get("architecture", "")), entry["digest"])) + else: + arch = (json.loads(image) or {}).get("architecture") + found = [(_checked(arch), digest)] if arch else [] + if not found: + raise ValueError("the reference holds no platform image to sign") + return found + + +def main() -> int: # pragma: no cover - CLI seam + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("digest") + parser.add_argument("raw") + parser.add_argument("image") + args = parser.parse_args() + found = platforms(Path(args.raw).read_text(), Path(args.image).read_text(), args.digest) + for arch, digest in found: + print(arch, digest) + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/image_tools/sbom.py b/image_tools/sbom.py new file mode 100644 index 0000000..18a5503 --- /dev/null +++ b/image_tools/sbom.py @@ -0,0 +1,60 @@ +"""The CycloneDX SBOM of an image, as Trivy reads it. + +bin/sign.sh attaches it to each published platform image as an attestation, +next to the CIS result, so what base-os holds can be read from the registry. + + python -m image_tools.sbom <ref> <out.cdx.json> + +Standard library only; Trivy runs as the container image_tools/verify.py pins. +""" + +from __future__ import annotations + +import argparse +import json +import subprocess +from pathlib import Path + +from image_tools.verify import TIMEOUT, TRIVY_IMAGE + + +def cyclonedx(ref: str) -> dict: # pragma: no cover - container seam + proc = subprocess.run( + [ + "docker", + "run", + "--rm", + "-v", + "/var/run/docker.sock:/var/run/docker.sock", + "-e", + "TRIVY_USERNAME", + "-e", + "TRIVY_PASSWORD", + TRIVY_IMAGE, + "image", + "--quiet", + "--disable-telemetry", + "--skip-version-check", + "--format", + "cyclonedx", + ref, + ], + capture_output=True, + text=True, + timeout=TIMEOUT, + check=True, + ) + return json.loads(proc.stdout) + + +def main() -> int: # pragma: no cover - CLI seam + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("ref") + parser.add_argument("out") + args = parser.parse_args() + Path(args.out).write_text(json.dumps(cyclonedx(args.ref), indent=2)) + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/image_tools/verify.py b/image_tools/verify.py new file mode 100644 index 0000000..eca85c6 --- /dev/null +++ b/image_tools/verify.py @@ -0,0 +1,221 @@ +"""Does a candidate image improve on the pinned one? (#1308) + +Trivy lists both images' packages; netOrk's matcher in kb-server judges both in +one call, so against one state of its data (kb-server#155). The verdict: + +- **improved** when a critical or high CVE that has a fixed version is closed, + Debian made a point release (``/etc/debian_version`` differs), or the + candidate passes a CIS control the pin fails (``image_tools/cis.py``, both measured + in the same run); +- **same** otherwise, and whenever the matcher cannot be asked: no pin moves on + a guess. + +New critical or high CVEs in the candidate never block; the report lists them. + + python -m image_tools.verify <current-ref> <candidate-ref> [--packages packages.allow] \ + --json verdict.json --report report.md + +Standard library only; Trivy runs as a pinned container. +""" + +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import urllib.request +from collections.abc import Callable, Iterable +from dataclasses import asdict, dataclass, field +from pathlib import Path + +#: Trivy, pinned by its multi-platform index digest. +TRIVY_IMAGE = ( + "aquasec/trivy:0.75.0@sha256:af6acf9a6b85dfe389a1941505c0ce9efef52a4719635e1a962f022a3d855daa" +) +SEVERE = frozenset({"critical", "high"}) +KB_URL = os.environ.get("KB_URL", "https://kb.netork.io") +TIMEOUT = 300 + + +@dataclass +class Verdict: + improved: bool + closed: list[dict] = field(default_factory=list) + new: list[dict] = field(default_factory=list) + point_release: tuple[str, str] | None = None + cis_closed: list[str] = field(default_factory=list) + counts: dict = field(default_factory=dict) + error: str | None = None + + def to_json(self) -> str: + return json.dumps(asdict(self), indent=2) + + +def _patched(finding: dict) -> bool: + return (finding.get("severity") or "").lower() in SEVERE and bool(finding.get("fixed")) + + +def judge( + answer: dict, + current_debian: str, + candidate_debian: str, + cis_closed: Iterable[str] = (), +) -> Verdict: + """The verdict on the matcher's *answer*, both images' Debian versions and + the CIS controls the candidate passes and the pin fails.""" + closed = [f for f in answer.get("closed") or [] if _patched(f)] + new = [f for f in answer.get("new") or [] if (f.get("severity") or "").lower() in SEVERE] + point = None + if current_debian.strip() != candidate_debian.strip(): + point = (current_debian.strip(), candidate_debian.strip()) + counts = {"current": answer.get("current", {}), "candidate": answer.get("candidate", {})} + cis = sorted(cis_closed) + return Verdict(bool(closed) or point is not None or bool(cis), closed, new, point, cis, counts) + + +def verify( + current: str, + candidate: str, + *, + sbom: Callable[[str], dict] | None = None, + match: Callable[[dict, dict], dict] | None = None, + debian_version: Callable[[str], str] | None = None, + cis_failing: Callable[[str], set[str]] | None = None, +) -> Verdict: + """Judge *candidate* against *current*; never raises.""" + sbom = sbom or trivy_sbom + match = match or kb_match + debian_version = debian_version or debian_version_of + cis_failing = cis_failing or cis_failing_of + try: + answer = match(sbom(current), sbom(candidate)) + cis_closed = cis_failing(current) - cis_failing(candidate) + return judge(answer, debian_version(current), debian_version(candidate), cis_closed) + except (OSError, ValueError, subprocess.SubprocessError) as exc: + return Verdict(False, error=f"{type(exc).__name__}: {exc}") + + +def _row(f: dict) -> str: + fixed = f.get("fixed") or "–" + return f"| {f.get('cve')} | {f.get('package')} | {f.get('severity')} | {f.get('installed')} | {fixed} |" + + +def render(verdict: Verdict, current: str, candidate: str) -> str: + """The pull request body: what closed, what is new, against what.""" + head = "**Improved**" if verdict.improved else "**Same** (the pin stays)" + lines = [f"{head}: `{candidate}` against the pinned `{current}`.", ""] + if verdict.error: + lines += [f"The matcher could not be asked: {verdict.error}", ""] + if verdict.point_release: + old, new = verdict.point_release + lines += [f"Debian point release: {old} → {new}", ""] + if verdict.cis_closed: + lines += [ + f"CIS controls the pin fails and the candidate passes: {', '.join(verdict.cis_closed)}", + "", + ] + table = ["| CVE | Package | Severity | Installed | Fixed in |", "|---|---|---|---|---|"] + if verdict.closed: + lines += ["### Closed critical/high", "", *table, *map(_row, verdict.closed), ""] + if verdict.new: + lines += ["### New critical/high in the candidate", "", *table, *map(_row, verdict.new), ""] + if verdict.counts: + lines += [ + f"Counts: current {verdict.counts.get('current')}, candidate {verdict.counts.get('candidate')}." + ] + return "\n".join(lines).rstrip() + "\n" + + +def trivy_sbom(ref: str) -> dict: # pragma: no cover - container seam + """Trivy's package inventory of *ref* (no vulnerability DB).""" + proc = subprocess.run( + [ + "docker", + "run", + "--rm", + "-v", + "/var/run/docker.sock:/var/run/docker.sock", + "-e", + "TRIVY_USERNAME", + "-e", + "TRIVY_PASSWORD", + TRIVY_IMAGE, + "image", + "--quiet", + "--disable-telemetry", + "--skip-version-check", + "--format", + "json", + "--list-all-pkgs", + "--scanners", + "license", + ref, + ], + capture_output=True, + text=True, + timeout=TIMEOUT, + check=True, + ) + return json.loads(proc.stdout) + + +def kb_match(current: dict, candidate: dict) -> dict: # pragma: no cover - network seam + """kb-server's verdict on both inventories, in one call (kb-server#155).""" + body = json.dumps({"current": current, "candidate": candidate}).encode() + request = urllib.request.Request( + f"{KB_URL}/api/v1/match", + data=body, + method="POST", + headers={ + "Content-Type": "application/json", + "Authorization": f"Bearer {os.environ['NETORK_LICENSE_KEY']}", + }, + ) + with urllib.request.urlopen(request, timeout=TIMEOUT) as response: # noqa: S310 - https + return json.load(response) + + +def debian_version_of(ref: str) -> str: # pragma: no cover - container seam + proc = subprocess.run( + ["docker", "run", "--rm", "--entrypoint", "cat", ref, "/etc/debian_version"], + capture_output=True, + text=True, + timeout=120, + check=True, + ) + return proc.stdout.strip() + + +def cis_failing_of(ref: str, packages: str = "packages.allow") -> set[str]: # pragma: no cover + """The CIS controls *ref* fails, measured as image_tools/cis.py measures a build; + *packages* is the image repository's allow list.""" + from image_tools import cis + + allowed = cis.load_packages(Path(packages).read_text()) + return {c.id for c in cis.measure(ref, allowed) if c.result == cis.FAIL} + + +def main() -> int: # pragma: no cover - CLI seam + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("current") + parser.add_argument("candidate") + parser.add_argument("--json", required=True) + parser.add_argument("--report", required=True) + parser.add_argument("--packages", default="packages.allow") + args = parser.parse_args() + verdict = verify( + args.current, + args.candidate, + cis_failing=lambda ref: cis_failing_of(ref, args.packages), + ) + with open(args.json, "w") as fh: + fh.write(verdict.to_json()) + with open(args.report, "w") as fh: + fh.write(render(verdict, args.current, args.candidate)) + print("improved" if verdict.improved else "same") + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..512e30c --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,23 @@ +[project] +name = "image-tools" +version = "0.1.0" +description = "Judge, sign and verify netOrk container images" +requires-python = ">=3.11" +license = "MIT" + +[build-system] +requires = ["setuptools>=69"] +build-backend = "setuptools.build_meta" + +[tool.setuptools] +packages = ["image_tools"] + +[tool.setuptools.package-data] +image_tools = ["cis_os.sh"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +pythonpath = ["."] + +[tool.ruff] +line-length = 100 diff --git a/tests/test_actions.py b/tests/test_actions.py new file mode 100644 index 0000000..5b630ba --- /dev/null +++ b/tests/test_actions.py @@ -0,0 +1,38 @@ +"""The composite actions every image repository of NetOrk uses (NetOrk/netork#1390). + + uses: https://git.netork.io/NetOrk/image-tools/actions/<name>@<commit> + +They only call what this repository holds, so a pinned commit fixes what runs. +""" + +from __future__ import annotations + +import re +from pathlib import Path + +import pytest +import yaml + +ROOT = Path(__file__).resolve().parents[1] +ACTIONS = sorted((ROOT / "actions").glob("*/action.yml")) + + +def test_the_three_actions(): + assert [a.parent.name for a in ACTIONS] == ["judge", "sign", "verify-base"] + + +@pytest.mark.parametrize("action", ACTIONS, ids=lambda a: a.parent.name) +def test_an_action_runs_only_what_this_repository_holds(action): + spec = yaml.safe_load(action.read_text()) + + assert spec["runs"]["using"] == "composite" + for step in spec["runs"]["steps"]: + for script in re.findall(r"github\.action_path }}/\.\./\.\./(bin/[\w.-]+)", step["run"]): + assert (ROOT / script).is_file(), script + assert "${{ inputs." not in step["run"], "inputs reach a script through env" + + +def test_signing_needs_its_secrets_from_the_caller(): + text = (ROOT / "actions/sign/action.yml").read_text() + + assert "COSIGN_PRIVATE_KEY_B64" in text and "secrets." not in text diff --git a/tests/test_cis.py b/tests/test_cis.py new file mode 100644 index 0000000..45d3079 --- /dev/null +++ b/tests/test_cis.py @@ -0,0 +1,251 @@ +"""Whether an image is what CIS asks of it (CIS Docker 1.6.0 §4, CIS Debian Linux 12). + +Trivy measures what it automates, ``image_tools/cis_os.sh`` what can be seen from +inside the image, and ``cis.toml`` decides the rest. The gate is red for a +failure nobody decided, an expired exception, a control neither measured nor +decided, a decision for a control nothing knows, and a measurement not taken. +""" + +from __future__ import annotations + +import datetime +import json +import subprocess + +import pytest + +from image_tools.cis import ( + Control, + Decision, + check, + from_os, + from_trivy, + judge, + load_decisions, + load_packages, + merge, + render, +) + +TODAY = datetime.date(2026, 10, 21) +SUMMARY = { + "ID": "docker-cis-1.6.0", + "SummaryControls": [ + {"ID": "4.1", "Name": "Ensure a user for the container has been created", "TotalFail": 1}, + {"ID": "4.3", "Name": "Ensure unnecessary packages are not installed (Manual)"}, + {"ID": "4.7", "Name": "Ensure update instructions are not used alone", "TotalFail": 0}, + {"ID": "4.8", "Name": "Ensure setuid and setgid permissions are removed (Manual)"}, + ], +} +OS = ( + "docker:4.3\tPASS\tUnnecessary packages are not installed (packages.allow)\t\n" + "docker:4.8\tFAIL\tsetuid and setgid permissions are removed\t/usr/bin/su /usr/bin/mount\n" + "debian:7.2.5\tPASS\tNo duplicate UIDs exist\t\n" +) + + +def _c(cid, result, detail=""): + return Control(cid, cid, result, detail) + + +class TestMeasurements: + def test_trivy_without_a_count_did_not_measure(self): + controls = {c.id: c for c in from_trivy(SUMMARY)} + + assert controls["docker:4.1"].result == "fail" + assert controls["docker:4.3"].result == "unmeasured" + assert controls["docker:4.7"].result == "pass" + + def test_the_os_checks_say_what_fails(self): + controls = {c.id: c for c in from_os(OS)} + + assert controls["docker:4.8"].result == "fail" + assert controls["docker:4.8"].detail == "/usr/bin/su /usr/bin/mount" + assert controls["debian:7.2.5"].result == "pass" + + def test_a_line_the_os_checks_should_not_print_is_an_error(self): + with pytest.raises(ValueError): + from_os("sh: 1: stat: not found\n") + + def test_the_os_checks_settle_what_trivy_leaves_to_manual(self): + controls = {c.id: c for c in merge(from_trivy(SUMMARY), from_os(OS))} + + assert controls["docker:4.3"].result == "pass" + assert controls["docker:4.8"].result == "fail" + assert controls["docker:4.1"].result == "fail" + assert list(controls)[:4] == ["docker:4.1", "docker:4.3", "docker:4.7", "docker:4.8"] + + def test_a_failure_wins_over_a_pass(self): + (control,) = merge([_c("docker:4.8", "fail")], [_c("docker:4.8", "pass")]) + + assert control.result == "fail" + + +class TestJudge: + def test_all_passing_is_green(self): + assert judge([_c("docker:4.1", "pass")], {}, TODAY).ok + + def test_a_failure_nobody_decided_is_red(self): + report = judge([_c("docker:4.1", "fail")], {}, TODAY) + + assert not report.ok + assert report.controls[0].state == "fail" + + def test_an_exception_keeps_a_failure_green(self): + decisions = {"docker:4.6": Decision("exception", "no service")} + + report = judge([_c("docker:4.6", "fail")], decisions, TODAY) + + assert report.ok + assert report.controls[0].state == "excepted" + + def test_an_expired_exception_is_red(self): + decisions = {"docker:4.6": Decision("exception", "for now", datetime.date(2026, 10, 1))} + + report = judge([_c("docker:4.6", "fail")], decisions, TODAY) + + assert not report.ok + assert report.controls[0].state == "expired" + + def test_an_exception_expiring_today_still_holds(self): + decisions = {"docker:4.6": Decision("exception", "for now", TODAY)} + + assert judge([_c("docker:4.6", "fail")], decisions, TODAY).ok + + def test_a_control_nothing_measures_must_be_decided(self): + assert not judge([_c("docker:4.2", "unmeasured")], {}, TODAY).ok + + decisions = {"docker:4.2": Decision("manual", "pinned by digest")} + report = judge([_c("docker:4.2", "unmeasured")], decisions, TODAY) + assert report.ok + assert report.controls[0].state == "manual" + + def test_manual_is_not_the_tools_word(self): + """4.4 belongs to the matcher, whose policy is that new CVEs never block.""" + decisions = {"docker:4.4": Decision("manual", "judged by the matcher")} + + assert judge([_c("docker:4.4", "fail")], decisions, TODAY).ok + + def test_an_exception_for_a_passing_control_is_named_stale(self): + decisions = {"docker:4.6": Decision("exception", "no service")} + + report = judge([_c("docker:4.6", "pass")], decisions, TODAY) + + assert report.ok + assert report.stale == ["docker:4.6"] + + def test_a_decision_for_a_control_nothing_knows_is_red(self): + """A typo in cis.toml must not silently decide nothing.""" + decisions = {"docker:4.66": Decision("exception", "typo")} + + report = judge([_c("docker:4.6", "pass")], decisions, TODAY) + + assert not report.ok + assert report.unknown == ["docker:4.66"] + + def test_nothing_measured_is_not_green(self): + assert not judge([], {}, TODAY).ok + + +class TestDecisions: + def test_both_kinds_with_their_reasons(self): + decisions = load_decisions( + """ +[[manual]] +id = "docker:4.2" +reason = "pinned by digest" + +[[exception]] +id = "docker:4.6" +reason = "no service" +expires = 2027-01-31 +""" + ) + + assert decisions["docker:4.2"] == Decision("manual", "pinned by digest") + assert decisions["docker:4.6"].expires == datetime.date(2027, 1, 31) + + @pytest.mark.parametrize( + "text", + [ + '[[manual]]\nid = "docker:4.2"\nreason = " "\n', + '[[manual]]\nid = "docker:4.2"\nreason = "a"\n[[exception]]\nid = "docker:4.2"\nreason = "b"\n', + '[[exception]]\nid = "docker:4.6"\nreason = "a"\nexpires = "soon"\n', + ], + ids=["no reason", "decided twice", "expires not a date"], + ) + def test_a_decision_without_substance_is_refused(self, text): + with pytest.raises(ValueError): + load_decisions(text) + + +class TestCheck: + def test_measured_and_judged(self): + seams = {"trivy": lambda ref: SUMMARY, "os_checks": lambda ref, pkgs: OS} + decisions = {"docker:4.8": Decision("exception", "not yet")} + + report = check("img", decisions, ["apt"], TODAY, **seams) + + assert not report.ok # 4.1 fails undecided + assert {c.id: c.state for c in report.controls}["docker:4.8"] == "excepted" + + def test_the_packages_reach_the_os_checks(self): + seen = [] + + def os_checks(ref, packages): + seen.append(packages) + return OS + + check("img", {}, ["apt", "bash"], TODAY, trivy=lambda r: SUMMARY, os_checks=os_checks) + + assert seen == [["apt", "bash"]] + + def test_a_measurement_not_taken_is_red(self): + def trivy(ref): + raise subprocess.CalledProcessError(1, "trivy") + + report = check("img", {}, [], TODAY, trivy=trivy, os_checks=lambda r, p: OS) + + assert not report.ok + assert "CalledProcessError" in report.error + + +def test_the_report_says_why(): + decisions = { + "docker:4.6": Decision("exception", "base-os runs no service"), + "docker:4.2": Decision("manual", "pinned by digest"), + } + controls = [ + _c("docker:4.1", "fail", "1 failed (Trivy)"), + _c("docker:4.2", "unmeasured"), + _c("docker:4.6", "fail", "1 failed (Trivy)"), + ] + + text = render(judge(controls, decisions, TODAY), "base-os:ci") + + assert "**CIS: red**" in text and "`base-os:ci`" in text + assert "| docker:4.1 | ❌ fail |" in text and "1 failed (Trivy)" in text + assert "base-os runs no service" in text + assert "pinned by digest" in text + + +def test_the_report_is_json_for_the_attestation(): + report = judge([_c("docker:4.1", "pass")], {}, TODAY) + + data = json.loads(report.to_json()) + + assert data["ok"] is True + assert data["controls"][0] == { + "id": "docker:4.1", + "title": "docker:4.1", + "result": "pass", + "detail": "", + "state": "pass", + "reason": "", + } + + +def test_the_allow_list_is_names_without_comments_or_blank_lines(): + text = "# what base-os may hold\napt\n\n bash \n# login.defs comes from login\nlogin\n" + + assert load_packages(text) == ["apt", "bash", "login"] diff --git a/tests/test_dockerfile.py b/tests/test_dockerfile.py new file mode 100644 index 0000000..b46e7bc --- /dev/null +++ b/tests/test_dockerfile.py @@ -0,0 +1,60 @@ +"""The images a Dockerfile builds on, so their signatures can be checked first. + +Every link of the chain (base-os → base-python → engine; base-os → ui) checks +the signature of what it pins before it builds. That starts with the external +references of its FROM lines: a stage of the same file is no image, and an +image not pinned by digest cannot be verified at all. +""" + +from __future__ import annotations + +import pytest + +from image_tools.dockerfile import external_bases + +BASE = "registry.netork.io/netork/base-python:3.12.15-13-20261020.1@sha256:" + "c" * 64 +OS = "registry.netork.io/netork/base-os:13-20261022.2@sha256:" + "f" * 64 + + +def test_stages_of_the_same_file_are_not_images(): + text = f""" +FROM {BASE} AS builder +RUN true +FROM builder AS drivers +FROM {BASE} AS runtime +COPY --from=drivers /opt /opt +""" + + assert external_bases(text) == [BASE] + + +def test_every_distinct_image_once_in_order(): + text = f"FROM {OS} AS builder\nFROM {BASE} AS runtime\nFROM {OS}\n" + + assert external_bases(text) == [OS, BASE] + + +def test_flags_and_case_do_not_hide_an_image(): + text = f"from --platform=$BUILDPLATFORM {OS} as build\n" + + assert external_bases(text) == [OS] + + +def test_scratch_is_no_image(): + assert external_bases("FROM scratch\n") == [] + + +def test_continuation_lines_are_joined(): + text = f"FROM \\\n {OS} \\\n AS runtime\n" + + assert external_bases(text) == [OS] + + +@pytest.mark.parametrize( + "ref", + ["debian:trixie-slim", "registry.netork.io/netork/base-os:pinned", "${BASE_IMAGE}"], +) +def test_an_image_without_a_digest_is_refused(ref): + """A tag can move; only a digest says which image was checked.""" + with pytest.raises(ValueError, match="digest"): + external_bases(f"FROM {ref}\n") diff --git a/tests/test_platforms.py b/tests/test_platforms.py new file mode 100644 index 0000000..811344b --- /dev/null +++ b/tests/test_platforms.py @@ -0,0 +1,82 @@ +"""Which platform images a published reference has, for bin/sign.sh. + +base-os publishes an index of amd64 and arm64; netork's dev images are a single +amd64 manifest, and a release joins arm64 to it under a new index. Every +platform image gets its own CIS result and SBOM, so sign.sh needs each one's +architecture and digest, whatever shape the reference has. +""" + +from __future__ import annotations + +import json + +import pytest + +from image_tools.platforms import platforms + +AMD = "sha256:" + "a" * 64 +ARM = "sha256:" + "b" * 64 +ATT = "sha256:" + "c" * 64 +SELF = "sha256:" + "d" * 64 + + +def _index(*manifests: dict) -> str: + return json.dumps( + {"mediaType": "application/vnd.oci.image.index.v1+json", "manifests": list(manifests)} + ) + + +def _entry(digest: str, arch: str | None, os: str = "linux") -> dict: + entry: dict = {"digest": digest, "mediaType": "application/vnd.oci.image.manifest.v1+json"} + if arch: + entry["platform"] = {"os": os, "architecture": arch} + return entry + + +def test_an_index_names_each_linux_platform(): + raw = _index(_entry(AMD, "amd64"), _entry(ARM, "arm64")) + + assert platforms(raw, "{}", SELF) == [("amd64", AMD), ("arm64", ARM)] + + +def test_buildx_attestation_manifests_are_no_platform(): + raw = _index(_entry(AMD, "amd64"), _entry(ATT, "unknown", os="unknown"), _entry(ATT, None)) + + assert platforms(raw, "{}", SELF) == [("amd64", AMD)] + + +def test_a_single_manifest_is_its_own_platform(): + raw = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}) + image = json.dumps({"architecture": "amd64", "os": "linux"}) + + assert platforms(raw, image, SELF) == [("amd64", SELF)] + + +def test_a_docker_v2_manifest_list_counts_as_an_index(): + raw = json.dumps( + { + "mediaType": "application/vnd.docker.distribution.manifest.list.v2+json", + "manifests": [_entry(ARM, "arm64")], + } + ) + + assert platforms(raw, "{}", SELF) == [("arm64", ARM)] + + +@pytest.mark.parametrize( + "raw", + [ + _index(_entry(ATT, "unknown", os="unknown")), + json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json"}), + ], + ids=["index without an image", "manifest without an architecture"], +) +def test_nothing_to_sign_is_an_error(raw): + with pytest.raises(ValueError): + platforms(raw, "{}", SELF) + + +def test_an_architecture_not_built_here_is_an_error(): + """Only amd64 and arm64 are built (#934); anything else would be signed blind.""" + with pytest.raises(ValueError, match="riscv64"): + platforms(_index(_entry(AMD, "riscv64")), "{}", SELF) diff --git a/tests/test_signing.py b/tests/test_signing.py new file mode 100644 index 0000000..c08680e --- /dev/null +++ b/tests/test_signing.py @@ -0,0 +1,228 @@ +"""Signing and checking netOrk's images with cosign (CIS Docker 4.5, NetOrk/netork#1390). + +Docker Content Trust is deprecated; like Docker's Hardened Images, every image +of NetOrk is signed with cosign instead and carries its CIS result and SBOM as +attestations. The registry's names stay inside: nothing goes to a public +transparency log. Each link of the chain checks the signature of the image it +builds on before it builds. + +The scripts run here against a fake ``cosign`` and ``docker`` that record what +they are asked. +""" + +from __future__ import annotations + +import json +import os +import re +import subprocess +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +SIGN = ROOT / "bin/sign.sh" +VERIFY = ROOT / "bin/verify-signature.sh" +VERIFY_BASE = ROOT / "bin/verify-base.sh" +INSTALL = ROOT / "bin/install-cosign.sh" + +IMAGE = "registry.netork.io/netork/engine" +AMD = "sha256:" + "a" * 64 +ARM = "sha256:" + "b" * 64 +INDEX = "sha256:" + "e" * 64 + +_FAKE_COSIGN = """#!/bin/sh +echo "cosign $*" >> "$LOG" +case "$*" in *"{fail}"*) echo "no matching signatures" >&2; exit 1 ;; esac +exit 0 +""" +_FAKE_DOCKER = """#!/bin/sh +case "$*" in + *--raw*) cat "$RAW" ;; + *"{{json .Image}}"*) cat "$CONFIG" ;; + *) echo "docker $*" >> "$LOG" ;; +esac +""" + + +def _bin(tmp_path: Path, fail: str = "@@never@@") -> dict[str, str]: + bin_dir = tmp_path / "bin" + bin_dir.mkdir() + (bin_dir / "cosign").write_text(_FAKE_COSIGN.replace("{fail}", fail)) + (bin_dir / "docker").write_text(_FAKE_DOCKER) + for tool in bin_dir.iterdir(): + tool.chmod(0o755) + return {"PATH": f"{bin_dir}:{os.environ['PATH']}", "LOG": str(tmp_path / "log")} + + +def _log(tmp_path: Path) -> list[str]: + log = tmp_path / "log" + return log.read_text().splitlines() if log.exists() else [] + + +def _run(script: Path, *args: str, env: dict[str, str]) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ["bash", str(script), *args], env=env, capture_output=True, text=True, timeout=30 + ) + + +class TestSign: + def _attest_dir(self, tmp_path: Path, *arches: str) -> Path: + out = tmp_path / "attest" + out.mkdir() + for arch in arches: + (out / f"cis-{arch}.json").write_text("{}") + (out / f"sbom-{arch}.cdx.json").write_text("{}") + return out + + def _env(self, tmp_path: Path, raw: dict, config: dict, **extra: str) -> dict[str, str]: + (tmp_path / "raw.json").write_text(json.dumps(raw)) + (tmp_path / "config.json").write_text(json.dumps(config)) + env = _bin(tmp_path) + env.update( + RAW=str(tmp_path / "raw.json"), + CONFIG=str(tmp_path / "config.json"), + COSIGN_PRIVATE_KEY_B64="a2V5Cg==", + COSIGN_PASSWORD="pw", + **extra, + ) + return env + + def test_a_single_manifest_is_signed_and_attested(self, tmp_path): + """A netork dev image: one amd64 manifest, no index.""" + env = self._env(tmp_path, {"mediaType": "x", "layers": []}, {"architecture": "amd64"}) + attest = self._attest_dir(tmp_path, "amd64") + + result = _run(SIGN, f"{IMAGE}:main-abc@{AMD}", str(attest), env=env) + + assert result.returncode == 0, result.stderr + log = _log(tmp_path) + assert any("sign --yes --recursive" in ln and f"@{AMD}" in ln for ln in log) + attests = [ln for ln in log if " attest " in ln] + assert len(attests) == 2 and all(f"{IMAGE}@{AMD}" in ln for ln in attests) + assert f"signed and attested: {IMAGE}@{AMD} (amd64)" in result.stdout + + def test_an_index_attests_every_platform(self, tmp_path): + raw = { + "mediaType": "application/vnd.oci.image.index.v1+json", + "manifests": [ + {"digest": AMD, "platform": {"os": "linux", "architecture": "amd64"}}, + {"digest": ARM, "platform": {"os": "linux", "architecture": "arm64"}}, + ], + } + env = self._env(tmp_path, raw, {}, EXPECT_PLATFORMS="amd64 arm64") + attest = self._attest_dir(tmp_path, "amd64", "arm64") + + result = _run(SIGN, f"{IMAGE}:0.41.0@{INDEX}", str(attest), env=env) + + assert result.returncode == 0, result.stderr + verified = [ln for ln in _log(tmp_path) if "verify-attestation" in ln] + assert sum(f"@{AMD}" in ln for ln in verified) == 2 + assert sum(f"@{ARM}" in ln for ln in verified) == 2 + + def test_a_missing_expected_platform_stops_it(self, tmp_path): + env = self._env( + tmp_path, {"mediaType": "x"}, {"architecture": "amd64"}, EXPECT_PLATFORMS="amd64 arm64" + ) + + result = _run(SIGN, f"{IMAGE}@{AMD}", str(self._attest_dir(tmp_path, "amd64")), env=env) + + assert result.returncode != 0 + assert "expected amd64 arm64" in result.stderr + assert not any(" attest " in ln for ln in _log(tmp_path)) + + def test_a_reference_without_a_digest_is_refused(self, tmp_path): + env = self._env(tmp_path, {}, {}) + + result = _run(SIGN, f"{IMAGE}:latest", str(self._attest_dir(tmp_path)), env=env) + + assert result.returncode != 0 and "not pinned by digest" in result.stderr + + def test_without_the_secrets_it_says_so(self, tmp_path): + env = _bin(tmp_path) + + result = _run(SIGN, f"{IMAGE}@{AMD}", str(tmp_path), env=env) + + assert result.returncode != 0 and "secret is missing" in result.stderr + + +class TestVerify: + BASE = f"registry.netork.io/netork/base-python:3.12@{AMD}" + OS = f"registry.netork.io/netork/base-os:13@{ARM}" + + def _dockerfile(self, tmp_path: Path) -> Path: + path = tmp_path / "Dockerfile" + path.write_text( + f"FROM {self.BASE} AS builder\nFROM builder AS drivers\nFROM {self.OS} AS runtime\n" + ) + return path + + def test_every_base_of_the_dockerfile_is_verified_against_the_public_key(self, tmp_path): + result = _run(VERIFY_BASE, str(self._dockerfile(tmp_path)), env=_bin(tmp_path)) + + assert result.returncode == 0, result.stderr + log = _log(tmp_path) + assert log == [ + f"cosign verify --key cosign.pub --insecure-ignore-tlog {self.BASE}", + f"cosign verify --key cosign.pub --insecure-ignore-tlog {self.OS}", + ] + + def test_an_unsigned_base_stops_the_build(self, tmp_path): + env = _bin(tmp_path, fail="base-os") + + result = _run(VERIFY_BASE, str(self._dockerfile(tmp_path)), env=env) + + assert result.returncode != 0 + assert f"not signed with netOrk's key: {self.OS}" in result.stderr + + def test_a_base_without_a_digest_stops_the_build(self, tmp_path): + path = tmp_path / "Dockerfile" + path.write_text("FROM registry.netork.io/netork/base-os:pinned\n") + + result = _run(VERIFY_BASE, str(path), env=_bin(tmp_path)) + + assert result.returncode != 0 and "not pinned by digest" in result.stderr + assert _log(tmp_path) == [] + + def test_a_reference_without_a_digest_is_never_checked(self, tmp_path): + result = _run(VERIFY, "registry.netork.io/netork/base-os:pinned", env=_bin(tmp_path)) + + assert result.returncode != 0 and _log(tmp_path) == [] + + +def test_nothing_goes_to_a_public_transparency_log(): + config = json.loads((ROOT / "cosign-signing.json").read_text()) + text = SIGN.read_text() + + for services in ("rekorTlogUrls", "caUrls", "oidcUrls", "tsaUrls"): + assert config[services] == [] + signing = [ + ln + for ln in text.replace("\\\n", " ").splitlines() + if re.search(r"cosign (sign|attest) ", ln) + ] + assert len(signing) == 2 + assert all("--signing-config cosign-signing.json" in ln for ln in signing) + + +def test_the_public_key_is_in_the_repository(): + """Whoever pulls an image of NetOrk verifies against it; the private half is + the COSIGN_PRIVATE_KEY_B64 secret.""" + assert (ROOT / "cosign.pub").read_text().startswith("-----BEGIN PUBLIC KEY-----") + + +def test_the_private_key_arrives_as_one_line(): + """The runner masks a secret only where it stands whole; the PEM key's lines + did not match, and the key stood in a log (2026-10-22).""" + text = SIGN.read_text() + + assert '"${COSIGN_PRIVATE_KEY_B64:?' in text and "| base64 -d)" in text + + +@pytest.mark.parametrize("arch", ["AMD64", "ARM64"]) +def test_cosign_is_pinned_and_checked_for_both_runners(arch): + text = INSTALL.read_text() + + assert re.search(r"^VERSION=v\d+\.\d+\.\d+$", text, re.MULTILINE) + assert re.search(rf"^SHA256_{arch}=[0-9a-f]{{64}}$", text, re.MULTILINE) + assert "sha256sum -c -" in text diff --git a/tests/test_verify.py b/tests/test_verify.py new file mode 100644 index 0000000..66bf050 --- /dev/null +++ b/tests/test_verify.py @@ -0,0 +1,179 @@ +"""Whether a candidate image improves on the pinned one (#1308). + +Trivy lists the packages of both, netOrk's matcher in kb-server judges both in +one call (so against one state of its data, kb-server#155), and the verdict is +"improved" when a critical or high CVE with a fix is closed by a patch, or +Debian made a point release. Everything else is "same": the pin stays. + +A matcher that cannot be asked is never an improvement: no pin moves on a guess. +""" + +from __future__ import annotations + +import json +import subprocess + +import pytest + +from image_tools.verify import Verdict, judge, render, verify + +CLOSED = { + "cve": "CVE-2026-1111", + "package": "openssl", + "severity": "high", + "installed": "3.5.1-1", + "fixed": "3.5.1-1+deb13u1", +} +NEW = {"cve": "CVE-2026-2222", "package": "nginx", "severity": "critical", "installed": "1.26.3-4"} + + +def _answer(closed=(), new=()): + return { + "closed": list(closed), + "new": list(new), + "current": {"critical": 1, "high": 4}, + "candidate": {"critical": 1, "high": 3}, + } + + +class TestJudge: + def test_a_closed_cve_is_an_improvement(self): + verdict = judge(_answer(closed=[CLOSED]), "13.7", "13.7") + + assert verdict.improved + assert verdict.closed == [CLOSED] + assert not verdict.point_release + + def test_nothing_closed_is_the_same(self): + assert not judge(_answer(), "13.7", "13.7").improved + + def test_a_point_release_is_an_improvement_on_its_own(self): + verdict = judge(_answer(), "13.7", "13.8") + + assert verdict.improved + assert verdict.point_release == ("13.7", "13.8") + + def test_new_cves_do_not_block_they_are_reported(self): + verdict = judge(_answer(closed=[CLOSED], new=[NEW]), "13.7", "13.7") + + assert verdict.improved + assert verdict.new == [NEW] + + def test_new_cves_alone_are_no_improvement(self): + assert not judge(_answer(new=[NEW]), "13.7", "13.7").improved + + @pytest.mark.parametrize("severity", ["medium", "low", "unknown"]) + def test_only_critical_and_high_count(self, severity): + """kb-server already filters; a weaker one slipping through moves nothing.""" + closed = dict(CLOSED, severity=severity) + + assert not judge(_answer(closed=[closed]), "13.7", "13.7").improved + + def test_a_closed_cve_without_a_fix_is_not_a_patch(self): + """Gone without a fixed version means the data changed, not the image.""" + closed = dict(CLOSED, fixed="") + + assert not judge(_answer(closed=[closed]), "13.7", "13.7").improved + + +class TestVerify: + def _deps(self, answer=None, error=None, versions=("13.7", "13.7"), failing=((), ())): + calls = {"sbom": [], "match": []} + + def sbom(ref): + calls["sbom"].append(ref) + return {"ArtifactName": ref} + + def match(current, candidate): + calls["match"].append((current, candidate)) + if error: + raise error + return answer + + debian = dict(zip(("cur", "cand"), versions, strict=True)) + cis = dict(zip(("cur", "cand"), map(set, failing), strict=True)) + return calls, { + "sbom": sbom, + "match": match, + "debian_version": lambda ref: debian["cur" if ref == "cur" else "cand"], + "cis_failing": lambda ref: cis["cur" if ref == "cur" else "cand"], + } + + def test_both_go_to_the_matcher_in_one_call(self): + calls, seams = self._deps(answer=_answer(closed=[CLOSED])) + + verdict = verify("cur", "cand", **seams) + + assert calls["sbom"] == ["cur", "cand"] + assert calls["match"] == [({"ArtifactName": "cur"}, {"ArtifactName": "cand"})] + assert verdict.improved + + def test_a_matcher_that_cannot_be_asked_moves_nothing(self): + _, seams = self._deps(error=OSError("connection refused")) + + verdict = verify("cur", "cand", **seams) + + assert not verdict.improved + assert "connection refused" in verdict.error + + def test_even_a_point_release_waits_for_the_matcher(self): + _, seams = self._deps(error=OSError("down"), versions=("13.7", "13.8")) + + assert not verify("cur", "cand", **seams).improved + + def test_a_cis_control_the_pin_fails_and_the_candidate_passes_is_an_improvement(self): + _, seams = self._deps( + answer=_answer(), failing=({"docker:4.1", "docker:4.8"}, {"docker:4.8"}) + ) + + verdict = verify("cur", "cand", **seams) + + assert verdict.improved + assert verdict.cis_closed == ["docker:4.1"] + + def test_a_cis_control_the_candidate_newly_fails_is_no_improvement(self): + """The gate in check_image.sh stops such a build before; here it closes nothing.""" + _, seams = self._deps(answer=_answer(), failing=((), {"docker:4.8"})) + + assert not verify("cur", "cand", **seams).improved + + def test_a_cis_measurement_that_cannot_be_taken_moves_nothing(self): + _, seams = self._deps(answer=_answer(closed=[CLOSED])) + + def broken(ref): + raise subprocess.CalledProcessError(1, "docker") + + seams["cis_failing"] = broken + assert not verify("cur", "cand", **seams).improved + + +def test_the_report_names_what_closed_and_what_is_new(): + verdict = judge(_answer(closed=[CLOSED], new=[NEW]), "13.7", "13.8", ["docker:4.1"]) + text = render(verdict, "cur", "cand") + + assert "CVE-2026-1111" in text and "openssl" in text and "3.5.1-1+deb13u1" in text + assert "CVE-2026-2222" in text and "nginx" in text + assert "13.7 → 13.8" in text + assert "docker:4.1" in text + assert "improved" in text.lower() + + +def test_the_verdict_is_json_for_the_workflow(): + verdict = judge(_answer(closed=[CLOSED]), "13.7", "13.7") + + data = json.loads(verdict.to_json()) + + assert data["improved"] is True + assert data["closed"][0]["cve"] == "CVE-2026-1111" + + +def test_a_verdict_has_no_surprise_fields(): + assert set(Verdict.__dataclass_fields__) == { + "improved", + "closed", + "new", + "point_release", + "cis_closed", + "counts", + "error", + }