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, private=()):
    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, private_registries=private), 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())

class Killed(BaseException):
    """What an OOM kill looks like from inside the process: no `except Exception` runs."""


class KillingRunner(FakeRunner):
    def __init__(self, victim, data_dir):
        super().__init__()
        self.victim = victim
        self.data_dir = data_dir
        self.on_disk = None

    def scan(self, reference):
        if reference == self.victim:
            self.on_disk = json.loads(Path(self.data_dir, "state.json").read_text()).get("inFlight")
            raise Killed()
        return super().scan(reference)


with tempfile.TemporaryDirectory() as tmp:
    # R-VULN-05 BLOCKER: a scan that kills the container must cost that image only.
    eng, runner, clock = engine(tmp, runner=KillingRunner(POSTGRES, tmp))
    killed = False
    try:
        eng.tick()
    except Killed:
        killed = True
    check("kill: the simulated OOM really aborted the process path", killed)
    check("kill: inFlight is on disk BEFORE the scan starts", runner.on_disk == D4)
    eng2, runner2, clock2 = engine(tmp, runner=FakeRunner(), clock=clock)
    check("kill: startup clears inFlight", eng2.state["inFlight"] is None)
    check("kill: clearing is persisted", json.loads(Path(tmp, "state.json").read_text())["inFlight"] is None)
    eng2.publish()
    t = targets(eng2)
    check("kill: the target is failed", t[POSTGRES]["state"] == "failed")
    check("kill: the reason is the killed-scan text", t[POSTGRES]["reason"] == main.KILLED_REASON)
    eng2.tick()
    t = targets(eng2)
    check("kill: images after the killed one get scanned", PRIVATE in runner2.scans and t[PRIVATE]["state"] == "credential-required")
    check("kill: the killed image is not retried on restart (no wedge)", POSTGRES not in runner2.scans)
    check("kill: the killed target stays failed with its reason", t[POSTGRES]["reason"] == main.KILLED_REASON)
    check("kill: inFlight is clear after a normal cycle", eng2.state["inFlight"] is None)
    clock.advance(hours=21)
    eng2.tick()
    check("kill: the killed image is retried after the sit-out", runner2.scans.count(POSTGRES) == 1)
    check("kill: a successful retry clears the failure", targets(eng2)[POSTGRES]["state"] == "scanned")
    check("kill: a successful retry clears killedAt", eng2.state["scans"][D4]["killedAt"] is None)

with tempfile.TemporaryDirectory() as tmp:
    eng, runner, clock = engine(tmp)
    eng.refresh_inventory()
    eng2, runner2, clock2 = engine(tmp, runner=FakeRunner(), clock=clock)
    eng2.publish()
    check("mutation: a restart with no inFlight fails nothing", targets(eng2)[POSTGRES]["state"] == "pending")

with tempfile.TemporaryDirectory() as tmp:
    # A registry the scanner cannot reach is credential-required, and Trivy is never invoked.
    eng, runner, clock = engine(tmp, private=[" GHCR.io ", ""])
    eng.tick()
    t = targets(eng)
    check("private registry: reported credential-required", t[PRIVATE]["state"] == "credential-required")
    check("private registry: the reason names the registry", "ghcr.io" in t[PRIVATE]["reason"])
    check("private registry: Trivy is never invoked for it", PRIVATE not in runner.scans and NIGHTLY not in runner.scans)
    check("private registry: other registries still scan", NGINX in runner.scans)
    check("private registry: not counted as failed", report.failed_over_24h(eng.state, clock() + timedelta(hours=48)) == 0)
with tempfile.TemporaryDirectory() as tmp:
    eng, runner, clock = engine(tmp)
    eng.tick()
    check("mutation: with no private registry configured Trivy is invoked", PRIVATE in runner.scans)

with tempfile.TemporaryDirectory() as tmp:
    # A corrupt state.json degrades the run that rebuilds from it, not every run thereafter.
    Path(tmp, "state.json").write_text("{not json")
    eng, runner, clock = engine(tmp)
    check("loadError: set after a corrupt state.json", eng.state["loadError"])
    eng.tick()
    check("loadError: the rebuilding run is degraded", eng.report["run"]["status"] == "degraded")
    check("loadError: cleared after a full run with a good inventory", eng.state["loadError"] is None)
    clock.advance(hours=21)
    eng.tick()
    check("loadError: the next full run is complete", eng.report["run"]["status"] == "complete" and eng.state["lastCompleteRunAt"])
with tempfile.TemporaryDirectory() as tmp:
    Path(tmp, "state.json").write_text("{not json")
    eng, runner, clock = engine(tmp)
    eng.client.fail = True
    eng.tick()
    check("mutation: loadError is kept when the inventory failed", eng.state["loadError"])

with tempfile.TemporaryDirectory() as tmp:
    # lastCompleteRunAt means a report reached the store; a refused report must not advance it.
    eng, runner, clock = engine(tmp)
    real_publish = report.publish
    report.publish = lambda *_: False
    try:
        eng.tick()
    finally:
        report.publish = real_publish
    check("a refused publish does not advance lastCompleteRunAt", eng.state["lastCompleteRunAt"] is None)
    clock.advance(hours=21)
    eng.tick()
    check("mutation: a successful publish does advance it", eng.state["lastCompleteRunAt"] is not None)

