Files
home-network-k3s-impl/conformance/tests/l0-static/vuln-metrics-no-per-cve-label/check.sh
T

106 lines
4.7 KiB
Bash
Executable File

#!/usr/bin/env bash
# =============================================================================
# L0 static check (offline, no cluster) — vuln-metrics-no-per-cve-label
# Serves: R-VULN-06 ("Prometheus receives aggregates only ... MUST NOT receive a per-CVE series").
#
# GIVEN platform/vuln-scanner/src/metrics.py and platform/vuln-scanner/alerts.yaml,
# THEN (1) every label name attached to a vuln_scanner_* metric is in the allowlist
# (state, severity, fixable);
# (2) metrics.py exposes every metric name from the contract;
# (3) alerts.yaml selects vuln_scanner_* metrics with no label matcher outside the allowlist;
# AND (4) the extractor is proven on mutated text: a `cve=` label, a `{image=...}` selector and a
# per-workload label are each flagged, and label-free text is not.
# A missing metrics.py is a FAIL: this is a whole-repo-scoped check and the file is part of the same
# delivery (the check has no skip path).
# =============================================================================
set -euo pipefail
DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
ROOT="$(cd "${DIR}/../../../.." && pwd)"
exec python3 - "$ROOT" <<'PY'
import re, sys
from pathlib import Path
import yaml
ROOT = Path(sys.argv[1])
NAME = "vuln-metrics-no-per-cve-label"
FAILURES, CHECKS = [], 0
ALLOWED = {"state", "severity", "fixable"}
METRICS = (
"vuln_scanner_report_generated_timestamp_seconds",
"vuln_scanner_last_complete_run_timestamp_seconds",
"vuln_scanner_database_updated_timestamp_seconds",
"vuln_scanner_targets",
"vuln_scanner_findings",
"vuln_scanner_failed_targets_over_24h",
"vuln_scanner_critical_fix_overdue",
"vuln_scanner_triage_heartbeat_timestamp_seconds",
)
# Label keys that could only carry per-CVE or per-target identity, anywhere a metric could be
# written. Catches label construction the name{...} scan cannot see (e.g. "cve_id": cve_id).
IDENTITY_LABEL = re.compile(r'''["'](cve|cve_id|advisory|package|pkg|image|digest|workload|namespace|pod|container|target|target_id)["']\s*[:=]''')
def check(cond, desc):
global CHECKS
CHECKS += 1
if not cond:
FAILURES.append(desc)
def label_names(text):
"""Every label key in `vuln_scanner_x{key="value",...}` occurrences (Python or PromQL text),
tolerating f-strings' doubled braces and multi-line calls."""
found = set()
for m in re.finditer(r"vuln_scanner_[a-z0-9_]+\{", text):
tail = text[m.end(): m.end() + 200]
close = tail.find("}\n")
chunk = tail[: close if close != -1 else None].split("\n", 1)[0]
for key in re.findall(r"([a-zA-Z_][a-zA-Z0-9_]*)\s*=\s*\\?[\"'{]", chunk):
found.add(key)
return found
METRICS_PY = ROOT / "platform" / "vuln-scanner" / "src" / "metrics.py"
ALERTS = ROOT / "platform" / "vuln-scanner" / "alerts.yaml"
if not METRICS_PY.is_file():
print(f"CHECK FAIL ({NAME}): platform/vuln-scanner/src/metrics.py is absent (the SCANNER's "
f"deliverable); the aggregate-only property cannot be asserted without it", file=sys.stderr)
raise SystemExit(1)
src = METRICS_PY.read_text()
check(all(m in src for m in METRICS), f"(2) metrics.py names every contract metric "
f"(missing: {[m for m in METRICS if m not in src]})")
labels = label_names(src)
extra = labels - ALLOWED
check(not extra, f"(1) metrics.py attaches labels outside the allowlist: {sorted(extra)}")
hits = IDENTITY_LABEL.findall(src)
check(not hits, f"(1) metrics.py constructs an identity-bearing metric label: {sorted(set(hits))}")
rules = yaml.safe_load(ALERTS.read_text())
exprs = [r["expr"] for g in rules["spec"]["groups"] for r in g["rules"]]
check(len(exprs) >= 5, "(3) alerts.yaml parsed five or more rules")
alert_labels = set()
for e in exprs:
alert_labels |= label_names(e)
check(not (alert_labels - ALLOWED),
f"(3) alerts.yaml selects labels outside the allowlist: {sorted(alert_labels - ALLOWED)}")
# (4) non-vacuity
negatives = {
"cve label": 'vuln_scanner_findings{severity="HIGH",cve="CVE-1"} 1\n',
"image selector": 'max(vuln_scanner_findings{image="docker.io/x"})\n',
"workload label": 'f"vuln_scanner_targets{{workload=\\"{w}\\",state=\\"x\\"}} 1"\n',
}
for name, text in negatives.items():
check(bool(label_names(text) - ALLOWED), f"(4) the extractor missed: {name}")
ok_text = 'vuln_scanner_targets{state="scanned"} 3\nvuln_scanner_findings{severity="LOW",fixable="true"} 1\n'
check(not (label_names(ok_text) - ALLOWED), "(4) the extractor flagged the allowed labels")
if FAILURES:
print(f"CHECK FAIL ({NAME}): {len(FAILURES)} of {CHECKS} assertions failed", file=sys.stderr)
for f in FAILURES:
print(f" - {f}", file=sys.stderr)
raise SystemExit(1)
print(f"PASS ({NAME}): {CHECKS} assertions")
PY