diff --git a/conformance/tests/l1-rendering/vuln-scanner-acceptances/chainsaw-test.yaml b/conformance/tests/l1-rendering/vuln-scanner-acceptances/chainsaw-test.yaml new file mode 100644 index 0000000..4495743 --- /dev/null +++ b/conformance/tests/l1-rendering/vuln-scanner-acceptances/chainsaw-test.yaml @@ -0,0 +1,25 @@ +# L1 fixture test: runs check.py (python3 stdlib, source imported from platform/vuln-scanner/src, +# no cluster, no network beyond loopback). conformance/run.sh --layer l1 runs check.sh offline; +# this Chainsaw Test wraps the same script so an in-cluster run executes identical assertions. +apiVersion: chainsaw.kyverno.io/v1alpha1 +kind: Test +metadata: + name: vuln-scanner-acceptances + labels: + conformance.home.k3s/R-VULN-08: "true" + annotations: + conformance.home.k3s/requirements: "R-VULN-08" + conformance.home.k3s/layer: "L1" + conformance.home.k3s/safety-critical: "false" + # Whole-repo scope: the touched-path map keys nothing to platform/vuln-scanner, so without this + # a change to the scanner would select no vuln-scanner test. + conformance.home.k3s/scope: "whole-repo" + conformance.home.k3s/intent: >- + Given acceptance files, When parsed and applied, Then expiry reverts to open and malformed entries never hide a finding. +spec: + steps: + - name: run-offline-check + description: Run check.sh, the same assertions the offline harness runs. + try: + - script: + content: bash ./check.sh diff --git a/conformance/tests/l1-rendering/vuln-scanner-acceptances/check.py b/conformance/tests/l1-rendering/vuln-scanner-acceptances/check.py new file mode 100644 index 0000000..6fb3560 --- /dev/null +++ b/conformance/tests/l1-rendering/vuln-scanner-acceptances/check.py @@ -0,0 +1,111 @@ +#!/usr/bin/env python3 +"""R-VULN-08: acceptance parsing, expiry and fail-safe behaviour.""" +import sys +from datetime import date, datetime, timedelta, timezone +from pathlib import Path + +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE.parents[3] / "platform/vuln-scanner/src")) +import acceptances # noqa: E402 +import report # noqa: E402 + +FAILED = [] +TODAY = date(2026, 11, 9) +NOW = datetime(2026, 11, 9, 12, 0, tzinfo=timezone.utc) +A1 = "sha256:" + "a1" * 32 + + +def check(name, condition): + print(("ok " if condition else "FAIL ") + name) + if not condition: + FAILED.append(name) + + +def parse(text): + return acceptances.parse_text(text, TODAY) + + +entries, errors = parse("""# header comment +- id: CVE-2026-1 + image: docker.io/library/nginx # exact match + reason: not reachable - feature disabled + expires: 2027-02-01 + +- id: CVE-2026-2 + reason: "platform wide" + expires: 2026-12-01 +""") +check("valid flat entries parse with no errors", len(entries) == 2 and errors == []) +check("inline comment is stripped", entries[0]["image"] == "docker.io/library/nginx") +check("omitted image means platform-wide", entries[1]["image"] is None) +check("empty list form is accepted", parse("# none\n[]\n") == ([], [])) +check("empty file is accepted", parse("") == ([], [])) + +for name, text in ( + ("missing reason", "- id: CVE-1\n expires: 2026-12-01\n"), + ("missing expires", "- id: CVE-1\n reason: x\n"), + ("expires beyond 180 days", "- id: CVE-1\n reason: x\n expires: 2027-06-01\n"), + ("unknown key", "- id: CVE-1\n reason: x\n expires: 2026-12-01\n scope: all\n"), + ("not a date", "- id: CVE-1\n reason: x\n expires: soon\n"), + ("impossible date", "- id: CVE-1\n reason: x\n expires: 2026-13-45\n"), + ("duplicate key", "- id: CVE-1\n reason: x\n reason: y\n expires: 2026-12-01\n"), + ("flow value", "- id: CVE-1\n reason: [x]\n expires: 2026-12-01\n"), +): + found, problems = parse(text) + check("rejected: " + name, found == [] and len(problems) == 1) + +found, problems = parse("- id: CVE-1\n reason: x\n expires: 2026-12-01\n- id: bad entry\n expires: 2026-12-01\n") +check("one bad entry is ignored, the valid one kept", len(found) == 1 and len(problems) == 1) +found, problems = parse("- id: CVE-1\n reason: x\n") +check("bad indentation invalidates the whole file", found == [] and problems) +found, problems = parse("garbage\n- id: CVE-1\n reason: x\n expires: 2026-12-01\n") +check("a stray line invalidates the whole file", found == [] and problems) +check("missing file is an error, not an empty allowlist", acceptances.load("/nonexistent/acceptances.yaml")[1]) + +entry = {"id": "CVE-1", "image": None, "reason": "r", "expires": date(2026, 11, 9)} +check("acceptance holds on its expiry day", acceptances.match([entry], "CVE-1", "any/image", TODAY) is entry) +check("acceptance gone the day after", acceptances.match([entry], "CVE-1", "any/image", TODAY + timedelta(days=1)) is None) +scoped = {"id": "CVE-1", "image": "docker.io/library/nginx", "reason": "r", "expires": date(2026, 12, 1)} +check("image match is exact", acceptances.match([scoped], "CVE-1", "docker.io/library/nginx-extra", TODAY) is None) +check("image match accepts the exact image", acceptances.match([scoped], "CVE-1", "docker.io/library/nginx", TODAY) is scoped) + + +def state_with(findings): + target = {"id": "docker.io/library/nginx@" + A1, "type": "image", "image": "docker.io/library/nginx", "digest": A1, + "ref": "", "reason": None, "workloads": []} + return { + "inventory": [target], "database": {"updatedAt": "2026-11-09T06:00:00Z", "version": "2"}, + "run": {"startedAt": "2026-11-09T09:00:00Z", "finishedAt": "2026-11-09T09:30:00Z", "status": "complete"}, + "remediation": [], "inventoryError": None, "loadError": None, + "scans": {A1: {"firstDiscoveredAt": "2026-11-01T00:00:00Z", "lastInventoriedAt": "2026-11-09T09:00:00Z", + "lastAttemptAt": "2026-11-09T09:00:00Z", "lastSuccessAt": "2026-11-09T09:00:00Z", "outcome": "ok", + "reason": None, "failedSince": None, "findings": findings, + "firstSeen": {f["id"]: "2026-11-01T00:00:00Z" for f in findings}}}, + } + + +FINDING = {"id": "CVE-1", "severity": "CRITICAL", "cvss": 9.0, "package": "p", "installedVersion": "1", "fixedVersion": "2", + "title": "t", "url": ""} +ST = state_with([FINDING]) + + +def status(when, accepted, errs=()): + built = report.build(ST, accepted, list(errs), when) + return built, built["findings"][0] + + +entry = {"id": "CVE-1", "image": None, "reason": "not reachable", "expires": date(2026, 11, 9)} +built, finding = status(NOW, [entry]) +check("unexpired acceptance marks the finding accepted", finding["status"] == "accepted" + and finding["acceptance"] == {"reason": "not reachable", "expires": "2026-11-09"}) +check("accepted report is schema-valid", report.validate(built) == []) +_, finding = status(NOW + timedelta(days=1), [entry]) +check("expired acceptance reverts to the finding's own status", finding["status"] == "fix-available" and "acceptance" not in finding) +_, finding = status(NOW, []) +check("mutation: no acceptance leaves it open/fix-available", finding["status"] == "fix-available") +built, finding = status(NOW, [], ["bad entry"]) +check("an invalid entry degrades the run", built["run"]["status"] == "degraded") +check("an invalid entry never hides the finding", finding["status"] == "fix-available") + +print("FAILED: " + str(len(FAILED)) if FAILED else "all checks passed") +sys.exit(1 if FAILED else 0) diff --git a/conformance/tests/l1-rendering/vuln-scanner-acceptances/check.sh b/conformance/tests/l1-rendering/vuln-scanner-acceptances/check.sh new file mode 100755 index 0000000..77679ca --- /dev/null +++ b/conformance/tests/l1-rendering/vuln-scanner-acceptances/check.sh @@ -0,0 +1,7 @@ +#!/usr/bin/env bash +# Runs the fixture-driven assertions in check.py. See chainsaw-test.yaml for the requirement tags. +# What this does NOT prove: anything at runtime (real Trivy, real API, real browser). L2/L3 cover that. +set -euo pipefail +HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +export PYTHONDONTWRITEBYTECODE=1 +exec python3 "${HERE}/check.py" diff --git a/conformance/tests/l1-rendering/vuln-scanner-api-metrics/chainsaw-test.yaml b/conformance/tests/l1-rendering/vuln-scanner-api-metrics/chainsaw-test.yaml new file mode 100644 index 0000000..0d96fa4 --- /dev/null +++ b/conformance/tests/l1-rendering/vuln-scanner-api-metrics/chainsaw-test.yaml @@ -0,0 +1,28 @@ +# L1 fixture test: runs check.py (python3 stdlib, source imported from platform/vuln-scanner/src, +# no cluster, no network beyond loopback). conformance/run.sh --layer l1 runs check.sh offline; +# this Chainsaw Test wraps the same script so an in-cluster run executes identical assertions. +apiVersion: chainsaw.kyverno.io/v1alpha1 +kind: Test +metadata: + name: vuln-scanner-api-metrics + labels: + conformance.home.k3s/R-VULN-02: "true" + conformance.home.k3s/R-VULN-03: "true" + conformance.home.k3s/R-VULN-05: "true" + conformance.home.k3s/R-VULN-06: "true" + annotations: + conformance.home.k3s/requirements: "R-VULN-02, R-VULN-03, R-VULN-05, R-VULN-06" + conformance.home.k3s/layer: "L1" + conformance.home.k3s/safety-critical: "false" + # Whole-repo scope: the touched-path map keys nothing to platform/vuln-scanner, so without this + # a change to the scanner would select no vuln-scanner test. + conformance.home.k3s/scope: "whole-repo" + conformance.home.k3s/intent: >- + Given a running engine, When the HTTP surfaces are exercised, Then the API validates, metrics are aggregates only, and listeners answer during a scan. +spec: + steps: + - name: run-offline-check + description: Run check.sh, the same assertions the offline harness runs. + try: + - script: + content: bash ./check.sh diff --git a/conformance/tests/l1-rendering/vuln-scanner-api-metrics/check.py b/conformance/tests/l1-rendering/vuln-scanner-api-metrics/check.py new file mode 100644 index 0000000..4b81880 --- /dev/null +++ b/conformance/tests/l1-rendering/vuln-scanner-api-metrics/check.py @@ -0,0 +1,226 @@ +#!/usr/bin/env python3 +"""R-VULN-02/03/05/06: internal API, metrics, publish refusal, and HTTP during a scan.""" +import copy +import http.client +import json +import logging +import os +import re +import sys +import tempfile +import threading +from datetime import datetime, timedelta, timezone +from pathlib import Path + +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE.parents[3] / "platform/vuln-scanner/src")) +import main # noqa: E402 +import metrics # noqa: E402 +import report # noqa: E402 +import server # noqa: E402 +import state as st # noqa: E402 + +logging.disable(logging.CRITICAL) +FAILED = [] +NOW = datetime(2026, 11, 9, 12, 0, tzinfo=timezone.utc) +A1 = "sha256:" + "a1" * 32 + + +def check(name, condition): + print(("ok " if condition else "FAIL ") + name) + if not condition: + FAILED.append(name) + + +class Clock: + def __init__(self): + self.now = NOW + + def __call__(self): + return self.now + + +class Client: + def list_all(self, path): + if path.endswith("pods"): + return [{"metadata": {"namespace": "apps", "name": "p"}, "spec": {"containers": [{"name": "c", "image": "nginx:1"}]}, + "status": {"containerStatuses": [{"name": "c", "imageID": "docker.io/library/nginx@" + A1}]}}] + return [] + + +class Runner: + """Blocks on release while a scan is 'running', so a test can prove the HTTP path stays live.""" + + def __init__(self): + self.entered = threading.Event() + self.release = threading.Event() + self.release.set() + self.result = {"SchemaVersion": 2, "Results": [{"Vulnerabilities": [ + {"VulnerabilityID": "CVE-1", "Severity": "CRITICAL", "PkgName": "openssl", "InstalledVersion": "1", + "FixedVersion": "2", "Title": "t", "PrimaryURL": "https://example.test/1"}]}]} + + def download_db(self): + pass + + def db_info(self): + return {"updatedAt": "2026-11-09T06:00:00Z", "version": "2"} + + def scan(self, _reference): + self.entered.set() + self.release.wait(10) + return self.result + + def clean_cache(self): + pass + + +def request(port, method, path, body=None, headers=None): + conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10) + conn.request(method, path, body, headers or {}) + resp = conn.getresponse() + data = resp.read() + conn.close() + return resp.status, data, resp + + +def post(port, document, content_type="application/json"): + return request(port, "POST", "/triage", json.dumps(document), {"Content-Type": content_type}) + + +with tempfile.TemporaryDirectory() as tmp: + accept = os.path.join(tmp, "a.yaml") + Path(accept).write_text("[]\n") + clock = Clock() + runner = Runner() + eng = main.Engine(tmp, Client(), runner, accept, clock=clock, publish_seconds=0) + servers = server.start_all(eng, ("127.0.0.1", 0), ("127.0.0.1", 0), ("127.0.0.1", 0)) + page_port, internal, metrics_port = (s.server_address[1] for s in servers) + + status, body, _ = request(internal, "GET", "/report.json") + check("report.json is 503 before any report", status == 503) + status, body, resp = request(page_port, "GET", "/") + check("page is 503 before any report, with the CSP", status == 503 and resp.getheader("Content-Security-Policy")) + text = request(metrics_port, "GET", "/metrics")[1].decode() + check("last-complete-run metric is absent before the first run", "last_complete_run" not in text) + check("triage heartbeat metric is absent before the first POST", "triage_heartbeat" not in text) + + # HTTP must stay live while a scan is blocked mid-flight. + runner.release.clear() + worker = threading.Thread(target=eng.tick, daemon=True) + worker.start() + check("the scan is genuinely in flight", runner.entered.wait(10)) + status_h, _, _ = request(internal, "GET", "/healthz") + status_p, _, _ = request(page_port, "GET", "/healthz") + status_m, _, _ = request(metrics_port, "GET", "/metrics") + check("listeners answer during a scan", (status_h, status_p, status_m) == (200, 200, 200)) + runner.release.set() + worker.join(10) + check("the tick then completes", not worker.is_alive()) + + status, body, _ = request(internal, "GET", "/report.json") + check("report.json serves a schema-valid report after a tick", status == 200 and report.validate(json.loads(body)) == []) + + # Page and CSP over HTTP + status, body, resp = request(page_port, "GET", "/?view=inventory") + check("page served over HTTP carries the CSP and nosniff", + status == 200 and resp.getheader("Content-Security-Policy").startswith("default-src 'none'") and resp.getheader("X-Content-Type-Options") == "nosniff") + status, body, resp = request(page_port, "GET", "/missing") + check("a 404 from the page listener also carries the CSP", status == 404 and resp.getheader("Content-Security-Policy")) + + status, body, resp = request(page_port, "POST", "/") + check("an unsupported method on the page listener also carries the CSP", status == 501 and resp.getheader("Content-Security-Policy")) + + # Triage API + valid = {"heartbeatAt": "2026-11-09T11:00:00Z", "remediation": [ + {"image": "docker.io/library/nginx", "state": "pr-open", "url": "https://gitea.test/pulls/9", "updatedAt": "2026-11-09T11:00:00Z"}]} + status, body, _ = post(internal, valid) + check("valid triage is 204", status == 204 and body == b"") + check("remediation surfaces in the report", eng.report["remediation"][0]["state"] == "pr-open") + text = request(metrics_port, "GET", "/metrics")[1].decode() + check("heartbeat metric appears and uses server receipt time, not the bot's", f"triage_heartbeat_timestamp_seconds {int(NOW.timestamp())}" in text) + for name, mutate in ( + ("unknown top-level key", lambda d: d.update(extra=1)), + ("missing heartbeat", lambda d: d.pop("heartbeatAt")), + ("unknown remediation key", lambda d: d["remediation"][0].update(x=1)), + ("bad remediation state", lambda d: d["remediation"][0].update(state="merged")), + ("javascript url", lambda d: d["remediation"][0].update(url="javascript:x")), + ("bad timestamp", lambda d: d.update(heartbeatAt="yesterday")), + ): + document = copy.deepcopy(valid) + mutate(document) + status, _, _ = post(internal, document) + check("triage rejects " + name, status == 400) + check("triage rejects a wrong content type", post(internal, valid, "text/plain")[0] == 415) + status, _, _ = request(internal, "POST", "/triage", b"x" * 300000, {"Content-Type": "application/json"}) + check("triage rejects an oversized body", status == 413) + many = {"heartbeatAt": "2026-11-09T11:00:00Z", "remediation": [valid["remediation"][0]] * 5001} + check("triage rejects more than 5000 entries", post(internal, many)[0] in (400, 413)) + check("rejected triage leaves state untouched", eng.state["remediation"][0]["url"] == "https://gitea.test/pulls/9") + + # Metrics labels and values + text = request(metrics_port, "GET", "/metrics")[1].decode() + allowed = {"state", "severity", "fixable"} + labels = {k for line in text.splitlines() for k in re.findall(r"([A-Za-z_]+)=\"", line)} + check("metric labels are within the allowlist", labels and labels <= allowed) + check("no metric carries a CVE, package, image or workload", not re.search(r"CVE-|openssl|nginx|apps", text)) + for state in report.STATES: + check("targets gauge present for " + state, 'vuln_scanner_targets{state="' + state + '"}' in text) + check("critical fix findings counted with fixable true", 'vuln_scanner_findings{severity="CRITICAL",fixable="true"} 1' in text) + check("every severity and fixability appears", text.count("vuln_scanner_findings{") == 10) + check("failed-over-24h gauge exists", "vuln_scanner_failed_targets_over_24h 0" in text) + check("critical fix overdue is 0 inside the 72h grace", "vuln_scanner_critical_fix_overdue 0" in text) + clock.now += timedelta(hours=73) + text = request(metrics_port, "GET", "/metrics")[1].decode() + check("critical fix overdue is 1 after 72h", "vuln_scanner_critical_fix_overdue 1" in text) + check("last-complete-run metric present after a complete run", "vuln_scanner_last_complete_run_timestamp_seconds" in text) + for srv in servers: + srv.shutdown() + +# Invalid report: refuse and keep the old one. +with tempfile.TemporaryDirectory() as tmp: + good = {"schemaVersion": 1, "generatedAt": "2026-11-09T10:00:00Z", "database": {"updatedAt": "2026-11-09T06:00:00Z", "version": "2"}, + "run": {"startedAt": "2026-11-09T09:00:00Z", "finishedAt": "2026-11-09T09:30:00Z", "status": "complete"}, + "targets": [], "findings": [], "remediation": []} + check("a valid report publishes", report.publish(good, tmp) is True) + before = Path(tmp, "report.json").read_bytes() + for name, mutate in ( + ("wrong schemaVersion", lambda r: r.update(schemaVersion=2)), + ("extra top-level key", lambda r: r.update(extra=1)), + ("bad timestamp", lambda r: r.update(generatedAt="soon")), + ("unknown state", lambda r: r.update(targets=[{"id": "x", "type": "image", "image": "x", "digest": None, "state": "fine", + "reason": None, "lastScannedAt": None, "workloads": []}])), + ("dangling targetId", lambda r: r.update(findings=[{"id": "CVE-1", "targetId": "nope", "severity": "LOW", "cvss": None, + "package": "p", "installedVersion": "1", "fixedVersion": None, "title": "t", + "url": "", "firstSeenAt": "2026-11-01T00:00:00Z", "status": "open"}])), + ): + broken = copy.deepcopy(good) + mutate(broken) + check("invalid report refused: " + name, report.publish(broken, tmp) is False) + check("an invalid report writes nothing", Path(tmp, "report.json").read_bytes() == before) + check("accepted without acceptance is invalid", report.validate({**good, "targets": [{"id": "x", "type": "image", "image": "x", "digest": None, + "state": "scanned", "reason": None, "lastScannedAt": None, "workloads": []}], "findings": [{"id": "CVE-1", "targetId": "x", "severity": "LOW", + "cvss": None, "package": "p", "installedVersion": "1", "fixedVersion": None, "title": "t", "url": "", "firstSeenAt": "2026-11-01T00:00:00Z", + "status": "accepted"}]}) != []) + Path(tmp, "report.json").write_text("{not json") + check("a corrupt published report is not served", report.load_published(tmp) is None) + +# Crash-consistency of state: the temp+rename leaves no temp files behind and the file parses. +with tempfile.TemporaryDirectory() as tmp: + store = st.Store(tmp) + state = st.new_state() + state["triageHeartbeatAt"] = "2026-11-09T11:00:00Z" + store.save(state) + check("state saves with no temp file left behind", sorted(os.listdir(tmp)) == ["state.json"]) + check("state survives a reload", st.Store(tmp).load()["triageHeartbeatAt"] == "2026-11-09T11:00:00Z") + Path(tmp, "state.json").write_text("not json") + loaded = st.Store(tmp).load() + check("corrupt state starts empty and marks degraded", loaded["loadError"] and loaded["scans"] == {}) + check("corrupt state is preserved for inspection", Path(tmp, "state.json.corrupt").exists()) + +# Metrics module directly, without the report. +empty_state = st.new_state() +text = metrics.render(None, empty_state, NOW) +check("with no report only the state-derived gauge exists", text == "vuln_scanner_failed_targets_over_24h 0\n") + +print("FAILED: " + str(len(FAILED)) if FAILED else "all checks passed") +sys.exit(1 if FAILED else 0) diff --git a/conformance/tests/l1-rendering/vuln-scanner-api-metrics/check.sh b/conformance/tests/l1-rendering/vuln-scanner-api-metrics/check.sh new file mode 100755 index 0000000..77679ca --- /dev/null +++ b/conformance/tests/l1-rendering/vuln-scanner-api-metrics/check.sh @@ -0,0 +1,7 @@ +#!/usr/bin/env bash +# Runs the fixture-driven assertions in check.py. See chainsaw-test.yaml for the requirement tags. +# What this does NOT prove: anything at runtime (real Trivy, real API, real browser). L2/L3 cover that. +set -euo pipefail +HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +export PYTHONDONTWRITEBYTECODE=1 +exec python3 "${HERE}/check.py" diff --git a/conformance/tests/l1-rendering/vuln-scanner-inventory/chainsaw-test.yaml b/conformance/tests/l1-rendering/vuln-scanner-inventory/chainsaw-test.yaml new file mode 100644 index 0000000..49e46a7 --- /dev/null +++ b/conformance/tests/l1-rendering/vuln-scanner-inventory/chainsaw-test.yaml @@ -0,0 +1,25 @@ +# L1 fixture test: runs check.py (python3 stdlib, source imported from platform/vuln-scanner/src, +# no cluster, no network beyond loopback). conformance/run.sh --layer l1 runs check.sh offline; +# this Chainsaw Test wraps the same script so an in-cluster run executes identical assertions. +apiVersion: chainsaw.kyverno.io/v1alpha1 +kind: Test +metadata: + name: vuln-scanner-inventory + labels: + conformance.home.k3s/R-VULN-01: "true" + annotations: + conformance.home.k3s/requirements: "R-VULN-01" + conformance.home.k3s/layer: "L1" + conformance.home.k3s/safety-critical: "false" + # Whole-repo scope: the touched-path map keys nothing to platform/vuln-scanner, so without this + # a change to the scanner would select no vuln-scanner test. + conformance.home.k3s/scope: "whole-repo" + conformance.home.k3s/intent: >- + Given recorded API lists, When the inventory is built, Then owner chains, digests, by-tag CronJobs and not-covered targets resolve per R-VULN-01. +spec: + steps: + - name: run-offline-check + description: Run check.sh, the same assertions the offline harness runs. + try: + - script: + content: bash ./check.sh diff --git a/conformance/tests/l1-rendering/vuln-scanner-inventory/check.py b/conformance/tests/l1-rendering/vuln-scanner-inventory/check.py new file mode 100644 index 0000000..14feece --- /dev/null +++ b/conformance/tests/l1-rendering/vuln-scanner-inventory/check.py @@ -0,0 +1,155 @@ +#!/usr/bin/env python3 +"""R-VULN-01: inventory from recorded API lists.""" +import copy +import http.server +import json +import os +import sys +import tempfile +import threading +from pathlib import Path + +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE.parents[3] / "platform/vuln-scanner/src")) +import inventory # noqa: E402 + +FAILED = [] + + +def check(name, condition): + print(("ok " if condition else "FAIL ") + name) + if not condition: + FAILED.append(name) + + +CLUSTER = json.loads((HERE / "fixtures/cluster.json").read_text()) +A1 = "sha256:" + "a1" * 32 +B2 = "sha256:" + "b2" * 32 +C3 = "sha256:" + "c3" * 32 +D4 = "sha256:" + "d4" * 32 +E5 = "sha256:" + "e5" * 32 +F6 = "sha256:" + "f6" * 32 + + +def by_id(targets): + return {t["id"]: t for t in targets} + + +def workloads(target): + return {(w["namespace"], w["kind"], w["name"]): w["containers"] for w in target["workloads"]} + + +targets = inventory.build(CLUSTER) +ids = by_id(targets) + +nginx = ids["docker.io/library/nginx@" + A1] +check("ReplicaSet resolves to Deployment, not the ReplicaSet", workloads(nginx) == { + ("apps", "Deployment", "web"): ["app"], ("apps", "Deployment", "api"): ["server"]}) +check("docker-pullable:// prefix stripped and digest taken from imageID", nginx["digest"] == A1) +check("init container inventoried", ("apps", "Deployment", "web") in workloads(ids["ghcr.io/org/tool@" + B2])) +check("ephemeral container inventoried", "debug" in workloads(ids["docker.io/library/busybox@" + C3])[("apps", "Deployment", "web")]) + +backup = ids["ghcr.io/org/backup@" + E5] +check("Job resolves to CronJob", workloads(backup) == {("ops", "CronJob", "backup"): ["backup"]}) +check("a CronJob with a running pod is not also inventoried from its template", + not any(t["id"].startswith("ghcr.io/org/backup:") for t in targets)) + +nightly = ids["ghcr.io/org/nightly:2.1"] +check("idle CronJob inventoried from template with no digest", nightly["digest"] is None + and workloads(nightly) == {("ops", "CronJob", "nightly"): ["job"]}) + +solo = ids["docker.io/library/alpine:3.20"] +check("empty imageID gives a by-tag target from the spec image", solo["digest"] is None) +check("bare Pod is its own workload", ("ops", "Pod", "solo") in workloads(solo)) +check("Succeeded pod's image is not inventoried", not any(F6 in t["id"] for t in targets)) +check("virt-launcher stays owned by its VirtualMachineInstance", + ("kubevirt-vms", "VirtualMachineInstance", "vm1") in workloads(ids["quay.io/kubevirt/virt-launcher@sha256:" + "ab" * 32])) +vm, node = ids["vm/kubevirt-vms/vm1"], ids["node/prod-1"] +check("VM inventoried as non-container target", vm["type"] == "vm" and vm["digest"] is None and vm["reason"]) +check("node inventoried as non-container target", node["type"] == "node" and node["reason"]) + +# Non-vacuous: mutating the fixture changes the answer the assertions above rely on. +broken = copy.deepcopy(CLUSTER) +broken["replicasets"] = [] +broken["jobs"] = [] +broken_ids = by_id(inventory.build(broken)) +check("mutation: losing the ReplicaSet record leaves the ReplicaSet as the workload", + ("apps", "ReplicaSet", "web-5f") in workloads(broken_ids["docker.io/library/nginx@" + A1])) +check("mutation: losing the Job record leaves the Job as the workload", + ("ops", "Job", "backup-123") in workloads(broken_ids["ghcr.io/org/backup@" + E5])) +check("mutation: a Running phase on the Succeeded pod does inventory its image", any( + F6 in t["id"] for t in inventory.build({**CLUSTER, "pods": [ + {**p, "status": {**p["status"], "phase": "Running"}} if p["metadata"]["name"] == "done-1" else p + for p in CLUSTER["pods"]]}))) + +for reference, expected in ( + ("nginx:1.27", ("docker.io/library/nginx", "1.27", None)), + ("nginx", ("docker.io/library/nginx", None, None)), + ("bitnami/redis:7", ("docker.io/bitnami/redis", "7", None)), + ("ghcr.io/org/app:v1", ("ghcr.io/org/app", "v1", None)), + ("registry.local:5000/x/y:2", ("registry.local:5000/x/y", "2", None)), + ("index.docker.io/library/nginx:1", ("docker.io/library/nginx", "1", None)), + ("ghcr.io/org/app:v1@" + A1, ("ghcr.io/org/app", "v1", A1)), +): + check("normalise " + reference, inventory.normalise_reference(reference) == expected) +check("bare config id is not a digest", inventory.digest_from_image_id("sha256:" + "a1" * 32) is None) +check("empty imageID has no digest", inventory.digest_from_image_id("") is None) + + +class FakeListClient: + """list_all replaced; proves collect() tolerates exactly a 404 on the VMI API.""" + + def __init__(self, vmi_error): + self.vmi_error = vmi_error + + def list_all(self, path): + if "kubevirt" in path and self.vmi_error: + raise self.vmi_error + return [] + + +check("VMI API 404 tolerated as no VMs", inventory.collect(FakeListClient(inventory.NotFound("x")))["vmis"] == []) +try: + inventory.collect(FakeListClient(inventory.InventoryError("403"))) + check("a non-404 VMI error is not swallowed", False) +except inventory.InventoryError: + check("a non-404 VMI error is not swallowed", True) + + +class PagedAPI(http.server.BaseHTTPRequestHandler): + def log_message(self, *args): + return + + def do_GET(self): + if self.headers.get("Authorization") != "Bearer fixture-token": + self.send_response(403) + self.end_headers() + return + if self.path.startswith("/apis/kubevirt"): + self.send_response(404) + self.end_headers() + return + if "continue=" in self.path: + body = {"items": [{"metadata": {"name": "b"}}], "metadata": {}} + else: + body = {"items": [{"metadata": {"name": "a"}}], "metadata": {"continue": "page2"}} + payload = json.dumps(body).encode() + self.send_response(200) + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + +with tempfile.TemporaryDirectory() as tmp: + token = os.path.join(tmp, "token") + Path(token).write_text("fixture-token\n") + server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), PagedAPI) + threading.Thread(target=server.serve_forever, daemon=True).start() + client = inventory.KubeClient("http://127.0.0.1:" + str(server.server_address[1]), token, os.devnull) + items = client.list_all("/api/v1/pods") + check("pagination follows the continue token with the bearer token", [i["metadata"]["name"] for i in items] == ["a", "b"]) + check("collect over real HTTP tolerates VMI 404", inventory.collect(client)["vmis"] == []) + server.shutdown() + +print("FAILED: " + str(len(FAILED)) if FAILED else "all checks passed") +sys.exit(1 if FAILED else 0) diff --git a/conformance/tests/l1-rendering/vuln-scanner-inventory/check.sh b/conformance/tests/l1-rendering/vuln-scanner-inventory/check.sh new file mode 100755 index 0000000..77679ca --- /dev/null +++ b/conformance/tests/l1-rendering/vuln-scanner-inventory/check.sh @@ -0,0 +1,7 @@ +#!/usr/bin/env bash +# Runs the fixture-driven assertions in check.py. See chainsaw-test.yaml for the requirement tags. +# What this does NOT prove: anything at runtime (real Trivy, real API, real browser). L2/L3 cover that. +set -euo pipefail +HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +export PYTHONDONTWRITEBYTECODE=1 +exec python3 "${HERE}/check.py" diff --git a/conformance/tests/l1-rendering/vuln-scanner-inventory/fixtures/cluster.json b/conformance/tests/l1-rendering/vuln-scanner-inventory/fixtures/cluster.json new file mode 100644 index 0000000..c3656cc --- /dev/null +++ b/conformance/tests/l1-rendering/vuln-scanner-inventory/fixtures/cluster.json @@ -0,0 +1,395 @@ +{ + "cronjobs": [ + { + "metadata": { + "name": "backup", + "namespace": "ops" + }, + "spec": { + "jobTemplate": { + "spec": { + "template": { + "spec": { + "containers": [ + { + "image": "ghcr.io/org/backup:1.0", + "name": "job" + } + ] + } + } + } + } + } + }, + { + "metadata": { + "name": "nightly", + "namespace": "ops" + }, + "spec": { + "jobTemplate": { + "spec": { + "template": { + "spec": { + "containers": [ + { + "image": "ghcr.io/org/nightly:2.1", + "name": "job" + } + ] + } + } + } + } + } + } + ], + "jobs": [ + { + "metadata": { + "name": "backup-123", + "namespace": "ops", + "ownerReferences": [ + { + "controller": true, + "kind": "CronJob", + "name": "backup" + } + ] + } + }, + { + "metadata": { + "name": "backup-99", + "namespace": "ops", + "ownerReferences": [ + { + "controller": true, + "kind": "CronJob", + "name": "backup" + } + ] + } + } + ], + "nodes": [ + { + "metadata": { + "name": "prod-1" + } + }, + { + "metadata": { + "name": "prod-2" + } + } + ], + "pods": [ + { + "metadata": { + "name": "web-1", + "namespace": "apps", + "ownerReferences": [ + { + "controller": true, + "kind": "ReplicaSet", + "name": "web-5f" + } + ] + }, + "spec": { + "containers": [ + { + "image": "nginx:1.27", + "name": "app" + } + ], + "ephemeralContainers": [ + { + "image": "busybox:1.37", + "name": "debug" + } + ], + "initContainers": [ + { + "image": "ghcr.io/org/tool:v2", + "name": "init-db" + } + ] + }, + "status": { + "containerStatuses": [ + { + "imageID": "docker-pullable://docker.io/library/nginx@sha256:a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1", + "name": "app" + } + ], + "ephemeralContainerStatuses": [ + { + "imageID": "docker.io/library/busybox@sha256:c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3", + "name": "debug" + } + ], + "initContainerStatuses": [ + { + "imageID": "ghcr.io/org/tool@sha256:b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2", + "name": "init-db" + } + ], + "phase": "Running" + } + }, + { + "metadata": { + "name": "web-2", + "namespace": "apps", + "ownerReferences": [ + { + "controller": true, + "kind": "ReplicaSet", + "name": "web-5f" + } + ] + }, + "spec": { + "containers": [ + { + "image": "nginx:1.27", + "name": "app" + } + ], + "initContainers": [ + { + "image": "ghcr.io/org/tool:v2", + "name": "init-db" + } + ] + }, + "status": { + "containerStatuses": [ + { + "imageID": "docker.io/library/nginx@sha256:a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1", + "name": "app" + } + ], + "initContainerStatuses": [ + { + "imageID": "ghcr.io/org/tool@sha256:b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2", + "name": "init-db" + } + ], + "phase": "Running" + } + }, + { + "metadata": { + "name": "api-1", + "namespace": "apps", + "ownerReferences": [ + { + "controller": true, + "kind": "ReplicaSet", + "name": "api-77" + } + ] + }, + "spec": { + "containers": [ + { + "image": "nginx:1.27", + "name": "server" + } + ] + }, + "status": { + "containerStatuses": [ + { + "imageID": "docker.io/library/nginx@sha256:a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1", + "name": "server" + } + ], + "phase": "Running" + } + }, + { + "metadata": { + "name": "db-0", + "namespace": "data", + "ownerReferences": [ + { + "controller": true, + "kind": "StatefulSet", + "name": "db" + } + ] + }, + "spec": { + "containers": [ + { + "image": "postgres:16", + "name": "postgres" + } + ] + }, + "status": { + "containerStatuses": [ + { + "imageID": "docker.io/library/postgres@sha256:d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4", + "name": "postgres" + } + ], + "phase": "Running" + } + }, + { + "metadata": { + "name": "backup-123-x", + "namespace": "ops", + "ownerReferences": [ + { + "controller": true, + "kind": "Job", + "name": "backup-123" + } + ] + }, + "spec": { + "containers": [ + { + "image": "ghcr.io/org/backup:1.0", + "name": "backup" + } + ] + }, + "status": { + "containerStatuses": [ + { + "imageID": "ghcr.io/org/backup@sha256:e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5", + "name": "backup" + } + ], + "phase": "Running" + } + }, + { + "metadata": { + "name": "solo", + "namespace": "ops" + }, + "spec": { + "containers": [ + { + "image": "alpine:3.20", + "name": "toolbox" + } + ] + }, + "status": { + "containerStatuses": [ + { + "imageID": "", + "name": "toolbox" + } + ], + "phase": "Pending" + } + }, + { + "metadata": { + "name": "done-1", + "namespace": "ops", + "ownerReferences": [ + { + "controller": true, + "kind": "Job", + "name": "backup-99" + } + ] + }, + "spec": { + "containers": [ + { + "image": "ghcr.io/org/old-backup:0.1", + "name": "backup" + } + ] + }, + "status": { + "containerStatuses": [ + { + "imageID": "ghcr.io/org/old-backup@sha256:f6f6f6f6f6f6f6f6f6f6f6f6f6f6f6f6f6f6f6f6f6f6f6f6f6f6f6f6f6f6f6f6", + "name": "backup" + } + ], + "phase": "Succeeded" + } + }, + { + "metadata": { + "name": "virt-launcher-vm1-abc", + "namespace": "kubevirt-vms", + "ownerReferences": [ + { + "controller": true, + "kind": "VirtualMachineInstance", + "name": "vm1" + } + ] + }, + "spec": { + "containers": [ + { + "image": "quay.io/kubevirt/virt-launcher:v1.8", + "name": "compute" + } + ] + }, + "status": { + "containerStatuses": [ + { + "imageID": "quay.io/kubevirt/virt-launcher@sha256:abababababababababababababababababababababababababababababababab", + "name": "compute" + } + ], + "phase": "Running" + } + } + ], + "replicasets": [ + { + "metadata": { + "name": "web-5f", + "namespace": "apps", + "ownerReferences": [ + { + "controller": true, + "kind": "Deployment", + "name": "web" + } + ] + } + }, + { + "metadata": { + "name": "api-77", + "namespace": "apps", + "ownerReferences": [ + { + "controller": true, + "kind": "Deployment", + "name": "api" + } + ] + } + } + ], + "vmis": [ + { + "metadata": { + "name": "vm1", + "namespace": "kubevirt-vms" + } + } + ] +} diff --git a/conformance/tests/l1-rendering/vuln-scanner-page/chainsaw-test.yaml b/conformance/tests/l1-rendering/vuln-scanner-page/chainsaw-test.yaml new file mode 100644 index 0000000..50bbe42 --- /dev/null +++ b/conformance/tests/l1-rendering/vuln-scanner-page/chainsaw-test.yaml @@ -0,0 +1,26 @@ +# L1 fixture test: runs check.py (python3 stdlib, source imported from platform/vuln-scanner/src, +# no cluster, no network beyond loopback). conformance/run.sh --layer l1 runs check.sh offline; +# this Chainsaw Test wraps the same script so an in-cluster run executes identical assertions. +apiVersion: chainsaw.kyverno.io/v1alpha1 +kind: Test +metadata: + name: vuln-scanner-page + labels: + conformance.home.k3s/R-VULN-03: "true" + conformance.home.k3s/R-VULN-04: "true" + annotations: + conformance.home.k3s/requirements: "R-VULN-03, R-VULN-04" + conformance.home.k3s/layer: "L1" + conformance.home.k3s/safety-critical: "false" + # Whole-repo scope: the touched-path map keys nothing to platform/vuln-scanner, so without this + # a change to the scanner would select no vuln-scanner test. + conformance.home.k3s/scope: "whole-repo" + conformance.home.k3s/intent: >- + Given a report fixture, When the page renders, Then all CVEs, containers and workloads appear in all views and hostile input is inert. +spec: + steps: + - name: run-offline-check + description: Run check.sh, the same assertions the offline harness runs. + try: + - script: + content: bash ./check.sh diff --git a/conformance/tests/l1-rendering/vuln-scanner-page/check.py b/conformance/tests/l1-rendering/vuln-scanner-page/check.py new file mode 100644 index 0000000..3a41da1 --- /dev/null +++ b/conformance/tests/l1-rendering/vuln-scanner-page/check.py @@ -0,0 +1,181 @@ +#!/usr/bin/env python3 +"""R-VULN-03/04: the rendered page. Assertions run against the actual HTML output.""" +import base64 +import copy +import hashlib +import json +import re +import sys +from datetime import datetime, timedelta, timezone +from html.parser import HTMLParser +from pathlib import Path + +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE.parents[3] / "platform/vuln-scanner/src")) +import page # noqa: E402 + +FAILED = [] +NOW = datetime(2026, 11, 9, 12, 0, tzinfo=timezone.utc) +REPORT = json.loads((HERE / "fixtures/report.json").read_text()) +CLEAN_RE = re.compile(r"(?i)\bclean\b|no vulnerabilit") + + +def check(name, condition): + print(("ok " if condition else "FAIL ") + name) + if not condition: + FAILED.append(name) + + +def render(report=REPORT, query="", now=NOW): + status, headers, body = page.render(report, query, now) + return status, headers, body.decode("utf-8") + + +class Tags(HTMLParser): + def __init__(self): + super().__init__() + self.tags = [] + self.attrs = [] + + def handle_starttag(self, tag, attrs): + self.tags.append(tag) + self.attrs.append((tag, dict(attrs))) + + +def tags(body): + parser = Tags() + parser.feed(body) + return parser + + +def covers_everything(body, report): + """Every fixture CVE id, workload name and container name appears in the document.""" + missing = [f["id"] for f in report["findings"] if f["id"] not in body] + for target in report["targets"]: + for workload in target["workloads"]: + if workload["name"] not in body: + missing.append(workload["name"]) + missing.extend(c for c in workload["containers"] if c not in body) + return missing + + +def workload_view_covers(body, report): + """Every workload row with its containers, and every CVE in the expansions.""" + missing = covers_everything(body, report) + return missing + + +for view in ("cves", "workloads", "inventory"): + status, headers, body = render(query="view=" + view) + check(view + " view returns 200", status == 200) + check(view + " view has the Content-Security-Policy with no script source", "script-src" not in headers["Content-Security-Policy"] + and headers["Content-Security-Policy"].startswith("default-src 'none'")) + check(view + " view sets nosniff", headers["X-Content-Type-Options"] == "nosniff") + check(view + " view has no script element", "script" not in tags(body).tags) + style_body = re.search(r"", body, re.S).group(1) + expected = base64.b64encode(hashlib.sha256(style_body.encode()).digest()).decode() + check(view + " view's CSP hash matches its inline style", "'sha256-" + expected + "'" in headers["Content-Security-Policy"]) + check(view + " view header shows every state count", all(state + ": " in body for state in page.report_mod.STATES)) + check(view + " view shows a coverage warning", "INCOMPLETE COVERAGE" in body) + check(view + " view makes no clean claim with unscanned targets", not CLEAN_RE.search(body)) + +status, headers, cves = render(query="view=cves") +check("CVE view contains every CVE", all(f["id"] in cves for f in REPORT["findings"])) +affected = {f["targetId"] for f in REPORT["findings"]} +expected_in_cves = {(t["id"], w["name"]) for t in REPORT["targets"] if t["id"] in affected for w in t["workloads"]} +check("CVE view expansions name every workload of an affected target", all(name in cves for _, name in expected_in_cves)) +check("CVE view expansions name every container of an affected target", + all(c in cves for t in REPORT["targets"] if t["id"] in affected for w in t["workloads"] for c in w["containers"])) +for name in ("web", "api", "app", "server"): + check("CVE view expansion names " + name, name in cves) +status, headers, workloads = render(query="view=workloads") +check("workloads view contains every workload and container, including zero-finding ones", covers_everything(workloads, REPORT) == []) +status, headers, inventory = render(query="view=inventory") +check("inventory view contains every target", all(t["id"] in inventory for t in REPORT["targets"])) +check("inventory view contains every workload and container", covers_everything(inventory, REPORT) == []) +mutated = copy.deepcopy(REPORT) +mutated["findings"] = [f for f in mutated["findings"] if f["id"] != "CVE-2026-1003"] +check("mutation: the coverage evaluator catches a dropped CVE", "CVE-2026-1003" in covers_everything(render(mutated, "view=workloads")[2], REPORT)) + +_, _, body = render(query="view=cves") +check("hostile title is escaped", "<script>alert(1)</script>" in body) +check("hostile title is never live", "", + "url": "https://example.test/CVE-2026-1002" + }, + { + "acceptance": { + "expires": "2027-01-01", + "reason": "not reachable" + }, + "cvss": null, + "firstSeenAt": "2026-11-01T00:00:00Z", + "fixedVersion": "16.2", + "id": "CVE-2026-1003", + "installedVersion": "1.0", + "package": "libpq", + "severity": "HIGH", + "status": "accepted", + "targetId": "docker.io/library/postgres@sha256:d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4", + "title": "libpq flaw (retained from a prior scan)", + "url": "https://example.test/CVE-2026-1003" + }, + { + "cvss": null, + "firstSeenAt": "2026-11-01T00:00:00Z", + "fixedVersion": null, + "id": "CVE-2026-1004", + "installedVersion": "1.0", + "package": "curl", + "severity": "MEDIUM", + "status": "open", + "targetId": "ghcr.io/org/nightly:2.1", + "title": "curl flaw", + "url": "https://example.test/CVE-2026-1004" + } + ], + "generatedAt": "2026-11-09T10:00:00Z", + "remediation": [ + { + "image": "docker.io/library/nginx", + "state": "pr-open", + "updatedAt": "2026-11-09T10:00:00Z", + "url": "https://gitea.test/pulls/1" + } + ], + "run": { + "finishedAt": "2026-11-09T09:30:00Z", + "startedAt": "2026-11-09T09:00:00Z", + "status": "complete" + }, + "schemaVersion": 1, + "targets": [ + { + "digest": "sha256:a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1", + "id": "docker.io/library/nginx@sha256:a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1", + "image": "docker.io/library/nginx", + "lastScannedAt": "2026-11-09T10:00:00Z", + "reason": null, + "state": "scanned", + "type": "image", + "workloads": [ + { + "containers": [ + "app" + ], + "kind": "Deployment", + "name": "web", + "namespace": "apps" + }, + { + "containers": [ + "server" + ], + "kind": "Deployment", + "name": "api", + "namespace": "apps" + } + ] + }, + { + "digest": "sha256:d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4", + "id": "docker.io/library/postgres@sha256:d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4", + "image": "docker.io/library/postgres", + "lastScannedAt": "2026-11-07T09:00:00Z", + "reason": "trivy timed out", + "state": "failed", + "type": "image", + "workloads": [ + { + "containers": [ + "postgres" + ], + "kind": "StatefulSet", + "name": "db", + "namespace": "data" + } + ] + }, + { + "digest": "sha256:e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5", + "id": "ghcr.io/org/private@sha256:e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5", + "image": "ghcr.io/org/private", + "lastScannedAt": null, + "reason": "registry ghcr.io refused anonymous access", + "state": "credential-required", + "type": "image", + "workloads": [ + { + "containers": [ + "priv" + ], + "kind": "Deployment", + "name": "priv", + "namespace": "apps" + } + ] + }, + { + "digest": null, + "id": "ghcr.io/org/nightly:2.1", + "image": "ghcr.io/org/nightly", + "lastScannedAt": "2026-11-09T10:00:00Z", + "reason": null, + "state": "by-tag", + "type": "image", + "workloads": [ + { + "containers": [ + "job" + ], + "kind": "CronJob", + "name": "nightly", + "namespace": "ops" + } + ] + }, + { + "digest": null, + "id": "vm/kubevirt-vms/vm1", + "image": "", + "lastScannedAt": null, + "reason": "guest OS packages are not assessed by this scanner", + "state": "not-covered", + "type": "vm", + "workloads": [ + { + "containers": [], + "kind": "VirtualMachineInstance", + "name": "vm1", + "namespace": "kubevirt-vms" + } + ] + }, + { + "digest": null, + "id": "node/prod-1", + "image": "", + "lastScannedAt": null, + "reason": "node OS packages and the k3s binary are not assessed by this scanner", + "state": "not-covered", + "type": "node", + "workloads": [ + { + "containers": [], + "kind": "Node", + "name": "prod-1", + "namespace": "" + } + ] + } + ] +} diff --git a/conformance/tests/l1-rendering/vuln-scanner-scan-loop/chainsaw-test.yaml b/conformance/tests/l1-rendering/vuln-scanner-scan-loop/chainsaw-test.yaml new file mode 100644 index 0000000..68a3c56 --- /dev/null +++ b/conformance/tests/l1-rendering/vuln-scanner-scan-loop/chainsaw-test.yaml @@ -0,0 +1,27 @@ +# L1 fixture test: runs check.py (python3 stdlib, source imported from platform/vuln-scanner/src, +# no cluster, no network beyond loopback). conformance/run.sh --layer l1 runs check.sh offline; +# this Chainsaw Test wraps the same script so an in-cluster run executes identical assertions. +apiVersion: chainsaw.kyverno.io/v1alpha1 +kind: Test +metadata: + name: vuln-scanner-scan-loop + labels: + conformance.home.k3s/R-VULN-02: "true" + conformance.home.k3s/R-VULN-03: "true" + conformance.home.k3s/R-VULN-05: "true" + annotations: + conformance.home.k3s/requirements: "R-VULN-02, R-VULN-03, R-VULN-05" + conformance.home.k3s/layer: "L1" + conformance.home.k3s/safety-critical: "false" + # Whole-repo scope: the touched-path map keys nothing to platform/vuln-scanner, so without this + # a change to the scanner would select no vuln-scanner test. + conformance.home.k3s/scope: "whole-repo" + conformance.home.k3s/intent: >- + Given a fake Trivy runner and clock, When ticks run, Then shared digests scan once, failures isolate, and coverage states and first-seen persist. +spec: + steps: + - name: run-offline-check + description: Run check.sh, the same assertions the offline harness runs. + try: + - script: + content: bash ./check.sh diff --git a/conformance/tests/l1-rendering/vuln-scanner-scan-loop/check.py b/conformance/tests/l1-rendering/vuln-scanner-scan-loop/check.py new file mode 100644 index 0000000..f86ea2d --- /dev/null +++ b/conformance/tests/l1-rendering/vuln-scanner-scan-loop/check.py @@ -0,0 +1,266 @@ +#!/usr/bin/env python3 +"""R-VULN-02/03/05: scan loop semantics with a fake Trivy runner and a controllable clock.""" +import json +import logging +import os +import sys +import tempfile +import threading +from datetime import datetime, timedelta, timezone +from pathlib import Path + +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE.parents[3] / "platform/vuln-scanner/src")) +import main # noqa: E402 +import report # noqa: E402 +import trivy # noqa: E402 + +logging.disable(logging.CRITICAL) +FAILED = [] +A1 = "sha256:" + "a1" * 32 +D4 = "sha256:" + "d4" * 32 +E5 = "sha256:" + "e5" * 32 +NGINX = "docker.io/library/nginx@" + A1 +POSTGRES = "docker.io/library/postgres@" + D4 +PRIVATE = "ghcr.io/org/private@" + E5 +NIGHTLY = "ghcr.io/org/nightly:2.1" +CLUSTER = json.loads((HERE / "fixtures/cluster.json").read_text()) + + +def fixture(name): + return json.loads((HERE / "fixtures" / name).read_text()) + + +def check(name, condition): + print(("ok " if condition else "FAIL ") + name) + if not condition: + FAILED.append(name) + + +class Clock: + def __init__(self): + self.now = datetime(2026, 11, 9, 12, 0, tzinfo=timezone.utc) + + def __call__(self): + return self.now + + def advance(self, **delta): + self.now += timedelta(**delta) + + +class FakeClient: + def __init__(self, cluster): + self.cluster = cluster + self.fail = False + + def list_all(self, path): + if self.fail: + raise RuntimeError("api down") + for name, item in (("pods", "/api/v1/pods"), ("replicasets", "replicasets"), ("jobs", "/jobs"), + ("cronjobs", "cronjobs"), ("nodes", "/nodes"), ("vmis", "virtualmachineinstances")): + if path.endswith(item): + return self.cluster[name] + return [] + + +class FakeRunner: + """Recorded results, no Trivy binary.""" + + def __init__(self): + self.scans = [] + self.outcomes = { + NGINX: fixture("trivy-nginx.json"), POSTGRES: fixture("trivy-postgres.json"), + NIGHTLY: fixture("trivy-nightly.json"), + PRIVATE: trivy.classify_failure("ghcr.io", "FATAL: GET https://ghcr.io/: UNAUTHORIZED: authentication required"), + } + self.db_age_hours = 1 + self.db_fails = False + self.clock = None + + def download_db(self): + if self.db_fails: + raise trivy.ScanFailure("failed", "database update failed") + + def db_info(self): + return {"updatedAt": (self.clock() - timedelta(hours=self.db_age_hours)).strftime("%Y-%m-%dT%H:%M:%SZ"), "version": "2"} + + def scan(self, reference): + self.scans.append(reference) + outcome = self.outcomes[reference] + if isinstance(outcome, Exception): + raise outcome + return outcome + + def clean_cache(self): + pass + + +def engine(tmp, runner=None, clock=None, acc="[]\n", cluster=CLUSTER): + path = os.path.join(tmp, "acceptances.yaml") + if not os.path.exists(path): + Path(path).write_text(acc) + clock = clock or Clock() + runner = runner or FakeRunner() + runner.clock = clock + return main.Engine(tmp, FakeClient(cluster), runner, path, clock=clock, publish_seconds=0), runner, clock + + +def targets(eng): + return {t["id"]: t for t in eng.report["targets"]} + + +def findings(eng, target_id): + return [f for f in eng.report["findings"] if f["targetId"] == target_id] + + +def one_scan_per_digest(scans): + return len(scans) == len(set(scans)) + + +with tempfile.TemporaryDirectory() as tmp: + eng, runner, clock = engine(tmp) + eng.tick() + t = targets(eng) + check("two workloads sharing a digest produce exactly one scan", runner.scans.count(NGINX) == 1 and one_scan_per_digest(runner.scans)) + check("mutation: the digest-uniqueness evaluator rejects a double scan", not one_scan_per_digest(["a", "a"])) + check("the shared digest maps back to both workloads", + {(w["kind"], w["name"]) for w in t[NGINX]["workloads"]} == {("Deployment", "web"), ("Deployment", "api")}) + check("shared digest findings appear once per target", len(findings(eng, NGINX)) == 2) + check("digest target is scanned", t[NGINX]["state"] == "scanned") + check("idle CronJob template is by-tag after scanning, not scanned", t[NIGHTLY]["state"] == "by-tag") + check("node is not-covered", t["node/prod-1"]["state"] == "not-covered") + check("report is schema-valid", report.validate(eng.report) == []) + check("credential-required is its own state, not failed", t[PRIVATE]["state"] == "credential-required") + check("credential-required reason names the registry", "ghcr.io" in t[PRIVATE]["reason"]) + check("credential-required does not count toward failed_over_24h", report.failed_over_24h(eng.state, clock() + timedelta(hours=48)) == 0) + check("report is written to disk", json.loads(Path(tmp, "report.json").read_text())["generatedAt"]) + first_seen = {f["id"]: f["firstSeenAt"] for f in findings(eng, NGINX)} + + clock.advance(hours=21) + runner.outcomes[POSTGRES] = trivy.ScanFailure("failed", "pull error: manifest unknown") + eng.tick() + t = targets(eng) + check("a failed image leaves the run continuing", runner.scans.count(NGINX) == 2 and runner.scans.count(NIGHTLY) == 2) + check("the failed target is failed", t[POSTGRES]["state"] == "failed") + check("the failed target's reason is shown", "manifest unknown" in t[POSTGRES]["reason"]) + check("the failed target keeps its prior findings", [f["id"] for f in findings(eng, POSTGRES)] == ["CVE-2026-0010"]) + check("other targets stay scanned", t[NGINX]["state"] == "scanned") + check("first-seen survives a re-scan", {f["id"]: f["firstSeenAt"] for f in findings(eng, NGINX)} == first_seen) + check("run with one failed image is still complete (failure is per-target)", eng.report["run"]["status"] == "complete") + + clock.advance(hours=25) + check("failed target is counted over 24h after 25h", report.failed_over_24h(eng.state, clock()) == 1) + eng.publish() + check("a target whose last success aged past 24h is stale", targets(eng)[NGINX]["state"] == "stale") + check("stale target keeps findings", len(findings(eng, NGINX)) == 2) + + # Restart: a new Engine loads everything from disk. + eng2, runner2, clock2 = engine(tmp, runner=FakeRunner(), clock=clock) + check("report survives restart", eng2.report is not None and eng2.report_bytes() is not None) + check("last complete run survives restart", eng2.state["lastCompleteRunAt"] is not None) + eng2.publish() + check("first-seen survives restart", {f["id"]: f["firstSeenAt"] for f in findings(eng2, NGINX)} == first_seen) + +with tempfile.TemporaryDirectory() as tmp: + # Not-scanned is never zero findings: the first tick has not scanned anything yet. + eng, runner, clock = engine(tmp) + eng.refresh_inventory() + eng.publish() + t = targets(eng) + check("before scanning, digest targets are pending", t[NGINX]["state"] == "pending" and t[POSTGRES]["state"] == "pending") + check("run object exists and is degraded before the first run", eng.report["run"]["status"] == "degraded") + check("no fabricated findings before scanning", eng.report["findings"] == []) + check("zero findings report still counts every target", len(eng.report["targets"]) == 5) + clock.advance(hours=3) + eng.publish() + t = targets(eng) + check("pending beyond 2h becomes failed, never silently absent", t[NGINX]["state"] == "failed") + check("failed reason is the contract's text", t[NGINX]["reason"] == report.NOT_SCANNED_REASON) + check("a never-scanned failed target has lastScannedAt null", t[NGINX]["lastScannedAt"] is None) + clock.advance(hours=24) + check("never-scanned target counts as failed over 24h", report.failed_over_24h(eng.state, clock()) >= 1) + +with tempfile.TemporaryDirectory() as tmp: + # Database problems degrade the run but it still reports. + eng, runner, clock = engine(tmp) + runner.db_fails = True + eng.tick() + check("DB update failure degrades the run", eng.report["run"]["status"] == "degraded") + check("DB update failure still scans and reports", runner.scans and eng.report["findings"]) + check("a degraded run is not recorded as complete", eng.state["lastCompleteRunAt"] is None) +with tempfile.TemporaryDirectory() as tmp: + eng, runner, clock = engine(tmp) + runner.db_age_hours = 49 + eng.tick() + check("a DB older than 48h degrades the run", eng.report["run"]["status"] == "degraded") + runner.db_age_hours = 1 + clock.advance(hours=21) + eng.tick() + check("mutation: a fresh DB gives a complete run", eng.report["run"]["status"] == "complete" and eng.state["lastCompleteRunAt"]) + +with tempfile.TemporaryDirectory() as tmp: + # A target leaving the inventory leaves the report, with its findings. + eng, runner, clock = engine(tmp) + eng.tick() + check("postgres is present first", POSTGRES in targets(eng)) + gone = dict(CLUSTER) + gone["pods"] = [p for p in CLUSTER["pods"] if p["metadata"]["name"] != "db-0"] + eng.client.cluster = gone + clock.advance(minutes=15) + eng.tick() + check("a target leaving the inventory leaves the report", POSTGRES not in targets(eng)) + check("its findings go with it", not findings(eng, POSTGRES)) + +with tempfile.TemporaryDirectory() as tmp: + # Unexpected exceptions are contained at each level. + eng, runner, clock = engine(tmp) + runner.outcomes[POSTGRES] = RuntimeError("boom") + eng.tick() + check("an unexpected scan exception fails only that target", targets(eng)[POSTGRES]["state"] == "failed") + check("the run continued past it", runner.scans.count(NIGHTLY) == 1) + eng.client.fail = True + clock.advance(minutes=15) + eng.tick() + check("an inventory failure keeps the previous inventory", NGINX in targets(eng)) + check("an inventory failure degrades the report", eng.report["run"]["status"] == "degraded") + + calls = [] + + def crashing_tick(): + calls.append(1) + if len(calls) == 1: + raise RuntimeError("cycle crash") + eng.stop.set() + + eng.tick = crashing_tick + eng.tick_seconds = 0 + thread = threading.Thread(target=eng.run_forever, daemon=True) + thread.start() + thread.join(10) + check("a crash in one cycle is logged and the loop continues", len(calls) == 2 and not thread.is_alive()) + +# Recorded Trivy output through the normaliser: untrusted text. +parsed = trivy.parse_results({"SchemaVersion": 2, "Results": [{"Vulnerabilities": [ + {"VulnerabilityID": "CVE-1", "Severity": "weird", "PkgName": "p", "InstalledVersion": "1", "Title": "x\x07y", + "PrimaryURL": "javascript:alert(1)", "CVSS": {"nvd": {"V3Score": 11}}}, + {"VulnerabilityID": "CVE-2", "Severity": "HIGH", "PkgName": "q", "InstalledVersion": "1", "FixedVersion": "2", + "PrimaryURL": "https://example.test/x", "CVSS": {"nvd": {"V3Score": 7.5}}}, +]}]}) +check("unknown severity becomes UNKNOWN", parsed[0]["severity"] == "UNKNOWN") +check("control characters are stripped from upstream text", "\x07" not in parsed[0]["title"]) +check("non-http(s) advisory URL becomes empty", parsed[0]["url"] == "") +check("out-of-range CVSS is dropped", parsed[0]["cvss"] is None) +check("http(s) URL is kept", parsed[1]["url"] == "https://example.test/x") +check("a recorded report with no Results is zero findings", trivy.parse_results(fixture("trivy-empty.json")) == []) +try: + trivy.parse_results({"not": "a report"}) + check("an unrecognised document is a failure, never zero findings", False) +except trivy.ScanFailure as exc: + check("an unrecognised document is a failure, never zero findings", exc.kind == "failed") +failure = trivy.classify_failure("ghcr.io", "GET failed: 401 Unauthorized") +check("401 is credential-required", failure.kind == "credential-required" and "ghcr.io" in failure.reason) +check("permission denied on local disk is a plain failure", trivy.classify_failure("r", "open /cache/x: permission denied").kind == "failed") +check("a pull timeout is failed, not credential-required", trivy.classify_failure("r", "context deadline exceeded").kind == "failed") + +print("FAILED: " + str(len(FAILED)) if FAILED else "all checks passed") +sys.exit(1 if FAILED else 0) diff --git a/conformance/tests/l1-rendering/vuln-scanner-scan-loop/check.sh b/conformance/tests/l1-rendering/vuln-scanner-scan-loop/check.sh new file mode 100755 index 0000000..77679ca --- /dev/null +++ b/conformance/tests/l1-rendering/vuln-scanner-scan-loop/check.sh @@ -0,0 +1,7 @@ +#!/usr/bin/env bash +# Runs the fixture-driven assertions in check.py. See chainsaw-test.yaml for the requirement tags. +# What this does NOT prove: anything at runtime (real Trivy, real API, real browser). L2/L3 cover that. +set -euo pipefail +HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +export PYTHONDONTWRITEBYTECODE=1 +exec python3 "${HERE}/check.py" diff --git a/conformance/tests/l1-rendering/vuln-scanner-scan-loop/fixtures/cluster.json b/conformance/tests/l1-rendering/vuln-scanner-scan-loop/fixtures/cluster.json new file mode 100644 index 0000000..5d4e491 --- /dev/null +++ b/conformance/tests/l1-rendering/vuln-scanner-scan-loop/fixtures/cluster.json @@ -0,0 +1,198 @@ +{ + "cronjobs": [ + { + "metadata": { + "name": "nightly", + "namespace": "ops" + }, + "spec": { + "jobTemplate": { + "spec": { + "template": { + "spec": { + "containers": [ + { + "image": "ghcr.io/org/nightly:2.1", + "name": "job" + } + ] + } + } + } + } + } + } + ], + "jobs": [], + "nodes": [ + { + "metadata": { + "name": "prod-1" + } + } + ], + "pods": [ + { + "metadata": { + "name": "web-1", + "namespace": "apps", + "ownerReferences": [ + { + "controller": true, + "kind": "ReplicaSet", + "name": "web-5f" + } + ] + }, + "spec": { + "containers": [ + { + "image": "nginx:1.27", + "name": "app" + } + ] + }, + "status": { + "containerStatuses": [ + { + "imageID": "docker.io/library/nginx@sha256:a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1", + "name": "app" + } + ], + "phase": "Running" + } + }, + { + "metadata": { + "name": "api-1", + "namespace": "apps", + "ownerReferences": [ + { + "controller": true, + "kind": "ReplicaSet", + "name": "api-77" + } + ] + }, + "spec": { + "containers": [ + { + "image": "nginx:1.27", + "name": "server" + } + ] + }, + "status": { + "containerStatuses": [ + { + "imageID": "docker.io/library/nginx@sha256:a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1", + "name": "server" + } + ], + "phase": "Running" + } + }, + { + "metadata": { + "name": "db-0", + "namespace": "data", + "ownerReferences": [ + { + "controller": true, + "kind": "StatefulSet", + "name": "db" + } + ] + }, + "spec": { + "containers": [ + { + "image": "postgres:16", + "name": "postgres" + } + ] + }, + "status": { + "containerStatuses": [ + { + "imageID": "docker.io/library/postgres@sha256:d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4", + "name": "postgres" + } + ], + "phase": "Running" + } + }, + { + "metadata": { + "name": "priv-1", + "namespace": "apps", + "ownerReferences": [ + { + "controller": true, + "kind": "ReplicaSet", + "name": "priv-1a" + } + ] + }, + "spec": { + "containers": [ + { + "image": "ghcr.io/org/private:1", + "name": "priv" + } + ] + }, + "status": { + "containerStatuses": [ + { + "imageID": "ghcr.io/org/private@sha256:e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5", + "name": "priv" + } + ], + "phase": "Running" + } + } + ], + "replicasets": [ + { + "metadata": { + "name": "web-5f", + "namespace": "apps", + "ownerReferences": [ + { + "controller": true, + "kind": "Deployment", + "name": "web" + } + ] + } + }, + { + "metadata": { + "name": "api-77", + "namespace": "apps", + "ownerReferences": [ + { + "controller": true, + "kind": "Deployment", + "name": "api" + } + ] + } + }, + { + "metadata": { + "name": "priv-1a", + "namespace": "apps", + "ownerReferences": [ + { + "controller": true, + "kind": "Deployment", + "name": "priv" + } + ] + } + } + ], + "vmis": [] +} diff --git a/conformance/tests/l1-rendering/vuln-scanner-scan-loop/fixtures/trivy-empty.json b/conformance/tests/l1-rendering/vuln-scanner-scan-loop/fixtures/trivy-empty.json new file mode 100644 index 0000000..b81b42d --- /dev/null +++ b/conformance/tests/l1-rendering/vuln-scanner-scan-loop/fixtures/trivy-empty.json @@ -0,0 +1,4 @@ +{ + "ArtifactName": "x", + "SchemaVersion": 2 +} diff --git a/conformance/tests/l1-rendering/vuln-scanner-scan-loop/fixtures/trivy-nginx.json b/conformance/tests/l1-rendering/vuln-scanner-scan-loop/fixtures/trivy-nginx.json new file mode 100644 index 0000000..bbe13b4 --- /dev/null +++ b/conformance/tests/l1-rendering/vuln-scanner-scan-loop/fixtures/trivy-nginx.json @@ -0,0 +1,33 @@ +{ + "ArtifactName": "x", + "Results": [ + { + "Target": "t", + "Vulnerabilities": [ + { + "CVSS": { + "nvd": { + "V3Score": 9.8 + } + }, + "FixedVersion": "3.0.2", + "InstalledVersion": "1.0", + "PkgName": "openssl", + "PrimaryURL": "https://example.test/CVE-2026-0001", + "Severity": "CRITICAL", + "Title": "OpenSSL flaw", + "VulnerabilityID": "CVE-2026-0001" + }, + { + "InstalledVersion": "1.0", + "PkgName": "zlib", + "PrimaryURL": "https://example.test/CVE-2026-0002", + "Severity": "LOW", + "Title": "t", + "VulnerabilityID": "CVE-2026-0002" + } + ] + } + ], + "SchemaVersion": 2 +} diff --git a/conformance/tests/l1-rendering/vuln-scanner-scan-loop/fixtures/trivy-nightly.json b/conformance/tests/l1-rendering/vuln-scanner-scan-loop/fixtures/trivy-nightly.json new file mode 100644 index 0000000..8d37a56 --- /dev/null +++ b/conformance/tests/l1-rendering/vuln-scanner-scan-loop/fixtures/trivy-nightly.json @@ -0,0 +1,19 @@ +{ + "ArtifactName": "x", + "Results": [ + { + "Target": "t", + "Vulnerabilities": [ + { + "InstalledVersion": "1.0", + "PkgName": "curl", + "PrimaryURL": "https://example.test/CVE-2026-0020", + "Severity": "MEDIUM", + "Title": "t", + "VulnerabilityID": "CVE-2026-0020" + } + ] + } + ], + "SchemaVersion": 2 +} diff --git a/conformance/tests/l1-rendering/vuln-scanner-scan-loop/fixtures/trivy-postgres.json b/conformance/tests/l1-rendering/vuln-scanner-scan-loop/fixtures/trivy-postgres.json new file mode 100644 index 0000000..efc3e3b --- /dev/null +++ b/conformance/tests/l1-rendering/vuln-scanner-scan-loop/fixtures/trivy-postgres.json @@ -0,0 +1,25 @@ +{ + "ArtifactName": "x", + "Results": [ + { + "Target": "t", + "Vulnerabilities": [ + { + "CVSS": { + "nvd": { + "V3Score": 7.5 + } + }, + "FixedVersion": "16.2", + "InstalledVersion": "1.0", + "PkgName": "libpq", + "PrimaryURL": "https://example.test/CVE-2026-0010", + "Severity": "HIGH", + "Title": "libpq flaw", + "VulnerabilityID": "CVE-2026-0010" + } + ] + } + ], + "SchemaVersion": 2 +} diff --git a/platform/vuln-scanner/src/acceptances.py b/platform/vuln-scanner/src/acceptances.py new file mode 100644 index 0000000..b8f4cd3 --- /dev/null +++ b/platform/vuln-scanner/src/acceptances.py @@ -0,0 +1,141 @@ +"""R-VULN-08 acceptances: a strict parser for exactly the spec's flat list, no YAML library. + + - id: CVE-2026-00000 + image: registry/repository # exact match; omit only for platform-wide + reason: not reachable - feature disabled + expires: 2027-02-01 # mandatory, at most 180 days out + +Fail-safe: a malformed entry is ignored and reported, so it can never hide a finding. Anything the +parser cannot place (bad indentation, a stray line) invalidates the whole file for the same reason. +""" +import re +import sys +from datetime import date, timedelta + +from state import utcnow + +MAX_DAYS = 180 +_KEYS = ("id", "image", "reason", "expires") +_ENTRY = re.compile(r"^- ([a-z]+):(?: (.*))?$") +_FIELD = re.compile(r"^ ([a-z]+):(?: (.*))?$") +_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]*$") +_DATE = re.compile(r"^\d{4}-\d{2}-\d{2}$") +_FORBIDDEN_START = tuple("[{&*!|>%@`") + + +def _scalar(raw): + """(value, None) or (None, error) for an unquoted or simply-quoted YAML scalar.""" + value = (raw or "").strip() + if value[:1] in ("'", '"'): + quote = value[0] + end = value.find(quote, 1) + if end == -1 or "\\" in value[:end]: + return None, "bad quoting" + rest = value[end + 1:].strip() + if rest and not rest.startswith("#"): + return None, "text after the closing quote" + return value[1:end], None + if value.startswith(_FORBIDDEN_START): + return None, "only plain scalars are allowed" + comment = re.search(r"\s#", value) + if comment: + value = value[:comment.start()].rstrip() + return value, None + + +def parse_text(source, today=None): + """(valid entries, error strings). Entries carry id, reason, expires (date) and optional image.""" + today = today or utcnow().date() + errors = [] + raw_entries = [] + current = None + for lineno, line in enumerate(source.splitlines(), 1): + if not line.strip() or line.lstrip().startswith("#"): + continue + entry = _ENTRY.match(line) + field = _FIELD.match(line) + if line.strip().split("#", 1)[0].strip() == "[]" and not raw_entries and current is None: + continue + if entry: + current = {"line": lineno, "fields": {}, "errors": []} + raw_entries.append(current) + key, raw = entry.group(1), entry.group(2) + elif field and current is not None: + key, raw = field.group(1), field.group(2) + else: + return [], ["line " + str(lineno) + ": not a flat acceptance entry; the whole file is ignored"] + value, problem = _scalar(raw) + if key not in _KEYS: + current["errors"].append("unknown key " + key) + elif problem: + current["errors"].append(key + ": " + problem) + elif key in current["fields"]: + current["errors"].append("duplicate key " + key) + else: + current["fields"][key] = value + entries = [] + for raw in raw_entries: + problems = list(raw["errors"]) + _validate(raw["fields"], today) + if problems: + errors.append("entry at line " + str(raw["line"]) + ": " + "; ".join(problems)) + continue + fields = raw["fields"] + entries.append({"id": fields["id"], "image": fields.get("image") or None, "reason": fields["reason"], + "expires": date.fromisoformat(fields["expires"])}) + return entries, errors + + +def _validate(fields, today): + problems = [] + if not _ID.match(fields.get("id") or ""): + problems.append("id is missing or malformed") + if not fields.get("reason"): + problems.append("reason is mandatory") + if "image" in fields and not fields["image"]: + problems.append("image is empty") + expires = fields.get("expires") or "" + if not _DATE.match(expires): + problems.append("expires is mandatory and must be YYYY-MM-DD") + return problems + try: + when = date.fromisoformat(expires) + except ValueError: + problems.append("expires is not a real date") + return problems + if when > today + timedelta(days=MAX_DAYS): + problems.append("expires is more than " + str(MAX_DAYS) + " days out") + return problems + + +def load(path, today=None): + """Entries and errors from a file. An unreadable file is an error, never silently empty.""" + try: + with open(path, encoding="utf-8") as handle: + source = handle.read(1024 * 1024 + 1) + except (OSError, UnicodeDecodeError) as exc: + return [], ["acceptances file unreadable: " + type(exc).__name__] + if len(source) > 1024 * 1024: + return [], ["acceptances file exceeds 1 MiB"] + return parse_text(source, today) + + +def match(entries, finding_id, image, today=None): + """The unexpired acceptance covering a finding, if any. Expiry day itself still counts.""" + today = today or utcnow().date() + best = None + for entry in entries: + if entry["id"] != finding_id or entry["expires"] < today: + continue + if entry["image"] is not None and entry["image"] != image: + continue + if best is None or entry["expires"] > best["expires"]: + best = entry + return best + + +if __name__ == "__main__": + found, problems = load(sys.argv[1]) + for problem in problems: + print("INVALID: " + problem, file=sys.stderr) + print(str(len(found)) + " valid entries") + sys.exit(1 if problems else 0) diff --git a/platform/vuln-scanner/src/inventory.py b/platform/vuln-scanner/src/inventory.py new file mode 100644 index 0000000..5751d55 --- /dev/null +++ b/platform/vuln-scanner/src/inventory.py @@ -0,0 +1,248 @@ +"""Inventory from the Kubernetes API: what is actually running, and which workloads run it.""" +import json +import logging +import os +import re +import ssl +import urllib.error +import urllib.parse +import urllib.request + +log = logging.getLogger("vuln-scanner.inventory") + +_ID_PREFIXES = ("docker-pullable://", "docker://", "containerd://", "cri-o://") +_DIGEST = re.compile(r"sha256:[0-9a-f]{64}") +_DOCKER_ALIASES = {"index.docker.io", "registry-1.docker.io", "docker.io"} +_CONTAINER_FIELDS = ( + ("containers", "containerStatuses"), + ("initContainers", "initContainerStatuses"), + ("ephemeralContainers", "ephemeralContainerStatuses"), +) +_VMI_PATH = "/apis/kubevirt.io/v1/virtualmachineinstances" +PATHS = { + "pods": "/api/v1/pods", + "replicasets": "/apis/apps/v1/replicasets", + "jobs": "/apis/batch/v1/jobs", + "cronjobs": "/apis/batch/v1/cronjobs", + "nodes": "/api/v1/nodes", +} + + +class InventoryError(Exception): + """The API could not be listed; the previous inventory stands.""" + + +class NotFound(InventoryError): + pass + + +class KubeClient: + """Paginated GET list against the API using the pod's service account.""" + + def __init__(self, base_url=None, token_file=None, ca_file=None): + sa = "/var/run/secrets/kubernetes.io/serviceaccount" + if base_url is None: + host = os.environ.get("KUBERNETES_SERVICE_HOST", "kubernetes.default.svc") + port = os.environ.get("KUBERNETES_SERVICE_PORT", "443") + host = "[" + host + "]" if ":" in host else host + base_url = "https://" + host + ":" + port + self.base_url = base_url.rstrip("/") + self.token_file = token_file or os.path.join(sa, "token") + self.ca_file = ca_file or os.path.join(sa, "ca.crt") + + def _get_json(self, path, params): + url = self.base_url + path + "?" + urllib.parse.urlencode(params) + with open(self.token_file, encoding="utf-8") as handle: # projected tokens rotate + token = handle.read().strip() + request = urllib.request.Request(url, headers={"Authorization": "Bearer " + token}) + context = ssl.create_default_context(cafile=self.ca_file) if url.startswith("https") else None + try: + with urllib.request.urlopen(request, timeout=30, context=context) as response: + return json.loads(response.read()) + except urllib.error.HTTPError as exc: + if exc.code == 404: + raise NotFound(path) from None + raise InventoryError(path + ": HTTP " + str(exc.code)) from None + except (urllib.error.URLError, OSError, ValueError) as exc: + raise InventoryError(path + ": " + type(exc).__name__) from None + + def list_all(self, path): + items = [] + params = {"limit": "500"} + while True: + page = self._get_json(path, params) + items.extend(page.get("items") or []) + token = (page.get("metadata") or {}).get("continue") + if not token: + return items + params = {"limit": "500", "continue": token} + + +def collect(client): + """Every list the inventory needs. A VMI API that 404s means no KubeVirt, hence no VMs.""" + lists = {name: client.list_all(path) for name, path in PATHS.items()} + try: + lists["vmis"] = client.list_all(_VMI_PATH) + except NotFound: + lists["vmis"] = [] + return lists + + +def normalise_reference(reference): + """(registry/repository, tag or None, digest or None) for an image reference.""" + text = reference.strip() + for prefix in _ID_PREFIXES: + if text.startswith(prefix): + text = text[len(prefix):] + digest = None + if "@" in text: + text, _, raw = text.partition("@") + digest = raw if _DIGEST.fullmatch(raw) else None + tag = None + tail = text.rsplit("/", 1)[-1] + if ":" in tail: + text, _, tag = text.rpartition(":") + first, slash, rest = text.partition("/") + if slash and ("." in first or ":" in first or first == "localhost"): + registry, repository = first.lower(), rest + else: + registry, repository = "docker.io", text + if registry in _DOCKER_ALIASES: + registry = "docker.io" + if "/" not in repository: + repository = "library/" + repository + return registry + "/" + repository, tag, digest + + +def digest_from_image_id(image_id): + """The repo digest in a container status imageID, or None. + + A bare `sha256:...` with no repository is a local config id, not a pullable digest, so it is + treated as absent (the target is then by-tag) rather than scanned as something it is not. + """ + text = image_id or "" + for prefix in _ID_PREFIXES: + if text.startswith(prefix): + text = text[len(prefix):] + if "@" not in text: + return None + candidate = text.rsplit("@", 1)[1] + return candidate if _DIGEST.fullmatch(candidate) else None + + +def _controller(obj): + references = (obj.get("metadata") or {}).get("ownerReferences") or [] + for ref in references: + if ref.get("controller"): + return ref + return references[0] if references else None + + +def resolve_owner(pod, replicasets, jobs): + """The top-level workload (kind, name) a pod belongs to.""" + meta = pod.get("metadata") or {} + namespace = meta.get("namespace", "") + owner = _controller(pod) + if owner is None: + return "Pod", meta.get("name", "") + kind, name = owner.get("kind", ""), owner.get("name", "") + index = {"ReplicaSet": replicasets, "Job": jobs}.get(kind) + if index is not None: + parent = index.get((namespace, name)) + upstream = _controller(parent) if parent else None + if upstream is not None: + return upstream.get("kind", kind), upstream.get("name", name) + return kind, name + + +def _index(items): + return {((i.get("metadata") or {}).get("namespace", ""), (i.get("metadata") or {}).get("name", "")): i + for i in items} + + +class _Builder: + def __init__(self): + self.targets = {} + + def add(self, target_id, image, digest, ref, workload, container): + target = self.targets.get(target_id) + if target is None: + target = self.targets[target_id] = { + "id": target_id, "type": "image", "image": image, "digest": digest, "ref": ref, + "reason": None, "workloads": {}, + } + if digest and not target["digest"]: + target["digest"] = digest + target["workloads"].setdefault(workload, set()).add(container) + + def add_image(self, spec_image, image_id, workload, container): + image, tag, spec_digest = normalise_reference(spec_image) + digest = digest_from_image_id(image_id) + if digest: + self.add(image + "@" + digest, image, digest, image + "@" + digest, workload, container) + else: + ref = image + "@" + spec_digest if spec_digest else image + ":" + (tag or "latest") + target_id = image + ":" + tag if tag else (ref if spec_digest else image + ":latest") + self.add(target_id, image, None, ref, workload, container) + + def add_uncovered(self, target_id, kind, reason, namespace, name, node_kind): + self.targets[target_id] = { + "id": target_id, "type": kind, "image": "", "digest": None, "ref": None, "reason": reason, + "workloads": {(namespace, node_kind, name): set()}, + } + + def finish(self): + result = [] + for target in sorted(self.targets.values(), key=lambda t: t["id"]): + target["workloads"] = [ + {"namespace": ns, "kind": kind, "name": name, "containers": sorted(containers)} + for (ns, kind, name), containers in sorted(target["workloads"].items()) + ] + result.append(target) + return result + + +def build(lists): + """Resolve raw API lists into target dicts.""" + replicasets = _index(lists.get("replicasets") or []) + jobs = _index(lists.get("jobs") or []) + builder = _Builder() + running = set() + for pod in lists.get("pods") or []: + status = pod.get("status") or {} + # A Succeeded/Failed pod runs nothing; counting it would let a stale evicted pod pin a digest. + if status.get("phase") in ("Succeeded", "Failed"): + continue + namespace = (pod.get("metadata") or {}).get("namespace", "") + kind, name = resolve_owner(pod, replicasets, jobs) + workload = (namespace, kind, name) + if kind == "CronJob": + running.add((namespace, name)) + spec = pod.get("spec") or {} + for spec_field, status_field in _CONTAINER_FIELDS: + statuses = {s.get("name"): s for s in status.get(status_field) or []} + for container in spec.get(spec_field) or []: + image_id = (statuses.get(container.get("name")) or {}).get("imageID", "") + builder.add_image(container.get("image", ""), image_id, workload, container.get("name", "")) + for cronjob in lists.get("cronjobs") or []: + meta = cronjob.get("metadata") or {} + namespace, name = meta.get("namespace", ""), meta.get("name", "") + if (namespace, name) in running: + continue + job_spec = ((cronjob.get("spec") or {}).get("jobTemplate") or {}).get("spec") or {} + pod_spec = (job_spec.get("template") or {}).get("spec") or {} + for field in ("containers", "initContainers"): + for container in pod_spec.get(field) or []: + builder.add_image(container.get("image", ""), "", (namespace, "CronJob", name), container.get("name", "")) + for vmi in lists.get("vmis") or []: + meta = vmi.get("metadata") or {} + namespace, name = meta.get("namespace", ""), meta.get("name", "") + builder.add_uncovered("vm/" + namespace + "/" + name, "vm", + "guest OS packages are not assessed by this scanner", + namespace, name, "VirtualMachineInstance") + for node in lists.get("nodes") or []: + name = (node.get("metadata") or {}).get("name", "") + builder.add_uncovered("node/" + name, "node", + "node OS packages and the k3s binary are not assessed by this scanner", + "", name, "Node") + return builder.finish() diff --git a/platform/vuln-scanner/src/main.py b/platform/vuln-scanner/src/main.py new file mode 100644 index 0000000..6e0f3e3 --- /dev/null +++ b/platform/vuln-scanner/src/main.py @@ -0,0 +1,276 @@ +"""Scanner entrypoint and the scan loop. Import has no side effects; main() runs the service.""" +import collections +import logging +import os +import signal +import sys +import threading +import time +from datetime import timedelta + +import acceptances +import inventory +import metrics +import page +import report as report_mod +import server +import state as st +import trivy +from state import fmt_ts, parse_ts + +log = logging.getLogger("vuln-scanner") + +FULL_RUN_INTERVAL = timedelta(hours=20) +DB_MAX_AGE = timedelta(hours=48) +DB_REFRESH_AGE = timedelta(hours=24) + + +class Engine: + def __init__(self, data_dir, client, runner, acceptances_path, clock=st.utcnow, tick_seconds=900, publish_seconds=120): + self.store = st.Store(data_dir) + self.data_dir = data_dir + self.client = client + self.runner = runner + self.acceptances_path = acceptances_path + self.clock = clock + self.tick_seconds = tick_seconds + self.publish_seconds = publish_seconds + self.lock = threading.RLock() + self.stop = threading.Event() + self.state = self.store.load() + self.report = report_mod.load_published(data_dir) + self.report_json = report_mod.serialise(self.report) if self.report else None + self.loop_thread = None + self._last_publish = 0.0 + self._last_inventory = None + + # --- views used by the HTTP servers; none of them waits for a scan -------------------------- + def report_bytes(self): + return self.report_json + + def render_page(self, query): + return page.render(self.report, query, self.clock()) + + def metrics_text(self): + with self.lock: + return metrics.render(self.report, self.state, self.clock()) + + def healthy(self): + return self.loop_thread is None or self.loop_thread.is_alive() + + def record_triage(self, document): + with self.lock: + self.state["remediation"] = document["remediation"] + self.state["triageHeartbeatAt"] = fmt_ts(self.clock()) # server receipt time, per contract + self.store.save(self.state) + self.publish() + + # --- persistence ----------------------------------------------------------------------------- + def publish(self): + with self.lock: + now = self.clock() + entries, errors = acceptances.load(self.acceptances_path, now.date()) + for problem in errors: + log.error("invalid acceptance ignored: %s", problem) + built = report_mod.build(self.state, entries, errors, now) + if report_mod.publish(built, self.data_dir): + self.report = built + self.report_json = report_mod.serialise(built) + self._last_publish = time.monotonic() + + # --- inventory ------------------------------------------------------------------------------- + def refresh_inventory(self): + """Install a fresh inventory. A failure keeps the previous one and degrades the report.""" + now = self.clock() + self._last_inventory = time.monotonic() + try: + targets = inventory.build(inventory.collect(self.client)) + except Exception as exc: # pylint: disable=broad-except + log.exception("inventory failed; keeping the previous inventory") + with self.lock: + self.state["inventoryError"] = type(exc).__name__ + self.store.save(self.state) + return False + with self.lock: + st.merge_inventory(self.state, targets, now) + self.store.save(self.state) + return True + + def _units(self): + """{scan key: reference}. Two targets sharing a digest are one unit, scanned once.""" + units = {} + for target in sorted(self.state["inventory"], key=lambda t: t["id"]): + key = st.scan_key(target) + if key is not None and key not in units: + units[key] = target["image"] + "@" + target["digest"] if target["digest"] else target["ref"] + return units + + def _never_attempted(self): + units = self._units() + return [k for k in units if self.state["scans"][k]["lastAttemptAt"] is None] + + def _full_run_due(self): + """Keyed on the last ATTEMPT, not the last success: a permanently failing image would + otherwise look 'older than 20h' forever and restart a full run every tick.""" + if self.state["run"] is None: + return True + now = self.clock() + for key in self._units(): + attempted = parse_ts(self.state["scans"][key]["lastAttemptAt"]) + if attempted is not None and now - attempted >= FULL_RUN_INTERVAL: + return True + return False + + # --- scanning -------------------------------------------------------------------------------- + def _refresh_database(self, always): + """True when a usable database is present. Failures are logged, never raised.""" + info = self._db_info() + if always or info is None or self._db_age(info) >= DB_REFRESH_AGE: + try: + self.runner.download_db() + except Exception: # pylint: disable=broad-except + log.exception("vulnerability database update failed") + return False + info = self._db_info() + if info is None: + return False + with self.lock: + self.state["database"] = info + self.store.save(self.state) + return True + + def _db_info(self): + try: + info = self.runner.db_info() + except Exception: # pylint: disable=broad-except + log.exception("could not read vulnerability database information") + return None + return dict(info) if info else None + + def _db_age(self, info): + moment = parse_ts(info.get("updatedAt")) + return timedelta.max if moment is None else self.clock() - moment + + def _scan_unit(self, key, reference): + try: + found = trivy.parse_results(self.runner.scan(reference)) + except trivy.ScanFailure as exc: + self._record_failure(key, exc.kind, exc.reason) + except Exception as exc: # pylint: disable=broad-except + log.exception("scan of %s failed unexpectedly", reference) + self._record_failure(key, "failed", "scan error: " + type(exc).__name__) + else: + self._record_success(key, found) + try: + self.runner.clean_cache() + except Exception: # pylint: disable=broad-except + log.exception("scan cache cleanup failed") + + def _record_failure(self, key, kind, reason): + now = fmt_ts(self.clock()) + with self.lock: + record = self.state["scans"].get(key) + if record is None: + return + record["lastAttemptAt"] = now + record["outcome"] = kind + record["reason"] = reason + record["failedSince"] = (record["failedSince"] or now) if kind == "failed" else None + self.store.save(self.state) + + def _record_success(self, key, found): + now = fmt_ts(self.clock()) + with self.lock: + record = self.state["scans"].get(key) + if record is None: + return + record["lastAttemptAt"] = now + record["lastSuccessAt"] = now + record["outcome"] = "ok" + record["reason"] = None + record["failedSince"] = None + record["findings"] = found + record["firstSeen"] = {f["id"]: record["firstSeen"].get(f["id"], now) for f in found} + self.store.save(self.state) + + def _scan_queue(self, keys): + """Scan sequentially, keeping the inventory and report fresh while a long run proceeds.""" + queue = collections.deque(keys) + while queue and not self.stop.is_set(): + if self._last_inventory is None or time.monotonic() - self._last_inventory >= self.tick_seconds: + if self.refresh_inventory(): + queue.extendleft(k for k in self._never_attempted() if k not in queue) + key = queue.popleft() + reference = self._units().get(key) + if reference is None: + continue # it left the inventory while queued + try: + self._scan_unit(key, reference) + except Exception: # pylint: disable=broad-except + log.exception("scan bookkeeping failed for %s", key) + if time.monotonic() - self._last_publish >= self.publish_seconds: + self.publish() + + def run_full(self): + started = self.clock() + log.info("full run starting") + database_ok = self._refresh_database(always=True) + self._scan_queue(sorted(self._units())) + info = self.state["database"] + now = self.clock() + degraded = not database_ok or info is None or self._db_age(info) > DB_MAX_AGE + _, errors = acceptances.load(self.acceptances_path, now.date()) + if errors or self.state.get("inventoryError") or self.state.get("loadError"): + degraded = True + status = "degraded" if degraded else "complete" + with self.lock: + self.state["run"] = {"startedAt": fmt_ts(started), "finishedAt": fmt_ts(now), "status": status} + if not degraded: + self.state["lastCompleteRunAt"] = fmt_ts(now) + self.store.save(self.state) + log.info("full run finished: %s", status) + + def tick(self): + self.refresh_inventory() + self.publish() + if self._full_run_due(): + self.run_full() + else: + never = self._never_attempted() + if never: + self._refresh_database(always=False) + self._scan_queue(never) + self.publish() + + def run_forever(self): + while not self.stop.is_set(): + try: + self.tick() + except Exception: # pylint: disable=broad-except + log.exception("scan cycle crashed; continuing with the next cycle") + self.stop.wait(self.tick_seconds) + + def start(self): + self.loop_thread = threading.Thread(target=self.run_forever, name="scan-loop", daemon=True) + self.loop_thread.start() + + +def main(): + logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s") + data_dir = os.environ.get("VULN_DATA_DIR", "/data") + cache_dir = os.environ.get("VULN_CACHE_DIR", "/cache") + tools_dir = os.environ.get("VULN_TOOLS_DIR", "/tools") + trivy_bin = os.environ.get("TRIVY_BIN", os.path.join(tools_dir, "trivy")) + runner = trivy.Runner(trivy_bin, cache_dir, int(os.environ.get("VULN_SCAN_TIMEOUT_SECONDS", "600"))) + engine = Engine(data_dir, inventory.KubeClient(), runner, os.environ.get("VULN_ACCEPTANCES", "/config/acceptances.yaml")) + server.start_all(engine, ("127.0.0.1", 8080), ("0.0.0.0", 8081), ("0.0.0.0", 9090)) + finished = threading.Event() + signal.signal(signal.SIGTERM, lambda *_: finished.set()) + signal.signal(signal.SIGINT, lambda *_: finished.set()) + engine.start() + finished.wait() + engine.stop.set() + + +if __name__ == "__main__": + main() diff --git a/platform/vuln-scanner/src/metrics.py b/platform/vuln-scanner/src/metrics.py new file mode 100644 index 0000000..c8fe7c6 --- /dev/null +++ b/platform/vuln-scanner/src/metrics.py @@ -0,0 +1,44 @@ +"""Prometheus text exposition: aggregates only (R-VULN-06). The label allowlist is state, severity, fixable.""" +from datetime import timedelta + +import report as report_mod +from state import parse_ts + +_CRITICAL_GRACE = timedelta(hours=72) +_SEVERITIES = ("CRITICAL", "HIGH", "MEDIUM", "LOW", "UNKNOWN") + + +def _sample(lines, name, value, labels=""): + lines.append(name + labels + " " + format(value, ".0f" if float(value).is_integer() else "g")) + + +def render(report, state, now): + lines = [] + if report is not None: + _sample(lines, "vuln_scanner_report_generated_timestamp_seconds", parse_ts(report["generatedAt"]).timestamp()) + _sample(lines, "vuln_scanner_database_updated_timestamp_seconds", parse_ts(report["database"]["updatedAt"]).timestamp()) + counts = {name: 0 for name in report_mod.STATES} + for target in report["targets"]: + counts[target["state"]] += 1 + for name in report_mod.STATES: + _sample(lines, "vuln_scanner_targets", counts[name], '{state="' + name + '"}') + findings = {(sev, fix): 0 for sev in _SEVERITIES for fix in ("true", "false")} + overdue = 0 + for finding in report["findings"]: + if finding["status"] == "accepted": + continue + findings[(finding["severity"], "true" if finding["fixedVersion"] else "false")] += 1 + if finding["severity"] == "CRITICAL" and finding["fixedVersion"]: + first_seen = parse_ts(finding["firstSeenAt"]) + if first_seen and now - first_seen > _CRITICAL_GRACE: + overdue += 1 + for (severity, fixable), value in sorted(findings.items()): + _sample(lines, "vuln_scanner_findings", value, '{severity="' + severity + '",fixable="' + fixable + '"}') + _sample(lines, "vuln_scanner_critical_fix_overdue", overdue) + _sample(lines, "vuln_scanner_failed_targets_over_24h", report_mod.failed_over_24h(state, now)) + for name, key in (("vuln_scanner_last_complete_run_timestamp_seconds", "lastCompleteRunAt"), + ("vuln_scanner_triage_heartbeat_timestamp_seconds", "triageHeartbeatAt")): + value = parse_ts(state.get(key)) + if value is not None: + _sample(lines, name, value.timestamp()) + return "\n".join(lines) + "\n" diff --git a/platform/vuln-scanner/src/page.py b/platform/vuln-scanner/src/page.py new file mode 100644 index 0000000..a2cd1a3 --- /dev/null +++ b/platform/vuln-scanner/src/page.py @@ -0,0 +1,482 @@ +"""The CVE page: server-rendered HTML with no script at all. + +Every string that reaches the document goes through e(), the single escape path; links additionally +go through safe_url(). The style block is the only inline resource and is allowed by hash. +""" +import base64 +import hashlib +import html +import math +from datetime import timedelta +from urllib.parse import parse_qs, urlencode + +import report as report_mod +import trivy +from state import parse_ts + +REPORT_STALE = timedelta(hours=26) +DB_STALE = timedelta(hours=48) +SEVERITIES = ("CRITICAL", "HIGH", "MEDIUM", "LOW", "UNKNOWN") +SEV_RANK = {name: i for i, name in enumerate(SEVERITIES)} +STATUS_RANK = {"fix-available": 0, "open": 1, "accepted": 2} +STATE_RANK = {name: i for i, name in enumerate(report_mod.STATES)} +PAGE_SIZES = (25, 50, 100, 200, 500) +DEFAULT_SIZE = 100 +_MAX_PARAM = 200 +_EXPANSION_LIMIT = 200 # per-target CVE list in the inventory view; the remainder is linked + +STYLE = """ +body{font:14px/1.45 system-ui,sans-serif;margin:0;background:#fff;color:#1a1a1a} +main{padding:0 16px 32px} +header{padding:12px 16px;background:#f4f4f6;border-bottom:1px solid #ccc} +h1{font-size:18px;margin:0 0 8px} +nav a{margin-right:12px} +.warn{background:#fff0d6;border:2px solid #b35900;padding:8px 12px;margin:8px 0;font-weight:600} +.status{font-weight:700} +.tiles span{display:inline-block;margin:2px 10px 2px 0} +table{border-collapse:collapse;width:100%;margin-top:8px} +th,td{border-bottom:1px solid #ddd;padding:4px 6px;text-align:left;vertical-align:top} +th a{text-decoration:none} +.sev-CRITICAL{color:#a00;font-weight:700} +.sev-HIGH{color:#b35900;font-weight:700} +form{margin:8px 0} +form label{margin-right:8px} +code{word-break:break-all} +@media (prefers-color-scheme:dark){body{background:#16181c;color:#e6e6e6} +header{background:#22252b;border-color:#3a3d44} +th,td{border-color:#33363c} +a{color:#7fb4ff} +.warn{background:#3d2a12;border-color:#c98422}} +""" +STYLE_HASH = base64.b64encode(hashlib.sha256(STYLE.encode("utf-8")).digest()).decode("ascii") +CSP = ("default-src 'none'; style-src 'sha256-" + STYLE_HASH + "'; base-uri 'none'; " + "form-action 'self'; frame-ancestors 'none'") +SECURITY_HEADERS = { + "Content-Security-Policy": CSP, + "X-Content-Type-Options": "nosniff", + "Content-Type": "text/html; charset=utf-8", + "Cache-Control": "no-store", +} + + +def e(value): + """The one escape path for anything placed in the document.""" + return "" if value is None else html.escape(str(value), quote=True) + + +def safe_url(value): + return trivy.safe_url(value) if isinstance(value, str) else "" + + +def _link(url, label): + url = safe_url(url) + return '' + e(label) + "" if url else e(label) + + +def _parse(query): + raw = parse_qs(query or "", keep_blank_values=False) + params = {k: v[0][:_MAX_PARAM] for k, v in raw.items() if v} + view = params.get("view", "cves") + params["view"] = view if view in ("cves", "workloads", "inventory") else "cves" + try: + params["page"] = max(1, int(params.get("page", "1"))) + except ValueError: + params["page"] = 1 + try: + size = int(params.get("size", str(DEFAULT_SIZE))) + except ValueError: + size = DEFAULT_SIZE + params["size"] = size if size in PAGE_SIZES else DEFAULT_SIZE + return params + + +def _qs(params, **overrides): + merged = dict(params) + merged.update(overrides) + pairs = sorted((k, str(v)) for k, v in merged.items() if v not in (None, "")) + return "?" + urlencode(pairs) + + +def _age(moment, now): + if moment is None: + return "unknown" + seconds = int((now - moment).total_seconds()) + if seconds < 0: + return "in the future" + hours, remainder = divmod(seconds, 3600) + minutes = remainder // 60 + return str(hours) + "h " + str(minutes) + "m ago" + + +def _finding_text(finding, target): + parts = [finding["id"], finding["package"], finding["title"], target["id"], target["image"]] + for workload in target["workloads"]: + parts.extend((workload["namespace"], workload["name"], *workload["containers"])) + return " ".join(parts).lower() + + +def _match_workload(target, params): + namespace, workload = params.get("namespace"), params.get("workload") + if not namespace and not workload: + return True + for item in target["workloads"]: + if namespace and item["namespace"] != namespace: + continue + if workload and workload not in (item["name"], item["kind"] + "/" + item["name"]): + continue + return True + return False + + +def _match_finding(finding, target, params): + if params.get("severity") and finding["severity"] != params["severity"].upper(): + return False + if params.get("status") and finding["status"] != params["status"]: + return False + fix = params.get("fix") + if fix == "yes" and not finding["fixedVersion"]: + return False + if fix == "no" and finding["fixedVersion"]: + return False + if not _match_workload(target, params): + return False + text = params.get("q", "").lower() + return not text or text in _finding_text(finding, target) + + +def _paginate(rows, params): + total = len(rows) + size = params["size"] + pages = max(1, math.ceil(total / size)) + page = min(params["page"], pages) + start = (page - 1) * size + return rows[start:start + size], total, page, pages, start + + +def _pager(params, total, page, pages, start, shown, noun): + first = start + 1 if total else 0 + summary = "Showing " + str(first) + "–" + str(start + shown) + " of " + str(total) + " " + noun \ + + " (page " + str(page) + " of " + str(pages) + ")" + links = [] + if page > 1: + links.append('previous') + if page < pages: + links.append('next') + return "