with tempfile.TemporaryDirectory() as tmp:
    # One upstream string costs that string. Each of these used to freeze the whole report.
    long_reason = "x" * 3000
    eng, runner, clock = engine(tmp, acc="- id: CVE-2026-0010\n  reason: " + long_reason + "\n  expires: 2026-12-01\n")
    runner.outcomes[NGINX] = {"SchemaVersion": 2, "Results": [{"Vulnerabilities": [
        {"VulnerabilityID": "CVE-UP", "Severity": "HIGH", "PkgName": "p", "InstalledVersion": "1",
         "PrimaryURL": "HTTPS://upper.example/x", "Title": "t\ud800"}]}]}
    eng.tick()
    check("hostile strings: the report still publishes", json.loads(Path(tmp, "report.json").read_text())["findings"] and report.validate(eng.report) == [])
    up = [f for f in findings(eng, NGINX) if f["id"] == "CVE-UP"][0]
    check("hostile strings: an uppercase scheme drops only the URL", up["url"] == "")
    check("hostile strings: a lone surrogate is removed", "\ud800" not in up["title"])
    check("hostile strings: an over-long acceptance is ignored, the finding stays open", [f for f in findings(eng, POSTGRES) if f["id"] == "CVE-2026-0010"][0]["status"] != "accepted")
    check("hostile strings: an over-long acceptance degrades the run", eng.report["run"]["status"] == "degraded")
    check("hostile strings: the degraded run is not recorded as complete", eng.state["lastCompleteRunAt"] is None)

with tempfile.TemporaryDirectory() as tmp:
    # Isolation: values already in state, whatever put them there, cannot refuse the whole report.
    eng, runner, clock = engine(tmp)
    eng.tick()
    clean = report.build(eng.state, [], [], clock())
    check("mutation: a clean state is not reported as repaired", clean["run"]["status"] == "complete" and clean == report.build(eng.state, [], [], clock()))
    record = eng.state["scans"][A1]
    record["findings"][0].update(url="HTTP://x", title="t" * 900, severity="weird", cvss=99, package="p" * 900)
    eng.state["database"] = {"updatedAt": "not-a-time", "version": "1"}
    eng.state["remediation"] = [{"image": "a/b", "state": "pr-open", "url": "HTTP://x", "updatedAt": "2026-11-09T00:00:00Z"}]
    target = next(i for i in eng.state["inventory"] if i["id"] == NIGHTLY)
    target["id"] = "ghcr.io/" + "o" * 3000 + ":1"
    target["image"] = "ghcr.io/" + "o" * 3000
    eng.state["scans"][target["id"]] = eng.state["scans"].pop(NIGHTLY)  # a by-tag unit is keyed by its id
    built = report.build(eng.state, [], [], clock())
    check("isolation: the repaired report is schema-valid", report.validate(built) == [])
    check("isolation: a repair degrades the run", built["run"]["status"] == "degraded")
    bad = [f for f in built["findings"] if f["targetId"] == NGINX and f["title"] == "t" * 512][0]
    check("isolation: the bad finding survives with sanitised fields",
          bad["url"] == "" and bad["severity"] == "UNKNOWN" and bad["cvss"] is None and len(bad["package"]) == 256)
    check("isolation: every other finding is untouched", len(built["findings"]) == len(clean["findings"]))
    check("isolation: a bad remediation loses only its url", built["remediation"][0]["url"] is None)
    check("isolation: an unusable database is reported as unavailable", built["database"]["version"] == "unavailable")
    long_ids = [t for t in built["targets"] if len(t["id"]) > 2000]
    check("isolation: an over-long target stays listed", len(long_ids) == 1 and len(long_ids[0]["id"]) <= 2048)
    check("isolation: findings follow the shortened target id",
          any(f["targetId"] == long_ids[0]["id"] for f in built["findings"]))


class FakeProcess:
    returncode = 0

    def __init__(self, stdout):
        self.stdout = stdout


class VersionRunner(trivy.Runner):
    def __init__(self, stdout):
        super().__init__("/nonexistent", "/nonexistent")
        self.stdout = stdout

    def _run(self, args, timeout):
        return FakeProcess(self.stdout)


def db_info(updated, version):
    return VersionRunner(json.dumps({"VulnerabilityDB": {"UpdatedAt": updated, "Version": version}})).db_info()


check("db_info rejects an unparseable timestamp", db_info("yesterday", "2") is None)
check("db_info rejects a non-string timestamp", db_info(5, "2") is None)
check("db_info bounds the version", len(db_info("2026-11-09T06:13:52.123456789Z", "v" * 500)["version"]) == 128)
check("mutation: db_info keeps a good database", db_info("2026-11-09T06:13:52Z", "2") == {"updatedAt": "2026-11-09T06:13:52Z", "version": "2"})
check("a credential failure's reason is bounded", len(trivy.classify_failure("r" * 3000, "401 unauthorized").reason) <= 1024)

for url, want in (("https://a.test/x", "https://a.test/x"), ("http://a.test/", "http://a.test/"), ("HTTPS://a.test/x", ""),
                  ("HTTP://a.test/x", ""), ("Https://a.test/x", ""), ("javascript:alert(1)", ""), ("", "")):
    check("safe_url " + repr(url), trivy.safe_url(url) == want)
check("every url safe_url keeps satisfies the report schema",
      report.re.search(report.SCHEMA["properties"]["findings"]["items"]["properties"]["url"]["pattern"], trivy.safe_url("https://a.test/x")))

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