home-network-k3s-impl/conformance/tests/l1-rendering/vuln-scanner-scan-loop/check.py

#!/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)