feat: judge, sign and verify netOrk's images in one place (NetOrk/netork#1390)
CI / Tests, lint (push) Successful in 18s

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.
This commit is contained in:
Christian Manivong committed 2026-10-22 16:23:00 +02:00
commit d139d66c3d
27 files changed
+2085

No files matched your search

+38
View File
@@ -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
+251
View File
@@ -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"]
+60
View File
@@ -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")
+82
View File
@@ -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)
+228
View File
@@ -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
+179
View File
@@ -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",
}