" + e(summary) + " " + " ".join(links) + "

" + + +def _th(params, label, key): + direction = "desc" if params.get("sort") == key and params.get("dir", "asc") == "asc" else "asc" + return '' + e(label) + "" + + +def _table(head, rows, empty): + if not rows: + return "

" + e(empty) + "

" + return "" + "".join(head) + "" + "".join(rows) + "
" + + +def _sev(severity): + return '' + e(severity) + "" + + +def _workload_lines(target): + items = [] + for workload in target["workloads"]: + items.append("
  • " + e(workload["namespace"]) + " / " + e(workload["kind"]) + " / " + e(workload["name"]) + + " — containers: " + e(", ".join(workload["containers"]) or "none") + "
    • ") + return "" + + +def _remediation_cell(images, remediation): + if not images: + return "" + out = [] + for image in sorted(images): + entry = remediation.get(image) + if entry is None: + out.append(e(image) + ": no data") + elif entry["state"] == "pr-open" and safe_url(entry.get("url")): + out.append(e(image) + ": " + _link(entry["url"], "pr-open")) + else: + out.append(e(image) + ": " + e(entry["state"])) + return "
    ".join(out) + + +# --- views ------------------------------------------------------------------------------------------ + +def _cves(report, targets, params, remediation): + groups = {} + for finding in report["findings"]: + target = targets[finding["targetId"]] + if not _match_finding(finding, target, params): + continue + group = groups.setdefault(finding["id"], { + "id": finding["id"], "severity": finding["severity"], "cvss": None, "packages": set(), "fixed": set(), + "firstSeen": finding["firstSeenAt"], "statuses": set(), "url": "", "title": "", "items": [], + }) + if SEV_RANK[finding["severity"]] < SEV_RANK[group["severity"]]: + group["severity"] = finding["severity"] + if finding["cvss"] is not None and (group["cvss"] is None or finding["cvss"] > group["cvss"]): + group["cvss"] = finding["cvss"] + group["packages"].add(finding["package"]) + if finding["fixedVersion"]: + group["fixed"].add(finding["fixedVersion"]) + group["firstSeen"] = min(group["firstSeen"], finding["firstSeenAt"]) + group["statuses"].add(finding["status"]) + group["url"] = group["url"] or finding["url"] + group["title"] = group["title"] or finding["title"] + group["items"].append((finding, target)) + rows = [] + for group in groups.values(): + group["status"] = min(group["statuses"], key=STATUS_RANK.get) + group["imageCount"] = len({t["id"] for _, t in group["items"]}) + group["workloadCount"] = len({(w["namespace"], w["kind"], w["name"]) for _, t in group["items"] for w in t["workloads"]}) + group["package"] = ", ".join(sorted(group["packages"])) + group["fix"] = ", ".join(sorted(group["fixed"])) + rows.append(group) + sorts = { + "id": lambda r: r["id"], "severity": lambda r: SEV_RANK[r["severity"]], "cvss": lambda r: r["cvss"] if r["cvss"] is not None else -1, + "package": lambda r: r["package"], "fixed": lambda r: r["fix"], "firstSeen": lambda r: r["firstSeen"], + "status": lambda r: STATUS_RANK[r["status"]], "images": lambda r: r["imageCount"], "workloads": lambda r: r["workloadCount"], + } + rows.sort(key=lambda r: r["id"]) + rows.sort(key=sorts.get(params.get("sort"), sorts["severity"]), reverse=params.get("dir") == "desc") + page_rows, total, page, pages, start = _paginate(rows, params) + html_rows = [] + for row in page_rows: + lines = [] + fix_images = set() + for finding, target in row["items"]: + if finding["status"] == "fix-available": + fix_images.add(target["image"]) + lines.append("
  • " + e(target["id"]) + " [" + e(target["state"]) + "] — " + e(finding["package"]) + " " + + e(finding["installedVersion"]) + " → " + e(finding["fixedVersion"] or "no fix") + + _workload_lines(target) + "
    • ") + acceptance = next((f.get("acceptance") for f, _ in row["items"] if f.get("acceptance")), None) + status_text = e(row["status"]) + if acceptance: + status_text += " (" + e(acceptance["reason"]) + ", until " + e(acceptance["expires"]) + ")" + html_rows.append( + "" + _link(row["url"], row["id"]) + "" + _sev(row["severity"]) + "" + + e("" if row["cvss"] is None else row["cvss"]) + "" + e(row["package"]) + "" + e(row["fix"]) + + "" + e(row["firstSeen"]) + "" + status_text + "" + _remediation_cell(fix_images, remediation) + + "
    " + e(row["imageCount"]) + " images, " + e(row["workloadCount"]) + + " workloads

    " + e(row["title"]) + "

    ") + head = [_th(params, "CVE", "id"), _th(params, "Severity", "severity"), _th(params, "CVSS", "cvss"), _th(params, "Package", "package"), + _th(params, "Fixed in", "fixed"), _th(params, "First seen", "firstSeen"), _th(params, "Status", "status"), + "Remediation", _th(params, "Affected", "images")] + return _pager(params, total, page, pages, start, len(page_rows), "CVEs") \ + + _table(head, html_rows, "No CVE rows match these filters.") + + +def _workload_key(item): + return item["namespace"], item["kind"], item["name"] + + +def _workloads(report, targets, params): + by_target = {} + for finding in report["findings"]: + target = targets[finding["targetId"]] + if _match_finding(finding, target, params): + by_target.setdefault(target["id"], []).append(finding) + rows = {} + finding_filter = params.get("severity") or params.get("status") or params.get("fix") + needle = params.get("q", "").lower() + for target in report["targets"]: + matching = by_target.get(target["id"], []) + for workload in target["workloads"]: + key = _workload_key(workload) + scope = {"namespace": params.get("namespace"), "workload": params.get("workload")} + if not _match_workload({"workloads": [workload]}, scope): + continue + text = " ".join((workload["namespace"], workload["name"], *workload["containers"], target["image"])).lower() + if not matching and (finding_filter or (needle and needle not in text)): + continue + row = rows.setdefault(key, {"key": key, "containers": set(), "images": set(), "findings": {}, "counts": dict.fromkeys(SEVERITIES, 0)}) + row["containers"].update(workload["containers"]) + row["images"].add(target["id"]) + for finding in matching: + row["findings"].setdefault((finding["id"], target["id"], finding["package"]), finding) + for row in rows.values(): + for finding in row["findings"].values(): + row["counts"][finding["severity"]] += 1 + items = list(rows.values()) + sorts = { + "namespace": lambda r: r["key"][0], "kind": lambda r: r["key"][1], "name": lambda r: r["key"][2], + "images": lambda r: len(r["images"]), "total": lambda r: -len(r["findings"]), + } + for severity in SEVERITIES: + sorts[severity.lower()] = (lambda s: lambda r: -r["counts"][s])(severity) + items.sort(key=lambda r: r["key"]) + if params.get("sort") in sorts: + items.sort(key=sorts[params["sort"]], reverse=params.get("dir") == "desc") + else: + items.sort(key=lambda r: tuple(-r["counts"][s] for s in SEVERITIES)) + page_rows, total, page, pages, start = _paginate(items, params) + html_rows = [] + for row in page_rows: + lines = "".join( + "
  • " + _link(f["url"], f["id"]) + " " + _sev(f["severity"]) + " " + e(f["package"]) + " — " + e(f["status"]) + "
    • " + for f in sorted(row["findings"].values(), key=lambda f: (SEV_RANK[f["severity"]], f["id"]))) + counts = "".join("" + e(row["counts"][s]) + "" for s in SEVERITIES) + html_rows.append( + "" + e(row["key"][0]) + "" + e(row["key"][1]) + "" + e(row["key"][2]) + "" + + e(", ".join(sorted(row["containers"])) or "none") + "" + "
    ".join(e(i) for i in sorted(row["images"])) + "" + + counts + "
    " + e(len(row["findings"])) + " findings
    ") + head = [_th(params, "Namespace", "namespace"), _th(params, "Kind", "kind"), _th(params, "Workload", "name"), "Containers", + _th(params, "Images", "images")] + [_th(params, s, s.lower()) for s in SEVERITIES] + [_th(params, "Findings", "total")] + return _pager(params, total, page, pages, start, len(page_rows), "workloads") \ + + _table(head, html_rows, "No workload rows match these filters.") + + +def _inventory(report, params): + counts = {} + by_target = {} + for finding in report["findings"]: + counts[finding["targetId"]] = counts.get(finding["targetId"], 0) + 1 + by_target.setdefault(finding["targetId"], []).append(finding) + needle = params.get("q", "").lower() + wanted_state = params.get("status") + rows = [] + for target in report["targets"]: + if wanted_state and target["state"] != wanted_state: + continue + if not _match_workload(target, params): + continue + if needle: + text = " ".join((target["id"], target["image"], target["state"], *[w["namespace"] + " " + w["name"] for w in target["workloads"]])).lower() + if needle not in text: + continue + rows.append(target) + sorts = { + "id": lambda t: t["id"], "type": lambda t: t["type"], "state": lambda t: STATE_RANK[t["state"]], + "lastScanned": lambda t: t["lastScannedAt"] or "", "findings": lambda t: counts.get(t["id"], 0), + } + rows.sort(key=lambda t: t["id"]) + rows.sort(key=sorts.get(params.get("sort"), sorts["state"]), reverse=params.get("dir") == "desc") + page_rows, total, page, pages, start = _paginate(rows, params) + html_rows = [] + for target in page_rows: + found = sorted(by_target.get(target["id"], []), key=lambda f: (SEV_RANK[f["severity"]], f["id"])) + listed = "".join("
  • " + _link(f["url"], f["id"]) + " " + _sev(f["severity"]) + " " + e(f["package"]) + " — " + e(f["status"]) + "
    • " + for f in found[:_EXPANSION_LIMIT]) + if len(found) > _EXPANSION_LIMIT: + listed += '
  • ' + e(len(found) - _EXPANSION_LIMIT) + " more in the CVE view
    • " + html_rows.append( + "" + e(target["id"]) + "" + e(target["type"]) + "" + e(target["state"]) + "" + + e(target["reason"]) + "" + e(target["lastScannedAt"] or "never") + "" + e(counts.get(target["id"], 0)) + + "
    " + e(len(target["workloads"])) + " workloads, " + e(len(found)) + " findings" + + _workload_lines(target) + ("" if listed else "") + "
    ") + head = [_th(params, "Target", "id"), _th(params, "Type", "type"), _th(params, "State", "state"), "Reason", + _th(params, "Last scanned", "lastScanned"), _th(params, "Findings", "findings"), "Workloads"] + return _pager(params, total, page, pages, start, len(page_rows), "targets") \ + + _table(head, html_rows, "No targets match these filters.") + + +# --- header ----------------------------------------------------------------------------------------- + +def _header(report, now): + generated = parse_ts(report["generatedAt"]) + database = parse_ts(report["database"]["updatedAt"]) + states = dict.fromkeys(report_mod.STATES, 0) + for target in report["targets"]: + states[target["state"]] += 1 + open_counts = dict.fromkeys(SEVERITIES, 0) + accepted = 0 + for finding in report["findings"]: + if finding["status"] == "accepted": + accepted += 1 + else: + open_counts[finding["severity"]] += 1 + warnings = [] + report_stale = generated is None or now - generated > REPORT_STALE + if report_stale: + warnings.append("STALE REPORT: generated " + _age(generated, now) + "; the expected refresh is within 26 hours.") + if report["run"]["status"] == "degraded": + warnings.append("DEGRADED RUN: the last run did not complete normally (database, acceptances file or inventory problem). Do not treat these results as current.") + db_epoch = report["database"]["version"] == "unavailable" + if db_epoch: + warnings.append("DATABASE UNAVAILABLE: no vulnerability database has been downloaded.") + elif database is None or now - database > DB_STALE: + warnings.append("DATABASE STALE: the vulnerability database is older than 48 hours.") + total_targets = len(report["targets"]) + not_scanned = total_targets - states["scanned"] + if total_targets == 0: + warnings.append("NO TARGETS: the inventory is empty; nothing has been evaluated.") + if not_scanned: + warnings.append("INCOMPLETE COVERAGE: " + str(not_scanned) + " of " + str(total_targets) + + " targets are not in the scanned state; findings for them are missing or may be out of date.") + open_total = sum(open_counts.values()) + if warnings: + status = "Status: ATTENTION NEEDED" + elif open_total or accepted: + status = "Status: CURRENT, findings present" + else: + status = "Status: CURRENT" + tiles = "".join("" + e(name) + ": " + e(states[name]) + "" for name in report_mod.STATES) + sev_tiles = "".join("" + _sev(s) + ": " + e(open_counts[s]) + "" for s in SEVERITIES) + claim = "" + if not warnings and not report["findings"] and total_targets > 0 and not_scanned == 0: + claim = "

    Clean: all " + e(total_targets) + " targets were evaluated within 24 hours with zero findings.

    " + return ("

    CVE visibility

    " + "

    " + e(status) + "

    " + + "".join('

    ' + e(w) + "

    " for w in warnings) + claim + + "

    Report age: " + e(_age(generated, now)) + " · Database age: " + + e("unavailable" if db_epoch else _age(database, now)) + " · Run: " + e(report["run"]["status"]) + "

    " + + '

    Targets (' + e(total_targets) + "): " + tiles + "

    " + + '

    Open findings by severity: ' + sev_tiles + "accepted: " + e(accepted) + "

    " + + "
    ") + + +def _nav(params): + links = [] + for view, label in (("cves", "CVEs"), ("workloads", "Workloads"), ("inventory", "Inventory")): + marker = ' aria-current="page"' if params["view"] == view else "" + target = {k: v for k, v in params.items() if k not in ("page", "sort", "dir")} + target["view"] = view + links.append('" + e(label) + "") + return "" + + +def _filters(params): + def field(name, label): + return "' + selected = params.get("fix", "") + fix_options = "".join( + '" + for v, t in (("", "any"), ("yes", "yes"), ("no", "no"))) + return ('
    ' + + field("q", "Search") + field("severity", "Severity") + field("namespace", "Namespace") + + field("workload", "Workload") + field("status", "Status") + + '" + + '
    ') + + +def _document(body): + return ("CVE visibility" + '" + "" + body + "").encode("utf-8") + + +def render(report, query, now): + """(status code, header dict, body bytes) for one page request.""" + if report is None: + body = ("

    CVE visibility

    NO REPORT YET: nothing has been evaluated. " + "Nothing can be said about this cluster until the first report is published.

    ") + return 503, dict(SECURITY_HEADERS), _document(body) + params = _parse(query) + targets = {t["id"]: t for t in report["targets"]} + remediation = {r["image"]: r for r in report["remediation"]} + view = params["view"] + if view == "workloads": + content = _workloads(report, targets, params) + elif view == "inventory": + content = _inventory(report, params) + else: + content = _cves(report, targets, params, remediation) + return 200, dict(SECURITY_HEADERS), _document(_header(report, now) + "
    " + _nav(params) + _filters(params) + content + "
    ") + + +def not_found(): + return 404, dict(SECURITY_HEADERS), _document("

    Not found.

    ") diff --git a/platform/vuln-scanner/src/report.py b/platform/vuln-scanner/src/report.py new file mode 100644 index 0000000..bb5e213 --- /dev/null +++ b/platform/vuln-scanner/src/report.py @@ -0,0 +1,222 @@ +"""Build, validate and publish report.json, the one interface to the page and the bot.""" +import json +import logging +import os +import re +from datetime import date, timedelta + +import acceptances as acceptances_mod +import state as st +from state import fmt_ts, parse_ts + +log = logging.getLogger("vuln-scanner.report") + +STATES = ("scanned", "pending", "stale", "failed", "credential-required", "by-tag", "not-covered") +SCAN_WINDOW = timedelta(hours=24) +PENDING_WINDOW = timedelta(hours=2) +DB_STALE = timedelta(hours=48) +NOT_SCANNED_REASON = "not scanned within 2h of first appearing" +EPOCH = "1970-01-01T00:00:00Z" # database.updatedAt when no database has ever been downloaded + +_SCHEMA_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "report.schema.json") +with open(_SCHEMA_PATH, encoding="utf-8") as _handle: + SCHEMA = json.load(_handle) + + +def coverage(target, record, now): + """(state, reason, failed-since datetime or None) for one target. The single place states come from.""" + if target["type"] != "image": + return "not-covered", target["reason"], None + record = record or st.new_record(now) + outcome = record["outcome"] + if outcome == "credential-required": + return "credential-required", record["reason"], None + if outcome == "failed": + return "failed", record["reason"], parse_ts(record["failedSince"]) or now + if record["lastAttemptAt"] is None: + appeared = parse_ts(record["firstDiscoveredAt"]) or now + if now - appeared > PENDING_WINDOW: + return "failed", NOT_SCANNED_REASON, appeared + PENDING_WINDOW + return "pending", None, None + last_success = parse_ts(record["lastSuccessAt"]) or now + if now - last_success > SCAN_WINDOW: + return "stale", "last successful scan was at " + record["lastSuccessAt"], None + return ("by-tag" if target["digest"] is None else "scanned"), None, None + + +def failed_over_24h(state, now): + count = 0 + for target in state["inventory"]: + status, _, since = coverage(target, state["scans"].get(st.scan_key(target)), now) + if status == "failed" and since is not None and now - since > SCAN_WINDOW: + count += 1 + return count + + +def build(state, accepted, acceptance_errors, now): + """The report for `state` at `now`; acceptance_errors degrades the run.""" + today = now.date() + targets = [] + findings = [] + for target in state["inventory"]: + record = state["scans"].get(st.scan_key(target)) + status, reason, _ = coverage(target, record, now) + last_scanned = record["lastSuccessAt"] if record else None + targets.append({ + "id": target["id"], "type": target["type"], "image": target["image"], "digest": target["digest"], + "state": status, "reason": reason, "lastScannedAt": last_scanned, "workloads": target["workloads"], + }) + for item in (record or {}).get("findings", []): + finding = {key: item[key] for key in + ("id", "severity", "cvss", "package", "installedVersion", "fixedVersion", "title", "url")} + finding["targetId"] = target["id"] + finding["firstSeenAt"] = record["firstSeen"].get(item["id"]) or fmt_ts(now) + acceptance = acceptances_mod.match(accepted, item["id"], target["image"], today) + if acceptance: + finding["status"] = "accepted" + finding["acceptance"] = {"reason": acceptance["reason"], "expires": acceptance["expires"].isoformat()} + else: + finding["status"] = "fix-available" if item["fixedVersion"] else "open" + findings.append(finding) + findings.sort(key=lambda f: (f["id"], f["targetId"], f["package"], f["installedVersion"])) + database = dict(state["database"]) if state["database"] else {"updatedAt": EPOCH, "version": "unavailable"} + run = dict(state["run"]) if state["run"] else {"startedAt": fmt_ts(now), "finishedAt": fmt_ts(now), "status": "degraded"} + db_time = parse_ts(database["updatedAt"]) + if acceptance_errors or state.get("inventoryError") or state.get("loadError") \ + or db_time is None or now - db_time > DB_STALE: + run["status"] = "degraded" + return { + "schemaVersion": 1, "generatedAt": fmt_ts(now), "database": database, "run": run, + "targets": targets, "findings": findings, "remediation": list(state["remediation"]), + } + + +# --- hand validator for the JSON-Schema subset report.schema.json uses ----------------------------- + +def _type_ok(value, name): + if name == "object": + return isinstance(value, dict) + if name == "array": + return isinstance(value, list) + if name == "string": + return isinstance(value, str) + if name == "null": + return value is None + if name == "boolean": + return isinstance(value, bool) + if name == "number": + return isinstance(value, (int, float)) and not isinstance(value, bool) + if name == "integer": + return isinstance(value, int) and not isinstance(value, bool) + return False + + +def _format_ok(value, name): + if name == "date-time": + return parse_ts(value) is not None + if name == "date": + try: + date.fromisoformat(value) + return True + except (ValueError, TypeError): + return False + return True + + +def _check(schema, value, path, errors): + if len(errors) >= 50: + return + if "type" in schema: + names = schema["type"] if isinstance(schema["type"], list) else [schema["type"]] + if not any(_type_ok(value, n) for n in names): + errors.append(path + ": wrong type") + return + if "const" in schema and value != schema["const"]: + errors.append(path + ": not the required constant") + if "enum" in schema and value not in schema["enum"]: + errors.append(path + ": not an allowed value") + if isinstance(value, str): + if "minLength" in schema and len(value) < schema["minLength"]: + errors.append(path + ": too short") + if "maxLength" in schema and len(value) > schema["maxLength"]: + errors.append(path + ": too long") + if "pattern" in schema and not re.search(schema["pattern"], value): + errors.append(path + ": does not match the pattern") + if "format" in schema and not _format_ok(value, schema["format"]): + errors.append(path + ": not a valid " + schema["format"]) + if isinstance(value, (int, float)) and not isinstance(value, bool): + if "minimum" in schema and value < schema["minimum"]: + errors.append(path + ": below the minimum") + if "maximum" in schema and value > schema["maximum"]: + errors.append(path + ": above the maximum") + if isinstance(value, dict): + for key in schema.get("required", []): + if key not in value: + errors.append(path + ": missing " + key) + properties = schema.get("properties", {}) + for key, item in value.items(): + if key in properties: + _check(properties[key], item, path + "." + key, errors) + elif schema.get("additionalProperties") is False: + errors.append(path + ": unexpected key " + key) + if isinstance(value, list) and "items" in schema: + for index, item in enumerate(value): + _check(schema["items"], item, path + "[" + str(index) + "]", errors) + if "not" in schema: + trial = [] + _check(schema["not"], value, path, trial) + if not trial: + errors.append(path + ": matches a forbidden shape") + if "if" in schema: + trial = [] + _check(schema["if"], value, path, trial) + branch = schema.get("then") if not trial else schema.get("else") + if branch: + _check(branch, value, path, errors) + + +def validate(report): + """Schema and referential errors; an empty list means the report may be served.""" + errors = [] + _check(SCHEMA, report, "$", errors) + if errors: + return errors + ids = [t["id"] for t in report["targets"]] + if len(set(ids)) != len(ids): + errors.append("$.targets: duplicate target id") + known = set(ids) + for finding in report["findings"]: + if finding["targetId"] not in known: + errors.append("$.findings: targetId has no target") + break + return errors + + +def serialise(report): + return json.dumps(report, sort_keys=True, separators=(",", ":")).encode("utf-8") + + +def publish(report, data_dir): + """Write report.json. An invalid report writes nothing and is logged loudly.""" + errors = validate(report) + if errors: + log.critical("REFUSING to publish an invalid report; the previous one stands: %s", errors[:10]) + return False + st.atomic_write(os.path.join(data_dir, "report.json"), serialise(report)) + return True + + +def load_published(data_dir): + """The published report if present and still valid, else None (never a corrupt report).""" + try: + with open(os.path.join(data_dir, "report.json"), "rb") as handle: + report = json.loads(handle.read()) + except FileNotFoundError: + return None + except (OSError, ValueError): + log.critical("published report.json is unreadable; serving no report until the next publish") + return None + if validate(report): + log.critical("published report.json is invalid; serving no report until the next publish") + return None + return report diff --git a/platform/vuln-scanner/src/report.schema.json b/platform/vuln-scanner/src/report.schema.json new file mode 100644 index 0000000..cebeba7 --- /dev/null +++ b/platform/vuln-scanner/src/report.schema.json @@ -0,0 +1,110 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "report.schema.json", + "title": "vuln-scanner report (docs/17, schemaVersion 1)", + "type": "object", + "additionalProperties": false, + "required": ["schemaVersion", "generatedAt", "database", "run", "targets", "findings", "remediation"], + "properties": { + "schemaVersion": {"const": 1}, + "generatedAt": {"type": "string", "format": "date-time"}, + "database": { + "type": "object", + "additionalProperties": false, + "required": ["updatedAt", "version"], + "properties": { + "updatedAt": {"type": "string", "format": "date-time"}, + "version": {"type": "string", "minLength": 1, "maxLength": 128} + } + }, + "run": { + "type": "object", + "additionalProperties": false, + "required": ["startedAt", "finishedAt", "status"], + "properties": { + "startedAt": {"type": "string", "format": "date-time"}, + "finishedAt": {"type": "string", "format": "date-time"}, + "status": {"enum": ["complete", "degraded"]} + } + }, + "targets": { + "type": "array", + "items": { + "type": "object", + "additionalProperties": false, + "required": ["id", "type", "image", "digest", "state", "reason", "lastScannedAt", "workloads"], + "properties": { + "id": {"type": "string", "minLength": 1, "maxLength": 2048}, + "type": {"enum": ["image", "vm", "node"]}, + "image": {"type": "string", "maxLength": 2048}, + "digest": {"type": ["string", "null"], "pattern": "^sha256:[0-9a-f]{64}$"}, + "state": {"enum": ["scanned", "pending", "stale", "failed", "credential-required", "by-tag", "not-covered"]}, + "reason": {"type": ["string", "null"], "maxLength": 1024}, + "lastScannedAt": {"type": ["string", "null"], "format": "date-time"}, + "workloads": { + "type": "array", + "items": { + "type": "object", + "additionalProperties": false, + "required": ["namespace", "kind", "name", "containers"], + "properties": { + "namespace": {"type": "string", "maxLength": 253}, + "kind": {"type": "string", "minLength": 1, "maxLength": 253}, + "name": {"type": "string", "minLength": 1, "maxLength": 253}, + "containers": {"type": "array", "items": {"type": "string", "maxLength": 253}} + } + } + } + } + } + }, + "findings": { + "type": "array", + "items": { + "type": "object", + "additionalProperties": false, + "required": ["id", "targetId", "severity", "cvss", "package", "installedVersion", "fixedVersion", + "title", "url", "firstSeenAt", "status"], + "properties": { + "id": {"type": "string", "minLength": 1, "maxLength": 256}, + "targetId": {"type": "string", "minLength": 1, "maxLength": 2048}, + "severity": {"enum": ["CRITICAL", "HIGH", "MEDIUM", "LOW", "UNKNOWN"]}, + "cvss": {"type": ["number", "null"], "minimum": 0, "maximum": 10}, + "package": {"type": "string", "maxLength": 256}, + "installedVersion": {"type": "string", "maxLength": 256}, + "fixedVersion": {"type": ["string", "null"], "maxLength": 512}, + "title": {"type": "string", "maxLength": 512}, + "url": {"type": "string", "maxLength": 2048, "pattern": "^(https?://.+)?$"}, + "firstSeenAt": {"type": "string", "format": "date-time"}, + "status": {"enum": ["open", "fix-available", "accepted"]}, + "acceptance": { + "type": "object", + "additionalProperties": false, + "required": ["reason", "expires"], + "properties": { + "reason": {"type": "string", "minLength": 1, "maxLength": 2048}, + "expires": {"type": "string", "format": "date"} + } + } + }, + "if": {"properties": {"status": {"const": "accepted"}}}, + "then": {"required": ["acceptance"]}, + "else": {"not": {"required": ["acceptance"]}} + } + }, + "remediation": { + "type": "array", + "items": { + "type": "object", + "additionalProperties": false, + "required": ["image", "state", "url", "updatedAt"], + "properties": { + "image": {"type": "string", "minLength": 1, "maxLength": 512}, + "state": {"enum": ["pr-open", "approval-requested", "no-update-identified", "not-tracked"]}, + "url": {"type": ["string", "null"], "maxLength": 2048, "pattern": "^https?://.+$"}, + "updatedAt": {"type": "string", "format": "date-time"} + } + } + } + } +} diff --git a/platform/vuln-scanner/src/server.py b/platform/vuln-scanner/src/server.py new file mode 100644 index 0000000..7746bcf --- /dev/null +++ b/platform/vuln-scanner/src/server.py @@ -0,0 +1,150 @@ +"""The three HTTP listeners: page (loopback, behind oauth2-proxy), internal API, metrics.""" +import json +import logging +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import urlsplit + +import page +import trivy +from state import parse_ts + +log = logging.getLogger("vuln-scanner.server") + +MAX_TRIAGE_BYTES = 256 * 1024 +MAX_REMEDIATION = 5000 +REMEDIATION_STATES = ("pr-open", "approval-requested", "no-update-identified", "not-tracked") + + +class _Handler(BaseHTTPRequestHandler): + server_version = "vuln-scanner" + sys_version = "" + timeout = 30 + engine = None + + def log_message(self, fmt, *args): # request lines carry filters; keep them out of the log + return + + def _send(self, status, headers, body): + self.send_response(status) + for name, value in headers.items(): + self.send_header(name, value) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def _text(self, status, body, content_type="text/plain; charset=utf-8"): + self._send(status, {"Content-Type": content_type}, body.encode("utf-8")) + + +class _PageHandler(_Handler): + def send_error(self, code, message=None, explain=None): # keep the CSP on error responses too + status, headers, body = page.not_found() + self._send(code, headers, body) + + def do_GET(self): + parts = urlsplit(self.path) + if parts.path == "/": + self._send(*self.engine.render_page(parts.query)) + elif parts.path == "/healthz": + self._text(200, "ok\n") + else: + self._send(*page.not_found()) + + +class _MetricsHandler(_Handler): + def do_GET(self): + path = urlsplit(self.path).path + if path == "/metrics": + self._text(200, self.engine.metrics_text(), "text/plain; version=0.0.4; charset=utf-8") + elif path == "/healthz": + healthy = self.engine.healthy() + self._text(200 if healthy else 503, "ok\n" if healthy else "scan loop stopped\n") + else: + self._text(404, "not found\n") + + +class _InternalHandler(_Handler): + def do_GET(self): + path = urlsplit(self.path).path + if path == "/report.json": + body = self.engine.report_bytes() + if body is None: + self._text(503, "no report yet\n") + else: + self._send(200, {"Content-Type": "application/json"}, body) + elif path == "/healthz": + self._text(200, "ok\n") + else: + self._text(404, "not found\n") + + def do_POST(self): + if urlsplit(self.path).path != "/triage": + self._text(404, "not found\n") + return + content_type = (self.headers.get("Content-Type") or "").split(";", 1)[0].strip().lower() + if content_type != "application/json": + self._text(415, "content type must be application/json\n") + return + try: + length = int(self.headers.get("Content-Length", "")) + except ValueError: + self._text(411, "a content length is required\n") + return + if length < 0 or length > MAX_TRIAGE_BYTES: + self._text(413, "body too large\n") + return + raw = self.rfile.read(length) + try: + document = json.loads(raw) + except ValueError: + self._text(400, "body is not JSON\n") + return + errors = validate_triage(document) + if errors: + self._text(400, errors[0] + "\n") + return + self.engine.record_triage(document) + self._send(204, {}, b"") + + +def validate_triage(document): + """Error strings for a POST /triage body; empty means valid. Exact shape, nothing extra.""" + if not isinstance(document, dict) or set(document) != {"heartbeatAt", "remediation"}: + return ["body must have exactly heartbeatAt and remediation"] + if parse_ts(document["heartbeatAt"]) is None or len(document["heartbeatAt"]) > 64: + return ["heartbeatAt must be RFC 3339"] + entries = document["remediation"] + if not isinstance(entries, list) or len(entries) > MAX_REMEDIATION: + return ["remediation must be a list of at most " + str(MAX_REMEDIATION) + " entries"] + for index, entry in enumerate(entries): + prefix = "remediation[" + str(index) + "]: " + if not isinstance(entry, dict) or set(entry) != {"image", "state", "url", "updatedAt"}: + return [prefix + "must have exactly image, state, url, updatedAt"] + if not isinstance(entry["image"], str) or not 0 < len(entry["image"]) <= 512: + return [prefix + "image must be a string of at most 512 characters"] + if entry["state"] not in REMEDIATION_STATES: + return [prefix + "unknown state"] + url = entry["url"] + if url is not None and (not isinstance(url, str) or len(url) > 2048 or trivy.safe_url(url) != url): + return [prefix + "url must be null or an http(s) URL"] + if parse_ts(entry["updatedAt"]) is None or len(entry["updatedAt"]) > 64: + return [prefix + "updatedAt must be RFC 3339"] + return [] + + +def serve(handler_type, address, engine): + handler = type(handler_type.__name__, (handler_type,), {"engine": engine}) + server = ThreadingHTTPServer(address, handler) + server.daemon_threads = True + thread = threading.Thread(target=server.serve_forever, name="http-" + str(address[1]), daemon=True) + thread.start() + return server + + +def start_all(engine, page_address, internal_address, metrics_address): + return [ + serve(_PageHandler, page_address, engine), + serve(_InternalHandler, internal_address, engine), + serve(_MetricsHandler, metrics_address, engine), + ] diff --git a/platform/vuln-scanner/src/state.py b/platform/vuln-scanner/src/state.py new file mode 100644 index 0000000..56912db --- /dev/null +++ b/platform/vuln-scanner/src/state.py @@ -0,0 +1,147 @@ +"""Persisted scanner state and the time helpers every module shares. + +state.json holds what must survive a restart: the last inventory, one record per unit of scan +(a digest, or a by-tag target id), the database info, the last run, remediation and heartbeat. +Every write is temp-file + fsync + rename, so a crash leaves the previous file intact. +""" +import json +import logging +import os +import re +import tempfile +from datetime import datetime, timedelta, timezone + +log = logging.getLogger("vuln-scanner.state") + +STATE_VERSION = 1 +RETENTION = timedelta(days=7) # how long a scan record outlives its target in the inventory + +_RFC3339 = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d+)?(Z|[+-]\d{2}:\d{2})$") + + +def utcnow(): + return datetime.now(timezone.utc) + + +def fmt_ts(value): + return value.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def parse_ts(value): + """A timezone-aware datetime for an RFC 3339 string, or None for anything else.""" + if not isinstance(value, str) or not _RFC3339.match(value): + return None + try: + return datetime.fromisoformat(value) + except ValueError: + return None + + +def atomic_write(path, data): + directory = os.path.dirname(path) or "." + fd, tmp = tempfile.mkstemp(dir=directory, prefix=".tmp-") + try: + with os.fdopen(fd, "wb") as handle: + handle.write(data) + handle.flush() + os.fsync(handle.fileno()) + os.replace(tmp, path) + except BaseException: + try: + os.unlink(tmp) + except OSError: + pass + raise + dir_fd = os.open(directory, os.O_RDONLY) + try: + os.fsync(dir_fd) + finally: + os.close(dir_fd) + + +def new_state(): + return { + "version": STATE_VERSION, + "inventory": [], + "inventoryAt": None, + "inventoryError": None, + "scans": {}, + "database": None, + "run": None, + "lastCompleteRunAt": None, + "remediation": [], + "triageHeartbeatAt": None, + "loadError": None, + } + + +def new_record(now): + return { + "firstDiscoveredAt": fmt_ts(now), + "lastInventoriedAt": fmt_ts(now), + "lastAttemptAt": None, + "lastSuccessAt": None, + "outcome": None, + "reason": None, + "failedSince": None, + "findings": [], + "firstSeen": {}, + } + + +class Store: + """state.json on the data volume.""" + + def __init__(self, data_dir): + self.data_dir = data_dir + self.path = os.path.join(data_dir, "state.json") + + def load(self): + try: + with open(self.path, "rb") as handle: + loaded = json.loads(handle.read()) + if not isinstance(loaded, dict) or loaded.get("version") != STATE_VERSION: + raise ValueError("unrecognised state version") + except FileNotFoundError: + return new_state() + except (OSError, ValueError) as exc: + log.critical("state.json is unreadable (%s); starting empty and marking the run degraded", exc) + try: + os.replace(self.path, self.path + ".corrupt") + except OSError: + pass + fresh = new_state() + fresh["loadError"] = "state.json was unreadable at startup" + return fresh + merged = new_state() + merged.update(loaded) + return merged + + def save(self, state): + atomic_write(self.path, json.dumps(state, sort_keys=True).encode("utf-8")) + + +def merge_inventory(state, targets, now): + """Install a fresh inventory; create a record per new unit; forget long-gone units.""" + state["inventory"] = targets + state["inventoryAt"] = fmt_ts(now) + state["inventoryError"] = None + stamp = fmt_ts(now) + for target in targets: + key = scan_key(target) + if key is None: + continue + record = state["scans"].get(key) + if record is None: + record = state["scans"][key] = new_record(now) + record["lastInventoriedAt"] = stamp + cutoff = now - RETENTION + for key in [k for k, r in state["scans"].items() if (parse_ts(r["lastInventoriedAt"]) or now) < cutoff]: + del state["scans"][key] + + +def scan_key(target): + """The unit of scan: the digest, else the by-tag target id. VMs and nodes have none.""" + if target["type"] != "image": + return None + return target["digest"] or target["id"] diff --git a/platform/vuln-scanner/src/trivy.py b/platform/vuln-scanner/src/trivy.py new file mode 100644 index 0000000..2481965 --- /dev/null +++ b/platform/vuln-scanner/src/trivy.py @@ -0,0 +1,153 @@ +"""Run Trivy and normalise its output. Everything Trivy emits is untrusted upstream text.""" +import json +import logging +import re +import subprocess +from urllib.parse import urlsplit + +log = logging.getLogger("vuln-scanner.trivy") + +SEVERITIES = ("CRITICAL", "HIGH", "MEDIUM", "LOW", "UNKNOWN") +_CREDENTIAL = re.compile( + r"unauthorized|authentication required|\bdenied:|access to the resource is denied|forbidden" + r"|status(?: code)?:? ?(?:401|403)\b|\b40[13] (?:unauthorized|forbidden)", + re.IGNORECASE, +) +_CONTROL = re.compile(r"[\x00-\x1f\x7f]") + + +class ScanFailure(Exception): + """A scan did not produce a result. kind is `failed` or `credential-required`.""" + + def __init__(self, kind, reason): + super().__init__(reason) + self.kind = kind + self.reason = reason + + +def text(value, limit=512): + """A bounded, control-character-free string; non-strings become empty.""" + if not isinstance(value, str): + return "" + return _CONTROL.sub(" ", value).strip()[:limit] + + +def safe_url(value): + """value if it is a plain http(s) URL, else the empty string.""" + url = text(value, 2048) + if not url or url != value or re.search(r"\s", url): + return "" + try: + parts = urlsplit(url) + except ValueError: + return "" + return url if parts.scheme.lower() in ("http", "https") and parts.netloc else "" + + +def _cvss(vuln): + best = None + for vendor in (vuln.get("CVSS") or {}).values() if isinstance(vuln.get("CVSS"), dict) else (): + if not isinstance(vendor, dict): + continue + for key in ("V40Score", "V3Score", "V2Score"): + score = vendor.get(key) + if isinstance(score, (int, float)) and not isinstance(score, bool) and 0 <= score <= 10: + best = max(best, float(score)) if best is not None else float(score) + break + return best + + +def parse_results(document): + """Findings (id, severity, cvss, package, installedVersion, fixedVersion, title, url), deduped.""" + if not isinstance(document, dict) or "SchemaVersion" not in document: + raise ScanFailure("failed", "trivy produced an unrecognised report") + results = document.get("Results") + if results is None: + return [] + if not isinstance(results, list): + raise ScanFailure("failed", "trivy report has a malformed Results field") + seen = {} + for result in results: + vulnerabilities = result.get("Vulnerabilities") if isinstance(result, dict) else None + if vulnerabilities is None: + continue + if not isinstance(vulnerabilities, list): + raise ScanFailure("failed", "trivy report has a malformed Vulnerabilities field") + for vuln in vulnerabilities: + if not isinstance(vuln, dict) or not text(vuln.get("VulnerabilityID")): + raise ScanFailure("failed", "trivy report has a finding with no identifier") + severity = text(vuln.get("Severity")).upper() + fixed = text(vuln.get("FixedVersion")) + finding = { + "id": text(vuln.get("VulnerabilityID"), 256), + "severity": severity if severity in SEVERITIES else "UNKNOWN", + "cvss": _cvss(vuln), + "package": text(vuln.get("PkgName"), 256), + "installedVersion": text(vuln.get("InstalledVersion"), 256), + "fixedVersion": fixed or None, + "title": text(vuln.get("Title") or vuln.get("Description"), 512), + "url": safe_url(vuln.get("PrimaryURL")), + } + seen[(finding["id"], finding["package"], finding["installedVersion"])] = finding + return sorted(seen.values(), key=lambda f: (f["id"], f["package"], f["installedVersion"])) + + +def classify_failure(registry, stderr): + """Sort a non-zero Trivy exit into credential-required or failed. Reasons name the registry.""" + if _CREDENTIAL.search(stderr or ""): + return ScanFailure("credential-required", "registry " + registry + " refused anonymous access") + lines = [line.strip() for line in (stderr or "").splitlines() if line.strip()] + return ScanFailure("failed", text(lines[-1], 400) if lines else "trivy exited with an error and no message") + + +class Runner: + """The real Trivy binary. Tests inject a runner with the same four methods.""" + + def __init__(self, trivy_bin, cache_dir, timeout=600): + self.bin = trivy_bin + self.cache_dir = cache_dir + self.timeout = timeout + + def _run(self, args, timeout): + return subprocess.run([self.bin] + args, capture_output=True, text=True, timeout=timeout, check=False) + + def download_db(self): + result = self._run(["image", "--download-db-only", "--no-progress", "--cache-dir", self.cache_dir], 900) + if result.returncode != 0: + raise ScanFailure("failed", "database update failed: " + text(result.stderr, 300)) + + def db_info(self): + """{updatedAt, version} of the DB on disk, or None when no DB is present.""" + result = self._run(["version", "--format", "json", "--cache-dir", self.cache_dir], 60) + if result.returncode != 0: + return None + try: + db = json.loads(result.stdout).get("VulnerabilityDB") + except (ValueError, AttributeError): + return None + if not isinstance(db, dict) or not db.get("UpdatedAt"): + return None + return {"updatedAt": db["UpdatedAt"], "version": str(db.get("Version", "unknown"))} + + def scan(self, reference): + args = [ + "image", "--skip-db-update", "--image-src", "remote", "--scanners", "vuln", "--format", "json", + "--no-progress", "--cache-dir", self.cache_dir, "--timeout", str(self.timeout) + "s", reference, + ] + try: + result = self._run(args, self.timeout + 30) + except subprocess.TimeoutExpired: + raise ScanFailure("failed", "timed out after " + str(self.timeout) + "s") from None + except OSError as exc: + raise ScanFailure("failed", "could not start trivy: " + type(exc).__name__) from None + if result.returncode < 0: + raise ScanFailure("failed", "trivy killed by signal " + str(-result.returncode) + " (out of memory?)") + if result.returncode != 0: + raise classify_failure(reference.split("/", 1)[0], result.stderr) + try: + return json.loads(result.stdout) + except ValueError: + raise ScanFailure("failed", "trivy output was not valid JSON") from None + + def clean_cache(self): + self._run(["clean", "--scan-cache", "--cache-dir", self.cache_dir], 120)