Merge branch 'wip/vuln-scanner' into feat/vuln-visibility

This commit is contained in:
jakman85 committed 2026-11-09 10:29:05 -06:00
commit 2b9dcef6c7
32 files changed
+3951

No files matched your search

@@ -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
@@ -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)
@@ -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"
@@ -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
@@ -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)
@@ -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"
@@ -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
@@ -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)
@@ -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"
@@ -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"
}
}
]
}
@@ -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
@@ -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"<style>(.*?)</style>", 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 + ": <strong>" 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", "&lt;script&gt;alert(1)&lt;/script&gt;" in body)
check("hostile title is never live", "<script>" not in body)
hostile = copy.deepcopy(REPORT)
hostile["findings"][0]["url"] = "javascript:alert(1)"
hostile["findings"][0]["title"] = '"><img src=x onerror=alert(1)>'
hostile["targets"][0]["reason"] = "<b onmouseover=alert(1)>x</b>"
hostile["remediation"][0]["url"] = "javascript:evil()"
_, _, body = render(hostile, "view=cves")
parsed = tags(body)
check("a javascript: advisory URL never becomes a link", not any("javascript:" in (a.get("href") or "") for t, a in parsed.attrs if t == "a"))
check("no element carries a hostile inline handler", not any(k.startswith("on") for _, a in parsed.attrs for k in a))
check("no img element was injected", "img" not in parsed.tags)
_, _, body = render(hostile, "view=inventory")
check("hostile reason is escaped in inventory", "<b onmouseover" not in body and "&lt;b onmouseover" in body)
_, _, body = render(query="view=cves&q=" + "%3Cscript%3E")
check("a hostile search term is escaped", "<script>" not in body)
check("a linked CVE points at its advisory", 'href="https://example.test/CVE-2026-1001"' in render(query="view=cves")[2])
stale = copy.deepcopy(REPORT)
stale["generatedAt"] = (NOW - timedelta(hours=27)).strftime("%Y-%m-%dT%H:%M:%SZ")
_, _, body = render(stale)
check("stale report shows a warning", "STALE REPORT" in body)
check("stale report shows no normal state", "Status: ATTENTION NEEDED" in body and "Status: CURRENT" not in body)
check("mutation: a fresh report has no stale warning", "STALE REPORT" not in render(REPORT)[2])
degraded = copy.deepcopy(REPORT)
degraded["run"]["status"] = "degraded"
_, _, body = render(degraded)
check("degraded run shows a warning and no normal state", "DEGRADED RUN" in body and "Status: ATTENTION NEEDED" in body and "Status: CURRENT" not in body)
allgood = copy.deepcopy(REPORT)
allgood["targets"] = [t for t in allgood["targets"] if t["state"] == "scanned"]
allgood["findings"] = []
_, _, body = render(allgood)
check("the clean claim appears when every target is scanned with zero findings", "Clean: all 1 targets" in body)
check("mutation: an all-scanned report shows the current state", "Status: CURRENT" in body)
one_pending = copy.deepcopy(allgood)
one_pending["targets"].append({"id": "x@sha256:" + "0" * 64, "type": "image", "image": "x", "digest": "sha256:" + "0" * 64,
"state": "pending", "reason": None, "lastScannedAt": None, "workloads": []})
_, _, body = render(one_pending)
check("one pending target removes the clean claim", "Clean:" not in body and not CLEAN_RE.search(body))
for state in ("failed", "stale", "credential-required", "by-tag", "not-covered"):
variant = copy.deepcopy(allgood)
variant["targets"].append({**one_pending["targets"][-1], "state": state})
check("a " + state + " target removes the clean claim", not CLEAN_RE.search(render(variant)[2]))
empty = copy.deepcopy(allgood)
empty["targets"] = []
check("zero targets is never a clean claim", not CLEAN_RE.search(render(empty)[2]))
degraded_clean = copy.deepcopy(allgood)
degraded_clean["run"]["status"] = "degraded"
check("a degraded run never makes a clean claim", not CLEAN_RE.search(render(degraded_clean)[2]))
status, headers, body = render(None)
check("no report yet is a 503 with the CSP", status == 503 and "Content-Security-Policy" in headers)
check("no report yet makes no clean claim", not CLEAN_RE.search(body))
_, _, body = render(query="view=cves&severity=CRITICAL")
check("the severity filter narrows the list", "CVE-2026-1001" in body and "CVE-2026-1004" not in body)
check("a filter is carried in pagination and sort links", "severity=CRITICAL" in body)
_, _, body = render(query="view=cves&status=accepted")
check("accepted findings stay listed with their reason", "CVE-2026-1003" in body and "not reachable" in body)
_, _, body = render(query="view=cves&fix=no")
check("the fix filter excludes fixed CVEs", "CVE-2026-1001" not in body and "CVE-2026-1002" in body)
_, _, body = render(query="view=inventory&status=credential-required")
check("inventory filters on coverage state", "ghcr.io/org/private" in body and "postgres" not in body.split("<main>")[1])
_, _, body = render(query="view=workloads&namespace=ops")
check("the namespace filter narrows workloads", "nightly" in body and "<td>web</td>" not in body)
_, _, body = render(query="view=cves&q=openssl")
check("free text search matches packages", "CVE-2026-1001" in body and "CVE-2026-1004" not in body)
_, _, body = render()
check("default view excludes nothing", all(f["id"] in body for f in REPORT["findings"]))
check("remediation column shows pr-open with a link", 'href="https://gitea.test/pulls/1"' in body and "pr-open" in body)
big = copy.deepcopy(REPORT)
big["findings"] = [dict(REPORT["findings"][0], id="CVE-2026-" + format(i, "05d")) for i in range(250)]
_, _, body = render(big, "view=cves&size=50&page=2")
check("pagination states the total", "of 250 CVEs (page 2 of 5)" in body)
check("pagination shows only its page", "CVE-2026-00049" not in body and "CVE-2026-00050" in body and "CVE-2026-00100" not in body)
check("every page's links include next and previous", ">next<" in body and ">previous<" in body)
print("FAILED: " + str(len(FAILED)) if FAILED else "all checks passed")
sys.exit(1 if FAILED else 0)
@@ -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"
@@ -0,0 +1,199 @@
{
"database": {
"updatedAt": "2026-11-09T06:00:00Z",
"version": "2"
},
"findings": [
{
"cvss": 9.1,
"firstSeenAt": "2026-11-01T00:00:00Z",
"fixedVersion": "3.0.2",
"id": "CVE-2026-1001",
"installedVersion": "1.0",
"package": "openssl",
"severity": "CRITICAL",
"status": "fix-available",
"targetId": "docker.io/library/nginx@sha256:a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1",
"title": "OpenSSL flaw",
"url": "https://example.test/CVE-2026-1001"
},
{
"cvss": null,
"firstSeenAt": "2026-11-01T00:00:00Z",
"fixedVersion": null,
"id": "CVE-2026-1002",
"installedVersion": "1.0",
"package": "zlib",
"severity": "LOW",
"status": "open",
"targetId": "docker.io/library/nginx@sha256:a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1",
"title": "<script>alert(1)</script>",
"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": ""
}
]
}
]
}
@@ -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
@@ -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)
@@ -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"
@@ -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": []
}
@@ -0,0 +1,4 @@
{
"ArtifactName": "x",
"SchemaVersion": 2
}
@@ -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
}
@@ -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
}
@@ -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
}
+141
View File
@@ -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)
+248
View File
@@ -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()
+276
View File
@@ -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()
+44
View File
@@ -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"
+482
View File
@@ -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 '<a href="' + e(url) + '" rel="noopener noreferrer">' + e(label) + "</a>" 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('<a href="' + e(_qs(params, page=page - 1)) + '">previous</a>')
if page < pages:
links.append('<a href="' + e(_qs(params, page=page + 1)) + '">next</a>')
return "<p>" + e(summary) + " " + " ".join(links) + "</p>"
def _th(params, label, key):
direction = "desc" if params.get("sort") == key and params.get("dir", "asc") == "asc" else "asc"
return '<th><a href="' + e(_qs(params, sort=key, dir=direction, page=1)) + '">' + e(label) + "</a></th>"
def _table(head, rows, empty):
if not rows:
return "<p>" + e(empty) + "</p>"
return "<table><thead><tr>" + "".join(head) + "</tr></thead><tbody>" + "".join(rows) + "</tbody></table>"
def _sev(severity):
return '<span class="sev-' + e(severity) + '">' + e(severity) + "</span>"
def _workload_lines(target):
items = []
for workload in target["workloads"]:
items.append("<li>" + e(workload["namespace"]) + " / " + e(workload["kind"]) + " / " + e(workload["name"])
+ " — containers: " + e(", ".join(workload["containers"]) or "none") + "</li>")
return "<ul>" + "".join(items) + "</ul>"
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 "<br>".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("<li>" + e(target["id"]) + " [" + e(target["state"]) + "] — " + e(finding["package"]) + " "
+ e(finding["installedVersion"]) + " → " + e(finding["fixedVersion"] or "no fix")
+ _workload_lines(target) + "</li>")
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(
"<tr><td>" + _link(row["url"], row["id"]) + "</td><td>" + _sev(row["severity"]) + "</td><td>"
+ e("" if row["cvss"] is None else row["cvss"]) + "</td><td>" + e(row["package"]) + "</td><td>" + e(row["fix"])
+ "</td><td>" + e(row["firstSeen"]) + "</td><td>" + status_text + "</td><td>" + _remediation_cell(fix_images, remediation)
+ "</td><td><details><summary>" + e(row["imageCount"]) + " images, " + e(row["workloadCount"])
+ " workloads</summary><p>" + e(row["title"]) + "</p><ul>" + "".join(lines) + "</ul></details></td></tr>")
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"),
"<th>Remediation</th>", _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(
"<li>" + _link(f["url"], f["id"]) + " " + _sev(f["severity"]) + " " + e(f["package"]) + " — " + e(f["status"]) + "</li>"
for f in sorted(row["findings"].values(), key=lambda f: (SEV_RANK[f["severity"]], f["id"])))
counts = "".join("<td>" + e(row["counts"][s]) + "</td>" for s in SEVERITIES)
html_rows.append(
"<tr><td>" + e(row["key"][0]) + "</td><td>" + e(row["key"][1]) + "</td><td>" + e(row["key"][2]) + "</td><td>"
+ e(", ".join(sorted(row["containers"])) or "none") + "</td><td>" + "<br>".join(e(i) for i in sorted(row["images"])) + "</td>"
+ counts + "<td><details><summary>" + e(len(row["findings"])) + " findings</summary><ul>" + lines + "</ul></details></td></tr>")
head = [_th(params, "Namespace", "namespace"), _th(params, "Kind", "kind"), _th(params, "Workload", "name"), "<th>Containers</th>",
_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("<li>" + _link(f["url"], f["id"]) + " " + _sev(f["severity"]) + " " + e(f["package"]) + " — " + e(f["status"]) + "</li>"
for f in found[:_EXPANSION_LIMIT])
if len(found) > _EXPANSION_LIMIT:
listed += '<li><a href="' + e(_qs({"view": "cves", "q": target["image"]})) + '">' + e(len(found) - _EXPANSION_LIMIT) + " more in the CVE view</a></li>"
html_rows.append(
"<tr><td><code>" + e(target["id"]) + "</code></td><td>" + e(target["type"]) + "</td><td>" + e(target["state"]) + "</td><td>"
+ e(target["reason"]) + "</td><td>" + e(target["lastScannedAt"] or "never") + "</td><td>" + e(counts.get(target["id"], 0))
+ "</td><td><details><summary>" + e(len(target["workloads"])) + " workloads, " + e(len(found)) + " findings</summary>"
+ _workload_lines(target) + ("<ul>" + listed + "</ul>" if listed else "") + "</details></td></tr>")
head = [_th(params, "Target", "id"), _th(params, "Type", "type"), _th(params, "State", "state"), "<th>Reason</th>",
_th(params, "Last scanned", "lastScanned"), _th(params, "Findings", "findings"), "<th>Workloads</th>"]
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("<span>" + e(name) + ": <strong>" + e(states[name]) + "</strong></span>" for name in report_mod.STATES)
sev_tiles = "".join("<span>" + _sev(s) + ": <strong>" + e(open_counts[s]) + "</strong></span>" for s in SEVERITIES)
claim = ""
if not warnings and not report["findings"] and total_targets > 0 and not_scanned == 0:
claim = "<p><strong>Clean: all " + e(total_targets) + " targets were evaluated within 24 hours with zero findings.</strong></p>"
return ("<header><h1>CVE visibility</h1>"
"<p class=\"status\">" + e(status) + "</p>"
+ "".join('<p class="warn">' + e(w) + "</p>" for w in warnings) + claim
+ "<p>Report age: " + e(_age(generated, now)) + " · Database age: "
+ e("unavailable" if db_epoch else _age(database, now)) + " · Run: " + e(report["run"]["status"]) + "</p>"
+ '<p class="tiles">Targets (' + e(total_targets) + "): " + tiles + "</p>"
+ '<p class="tiles">Open findings by severity: ' + sev_tiles + "<span>accepted: <strong>" + e(accepted) + "</strong></span></p>"
+ "</header>")
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('<a href="' + e(_qs(target)) + '"' + marker + ">" + e(label) + "</a>")
return "<nav>" + "".join(links) + "</nav>"
def _filters(params):
def field(name, label):
return "<label>" + e(label) + ' <input name="' + e(name) + '" value="' + e(params.get(name, "")) + '"></label>'
selected = params.get("fix", "")
fix_options = "".join(
'<option value="' + e(v) + '"' + (" selected" if selected == v else "") + ">" + e(t) + "</option>"
for v, t in (("", "any"), ("yes", "yes"), ("no", "no")))
return ('<form method="get"><input type="hidden" name="view" value="' + e(params["view"]) + '">'
+ field("q", "Search") + field("severity", "Severity") + field("namespace", "Namespace")
+ field("workload", "Workload") + field("status", "Status")
+ '<label>Fix available <select name="fix">' + fix_options + "</select></label>"
+ '<button type="submit">Filter</button></form>')
def _document(body):
return ("<!doctype html><html lang=\"en\"><head><meta charset=\"utf-8\"><title>CVE visibility</title>"
'<meta name="viewport" content="width=device-width, initial-scale=1"><style>' + STYLE + "</style></head>"
"<body>" + body + "</body></html>").encode("utf-8")
def render(report, query, now):
"""(status code, header dict, body bytes) for one page request."""
if report is None:
body = ("<main><h1>CVE visibility</h1><p class=\"warn\">NO REPORT YET: nothing has been evaluated. "
"Nothing can be said about this cluster until the first report is published.</p></main>")
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) + "<main>" + _nav(params) + _filters(params) + content + "</main>")
def not_found():
return 404, dict(SECURITY_HEADERS), _document("<main><p>Not found.</p></main>")
+222
View File
@@ -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
@@ -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"}
}
}
}
}
}
+150
View File
@@ -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),
]
+147
View File
@@ -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"]
+153
View File
@@ -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)