Merge branch 'wip/vuln-platform' into feat/vuln-visibility
This commit is contained in:
jakman85 committed
commit
b1a6fe6503
48 files changed
+2603
-6
No files matched your search
@@ -411,6 +411,28 @@ SCRAPE_OWNERS: list[dict] = [
|
||||
"Keeping the two in one file makes that pair inseparable",
|
||||
},
|
||||
},
|
||||
# ── ADDED 2026-11-09 (CVE visibility, ADR-0027). A NEW OWNER, NOT A RELAXED ASSERTION. ──────
|
||||
# Registers the vuln_scanner_* family and the Service+ServiceMonitor that scrapes it.
|
||||
{
|
||||
"owner": "vuln-scanner: Service + ServiceMonitor on port `metrics` (:9090)",
|
||||
"prefixes": ("vuln_scanner_",),
|
||||
"file": "platform/vuln-scanner/servicemonitor.yaml",
|
||||
"require": [
|
||||
r"^kind:\s*ServiceMonitor\s*$",
|
||||
r"^\s*-\s*port:\s*metrics\s*$",
|
||||
],
|
||||
"forbid": [],
|
||||
"why": "vuln_scanner_* powers the five VulnScan alerts, the only email path for scan "
|
||||
"failure, staleness and overdue critical fixes (R-VULN-06). Two of those rules fire "
|
||||
"on an absent metric, so a lost scrape is loud rather than inert",
|
||||
"companion": {
|
||||
"file": "platform/vuln-scanner/kustomization.yaml",
|
||||
"require": [r"^\s*-\s*networkpolicy\.yaml\s*$"],
|
||||
"why": "platform/vuln-scanner/networkpolicy.yaml ships the namespace's default-deny "
|
||||
"and the ingress allow for Prometheus on :9090 in one file, so a baseline "
|
||||
"without its allow cannot empty the TSDB",
|
||||
},
|
||||
},
|
||||
# ── ADDED 2026-11-08 (feat/cloudflare-tunnel). A NEW OWNER, NOT A RELAXED ASSERTION. ───────
|
||||
# Purely additive, the same shape as the kured entry: it registers the cloudflared_* family
|
||||
# and the Service+ServiceMonitor that scrapes it, and makes the check STRICTER for that component.
|
||||
|
||||
@@ -86,10 +86,10 @@ if not CONFIG.is_file():
|
||||
# headlamp: operator decision 2026-11-08 (LAN/VPN-only). qbittorrent/radarr/sonarr: catalog
|
||||
# `exposure: internal`, like jackett/pihole/semaphore.
|
||||
EXCLUDED_HOSTS = ("plex", "plex-vip", "jackett", "pihole", "pihole-dns", "semaphore",
|
||||
"qbittorrent", "radarr", "sonarr", "headlamp", "unifi", "crafty-mc")
|
||||
"qbittorrent", "radarr", "sonarr", "headlamp", "cve", "unifi", "crafty-mc")
|
||||
# Their Gateway-backend namespaces (unifi / crafty-mc are L4 VIPs, not Gateway backends).
|
||||
EXCLUDED_NAMESPACES = ("plex", "jackett", "pihole", "semaphore", "qbittorrent", "radarr",
|
||||
"sonarr", "headlamp")
|
||||
"sonarr", "headlamp", "vuln-scanner")
|
||||
FORBIDDEN_ENTITIES = {"kube-apiserver", "host", "remote-node", "cluster", "all"}
|
||||
# Any of these in a `service` means a route toward the Kubernetes API (or a raw-TCP/SSH bastion).
|
||||
API_MARKERS = ("kubernetes", "kube-apiserver", "apiserver", ":6443", "10.43.0.1", "tcp://",
|
||||
|
||||
@@ -187,6 +187,10 @@ KNOWN_NON_CATALOG_PUBLISHERS: dict[str, str] = {
|
||||
"decision: Headlamp is LAN/VPN-only) and the tunnel 404s the name. A Kubernetes web UI "
|
||||
"with native Keycloak OIDC; listed because it is a hand-authored route, and pinned "
|
||||
"disowned by (9d).",
|
||||
"platform/vuln-scanner/httproute.yaml:vuln-scanner/vuln-scanner":
|
||||
"publishes NOTHING: the route carries `controller: none` (docs/17: the CVE page is internal "
|
||||
"only) and the tunnel 404s the name. Hand-authored because the scanner is not a catalog "
|
||||
"service (its cluster-wide read role cannot be expressed in a ServiceDefinition, ADR-0027).",
|
||||
"platform/homepage/httproute.yaml:homepage/homepage":
|
||||
"publishes the APEX ${BASE_DOMAIN} itself -> the tunnel (proxied CNAME), via the https-apex "
|
||||
"listener. The apex is not a per-service name and cannot be withdrawn by an exposure "
|
||||
|
||||
@@ -0,0 +1,28 @@
|
||||
# =============================================================================
|
||||
# L0 STATIC — accepted CVE risk has a reason and a bounded expiry (R-VULN-08, docs/17 "Risk acceptance").
|
||||
# Offline, no cluster.
|
||||
# =============================================================================
|
||||
apiVersion: chainsaw.kyverno.io/v1alpha1
|
||||
kind: Test
|
||||
metadata:
|
||||
name: vuln-acceptances-schema
|
||||
labels:
|
||||
conformance.home.k3s/R-VULN-08: "true"
|
||||
annotations:
|
||||
conformance.home.k3s/requirements: "R-VULN-08"
|
||||
conformance.home.k3s/layer: "L0"
|
||||
conformance.home.k3s/scope: "whole-repo"
|
||||
conformance.home.k3s/safety-critical: "false"
|
||||
conformance.home.k3s/executable: "true"
|
||||
conformance.home.k3s/intent: >-
|
||||
platform/vuln-scanner/acceptances.yaml rejects any entry without a reason or expiry, with a
|
||||
malformed expiry, or expiring more than 180 days out. Proven with mutated negative fixtures.
|
||||
spec:
|
||||
steps:
|
||||
- name: assert-acceptance-schema
|
||||
description: >-
|
||||
Validate acceptances.yaml against the docs/17 shape, then run the validator on mutated
|
||||
negative fixtures to prove it can fail.
|
||||
try:
|
||||
- script:
|
||||
content: bash ./check.sh
|
||||
@@ -0,0 +1,105 @@
|
||||
#!/usr/bin/env bash
|
||||
# =============================================================================
|
||||
# L0 static check (offline, no cluster) — vuln-acceptances-schema
|
||||
# Serves: R-VULN-08 (docs/17 "Risk acceptance": "An entry with no reason or expiry is rejected by the gate").
|
||||
#
|
||||
# GIVEN platform/vuln-scanner/acceptances.yaml,
|
||||
# THEN (1) it is a YAML list;
|
||||
# (2) every entry has a non-empty id, a non-empty reason, an expires of the form YYYY-MM-DD
|
||||
# at most 180 days from today, and only the permitted keys (image is optional);
|
||||
# AND (3) the validator rejects mutated negative fixtures for each rule and accepts a good entry.
|
||||
# =============================================================================
|
||||
set -euo pipefail
|
||||
DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
ROOT="$(cd "${DIR}/../../../.." && pwd)"
|
||||
exec python3 - "$ROOT" <<'PY'
|
||||
import datetime as dt, re, sys
|
||||
from pathlib import Path
|
||||
import yaml
|
||||
|
||||
ROOT = Path(sys.argv[1])
|
||||
NAME = "vuln-acceptances-schema"
|
||||
FAILURES, CHECKS = [], 0
|
||||
TODAY = dt.datetime.now(dt.timezone.utc).date()
|
||||
MAX_DAYS = 180
|
||||
ALLOWED_KEYS = {"id", "image", "reason", "expires"}
|
||||
DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
|
||||
|
||||
|
||||
def check(cond, desc):
|
||||
global CHECKS
|
||||
CHECKS += 1
|
||||
if not cond:
|
||||
FAILURES.append(desc)
|
||||
|
||||
|
||||
def validate(entries):
|
||||
errs = []
|
||||
if not isinstance(entries, list):
|
||||
return ["acceptances must be a YAML list"]
|
||||
for i, e in enumerate(entries):
|
||||
if not isinstance(e, dict):
|
||||
errs.append(f"entry {i} is not a mapping")
|
||||
continue
|
||||
if set(e) - ALLOWED_KEYS:
|
||||
errs.append(f"entry {i} has unknown keys {sorted(set(e) - ALLOWED_KEYS)}")
|
||||
for field in ("id", "reason"):
|
||||
if not isinstance(e.get(field), str) or not e[field].strip():
|
||||
errs.append(f"entry {i} has no non-empty {field}")
|
||||
if "image" in e and (not isinstance(e["image"], str) or not e["image"].strip()):
|
||||
errs.append(f"entry {i} has an empty image")
|
||||
exp = e.get("expires")
|
||||
if isinstance(exp, dt.date) and not isinstance(exp, dt.datetime):
|
||||
exp_s = exp.isoformat()
|
||||
elif isinstance(exp, str):
|
||||
exp_s = exp
|
||||
else:
|
||||
errs.append(f"entry {i} has no expires")
|
||||
continue
|
||||
if not DATE_RE.match(exp_s):
|
||||
errs.append(f"entry {i} expires is malformed: {exp_s!r}")
|
||||
continue
|
||||
try:
|
||||
d = dt.date.fromisoformat(exp_s)
|
||||
except ValueError:
|
||||
errs.append(f"entry {i} expires is not a real date: {exp_s!r}")
|
||||
continue
|
||||
if (d - TODAY).days > MAX_DAYS:
|
||||
errs.append(f"entry {i} expires {exp_s}, more than {MAX_DAYS} days out")
|
||||
return errs
|
||||
|
||||
|
||||
path = ROOT / "platform" / "vuln-scanner" / "acceptances.yaml"
|
||||
if not path.is_file():
|
||||
print(f"CHECK FAIL ({NAME}): platform/vuln-scanner/acceptances.yaml is absent", file=sys.stderr)
|
||||
raise SystemExit(1)
|
||||
entries = yaml.safe_load(path.read_text())
|
||||
check(entries == [] or (isinstance(entries, list) and entries), "(1) acceptances.yaml parsed to a list")
|
||||
errs = validate(entries)
|
||||
check(not errs, f"(2) acceptances.yaml is valid: {errs}")
|
||||
|
||||
soon = (TODAY + dt.timedelta(days=30)).isoformat()
|
||||
far = (TODAY + dt.timedelta(days=200)).isoformat()
|
||||
good = {"id": "CVE-2026-00000", "image": "docker.io/library/x", "reason": "not reachable", "expires": soon}
|
||||
check(validate([good]) == [], "(3) a well-formed entry is accepted")
|
||||
check(validate([{k: v for k, v in good.items() if k != "image"}]) == [], "(3) image is optional (platform-wide)")
|
||||
for name, entry in {
|
||||
"no reason": {k: v for k, v in good.items() if k != "reason"},
|
||||
"empty reason": dict(good, reason=""),
|
||||
"no expires": {k: v for k, v in good.items() if k != "expires"},
|
||||
"no id": {k: v for k, v in good.items() if k != "id"},
|
||||
"malformed expiry": dict(good, expires="next-tuesday"),
|
||||
"impossible date": dict(good, expires="2027-13-45"),
|
||||
"past 180 days": dict(good, expires=far),
|
||||
"unknown key": dict(good, owner="me"),
|
||||
}.items():
|
||||
check(bool(validate([entry])), f"(3) the validator accepted a bad entry: {name}")
|
||||
check(bool(validate({"not": "a list"})), "(3) the validator accepted a non-list")
|
||||
|
||||
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
|
||||
@@ -0,0 +1,29 @@
|
||||
# =============================================================================
|
||||
# L0 STATIC — the five scanner alerts exist and are unit-tested, including the absent-metric cases
|
||||
# (R-VULN-06, docs/17 "Acceptance gates"). Offline, no cluster.
|
||||
# =============================================================================
|
||||
apiVersion: chainsaw.kyverno.io/v1alpha1
|
||||
kind: Test
|
||||
metadata:
|
||||
name: vuln-alerts-unit-tested
|
||||
labels:
|
||||
conformance.home.k3s/R-VULN-06: "true"
|
||||
annotations:
|
||||
conformance.home.k3s/requirements: "R-VULN-06"
|
||||
conformance.home.k3s/layer: "L0"
|
||||
conformance.home.k3s/scope: "whole-repo"
|
||||
conformance.home.k3s/safety-critical: "false"
|
||||
conformance.home.k3s/executable: "true"
|
||||
conformance.home.k3s/intent: >-
|
||||
VulnScanStale, VulnDatabaseStale, VulnCoverageGap, VulnCriticalFixAvailable and
|
||||
VulnTriageBotSilent exist, each promtool-tested firing, and the two absent-metric rules fire
|
||||
in a test with no such series. Proven by mutation.
|
||||
spec:
|
||||
steps:
|
||||
- name: assert-alert-unit-tests
|
||||
description: >-
|
||||
Parse alerts.yaml and alerts.promtool-test.yaml, assert every alert and the two absent cases
|
||||
are pinned, run the harness when promtool is present, and run the evaluator on mutated copies.
|
||||
try:
|
||||
- script:
|
||||
content: bash ./check.sh
|
||||
@@ -0,0 +1,153 @@
|
||||
#!/usr/bin/env bash
|
||||
# =============================================================================
|
||||
# L0 static check (offline, no cluster) — vuln-alerts-unit-tested
|
||||
# Serves: R-VULN-06 (docs/17 "Acceptance gates": alerts unit-tested, including the absent-metric cases).
|
||||
#
|
||||
# GIVEN platform/vuln-scanner/alerts.yaml and alerts.promtool-test.yaml,
|
||||
# THEN (1) the rules are exactly the five contract alerts, with the two absent()-bearing rules;
|
||||
# (2) each alert is asserted FIRING in a promtool test;
|
||||
# (3) VulnScanStale and VulnTriageBotSilent each fire in a test with NO input series for their
|
||||
# own metric (the absent case) and in one where it vanishes (re-fire);
|
||||
# (4) each alert carries severity and a runbook_url;
|
||||
# (5) when promtool is present (the toolbox), the actual harness passes on this file;
|
||||
# AND (6) the evaluator is run on mutated copies (the absent arm deleted, a firing assertion
|
||||
# removed, an alert dropped) and must flag each (non-vacuity).
|
||||
# =============================================================================
|
||||
set -euo pipefail
|
||||
DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
ROOT="$(cd "${DIR}/../../../.." && pwd)"
|
||||
exec python3 - "$ROOT" <<'PY'
|
||||
import copy, re, shutil, subprocess, sys
|
||||
from pathlib import Path
|
||||
import yaml
|
||||
|
||||
ROOT = Path(sys.argv[1])
|
||||
NAME = "vuln-alerts-unit-tested"
|
||||
FAILURES, CHECKS = [], 0
|
||||
ALERTS = {"VulnScanStale", "VulnDatabaseStale", "VulnCoverageGap", "VulnCriticalFixAvailable", "VulnTriageBotSilent"}
|
||||
ABSENT_ALERTS = {
|
||||
"VulnScanStale": "vuln_scanner_last_complete_run_timestamp_seconds",
|
||||
"VulnTriageBotSilent": "vuln_scanner_triage_heartbeat_timestamp_seconds",
|
||||
}
|
||||
|
||||
|
||||
def check(cond, desc):
|
||||
global CHECKS
|
||||
CHECKS += 1
|
||||
if not cond:
|
||||
FAILURES.append(desc)
|
||||
|
||||
|
||||
COMP = ROOT / "platform" / "vuln-scanner"
|
||||
for f in ("alerts.yaml", "alerts.promtool-test.yaml"):
|
||||
if not (COMP / f).is_file():
|
||||
print(f"CHECK FAIL ({NAME}): platform/vuln-scanner/{f} is absent", file=sys.stderr)
|
||||
raise SystemExit(1)
|
||||
rules = yaml.safe_load((COMP / "alerts.yaml").read_text())
|
||||
tests = yaml.safe_load((COMP / "alerts.promtool-test.yaml").read_text())
|
||||
|
||||
|
||||
def alert_rules(doc):
|
||||
return {r["alert"]: r for g in doc["spec"]["groups"] for r in g["rules"]}
|
||||
|
||||
|
||||
def eval_alerts(doc):
|
||||
bad = []
|
||||
got = alert_rules(doc)
|
||||
if set(got) != ALERTS:
|
||||
bad.append(f"alerts are {sorted(got)}, expected {sorted(ALERTS)}")
|
||||
for name, metric in ABSENT_ALERTS.items():
|
||||
if name in got and f"absent({metric})" not in got[name]["expr"].replace(" ", ""):
|
||||
bad.append(f"{name} lacks the absent({metric}) arm")
|
||||
for name, r in got.items():
|
||||
if not (r.get("labels") or {}).get("severity"):
|
||||
bad.append(f"{name} has no severity label")
|
||||
if "/runbooks/" not in ((r.get("annotations") or {}).get("runbook_url") or ""):
|
||||
bad.append(f"{name} has no runbook_url")
|
||||
return bad
|
||||
|
||||
|
||||
def firing_assertions(test):
|
||||
"""(alertname -> assertion list) for the firing expectations of one promtool test."""
|
||||
out = {}
|
||||
for pe in test.get("promql_expr_test") or []:
|
||||
expr, samples = pe.get("expr", ""), pe.get("exp_samples") or []
|
||||
m = re.search(r'alertname="(Vuln\w+)"', expr)
|
||||
if m and samples and 'alertstate="firing"' in (expr + " " + " ".join(s.get("labels", "") for s in samples)):
|
||||
out.setdefault(m.group(1), []).append(pe)
|
||||
return out
|
||||
|
||||
|
||||
def series_for(test):
|
||||
return [s.get("series", "") for s in test.get("input_series") or []]
|
||||
|
||||
|
||||
def eval_tests(doc):
|
||||
bad = []
|
||||
all_tests = doc.get("tests") or []
|
||||
fired = set()
|
||||
for t in all_tests:
|
||||
fired |= set(firing_assertions(t))
|
||||
missing = ALERTS - fired
|
||||
if missing:
|
||||
bad.append(f"no firing promtool assertion for {sorted(missing)}")
|
||||
for alert, metric in ABSENT_ALERTS.items():
|
||||
absent_case = any(
|
||||
alert in firing_assertions(t) and not any(s.startswith(metric) for s in series_for(t))
|
||||
for t in all_tests)
|
||||
if not absent_case:
|
||||
bad.append(f"{alert} has no firing test with NO {metric} series (the absent case)")
|
||||
vanish_case = any(
|
||||
alert in firing_assertions(t) and any(s.startswith(metric) and "stale" in str(i.get("values", ""))
|
||||
for i in t.get("input_series", []) for s in [i.get("series", "")])
|
||||
for t in all_tests)
|
||||
if not vanish_case:
|
||||
bad.append(f"{alert} has no test where its metric goes stale (the re-fire case)")
|
||||
return bad
|
||||
|
||||
|
||||
for v in eval_alerts(rules):
|
||||
check(False, f"(1,4) {v}")
|
||||
for v in eval_tests(tests):
|
||||
check(False, f"(2,3) {v}")
|
||||
check(len(alert_rules(rules)) == 5, "(1) five alert rules parsed")
|
||||
|
||||
# (5) the real harness, only when the rule engine is present.
|
||||
if shutil.which("promtool"):
|
||||
rc = subprocess.run(["bash", str(ROOT / "platform/monitoring/tests/run.sh"),
|
||||
"platform/vuln-scanner/alerts.promtool-test.yaml"],
|
||||
cwd=str(ROOT), capture_output=True, text=True)
|
||||
check(rc.returncode == 0, f"(5) the promtool harness failed: rc={rc.returncode}\n{rc.stdout[-800:]}{rc.stderr[-800:]}")
|
||||
|
||||
# (6) non-vacuity
|
||||
m = copy.deepcopy(rules)
|
||||
for g in m["spec"]["groups"]:
|
||||
for r in g["rules"]:
|
||||
if r["alert"] == "VulnTriageBotSilent":
|
||||
r["expr"] = r["expr"].split(" or ")[0]
|
||||
check(bool(eval_alerts(m)), "(6) evaluator missed: absent arm deleted")
|
||||
m = copy.deepcopy(rules)
|
||||
m["spec"]["groups"][0]["rules"].pop(1)
|
||||
check(bool(eval_alerts(m)), "(6) evaluator missed: alert dropped")
|
||||
m = copy.deepcopy(rules)
|
||||
for g in m["spec"]["groups"]:
|
||||
for r in g["rules"]:
|
||||
r["annotations"].pop("runbook_url", None)
|
||||
check(bool(eval_alerts(m)), "(6) evaluator missed: runbook_url removed")
|
||||
|
||||
m = copy.deepcopy(tests)
|
||||
for t in m["tests"]:
|
||||
t["promql_expr_test"] = [p for p in t["promql_expr_test"]
|
||||
if 'VulnCoverageGap' not in p.get("expr", "")]
|
||||
check(bool(eval_tests(m)), "(6) evaluator missed: firing assertion removed")
|
||||
m = copy.deepcopy(tests)
|
||||
m["tests"] = [t for t in m["tests"] if "ABSENT" not in t.get("name", "")]
|
||||
check(bool(eval_tests(m)), "(6) evaluator missed: absent-case tests removed")
|
||||
|
||||
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
|
||||
@@ -0,0 +1,28 @@
|
||||
# =============================================================================
|
||||
# L0 STATIC — Prometheus receives aggregates only (R-VULN-06, docs/17 "Alerting").
|
||||
# Offline, no cluster. Depends on platform/vuln-scanner/src/metrics.py (the SCANNER's file).
|
||||
# =============================================================================
|
||||
apiVersion: chainsaw.kyverno.io/v1alpha1
|
||||
kind: Test
|
||||
metadata:
|
||||
name: vuln-metrics-no-per-cve-label
|
||||
labels:
|
||||
conformance.home.k3s/R-VULN-06: "true"
|
||||
annotations:
|
||||
conformance.home.k3s/requirements: "R-VULN-06"
|
||||
conformance.home.k3s/layer: "L0"
|
||||
conformance.home.k3s/scope: "whole-repo"
|
||||
conformance.home.k3s/safety-critical: "false"
|
||||
conformance.home.k3s/executable: "true"
|
||||
conformance.home.k3s/intent: >-
|
||||
No vuln_scanner_* metric or alert rule carries a label other than state, severity or fixable,
|
||||
so Prometheus can never hold a per-CVE, per-image or per-workload series. Proven by mutation.
|
||||
spec:
|
||||
steps:
|
||||
- name: assert-metric-labels-are-aggregates
|
||||
description: >-
|
||||
Extract every vuln_scanner_* metric label from metrics.py and alerts.yaml, assert the
|
||||
allowlist, and run the extractor on mutated text to prove it can fail.
|
||||
try:
|
||||
- script:
|
||||
content: bash ./check.sh
|
||||
@@ -0,0 +1,105 @@
|
||||
#!/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
|
||||
@@ -0,0 +1,30 @@
|
||||
# =============================================================================
|
||||
# L0 STATIC — the scanner's Flux tier cannot stall the platform, and its network reach is exactly what
|
||||
# docs/17 "Security of the scanner itself" allows (R-VULN-09). Offline, no cluster.
|
||||
# =============================================================================
|
||||
apiVersion: chainsaw.kyverno.io/v1alpha1
|
||||
kind: Test
|
||||
metadata:
|
||||
name: vuln-scanner-isolation
|
||||
labels:
|
||||
conformance.home.k3s/R-VULN-09: "true"
|
||||
annotations:
|
||||
conformance.home.k3s/requirements: "R-VULN-09"
|
||||
conformance.home.k3s/layer: "L0"
|
||||
conformance.home.k3s/scope: "whole-repo"
|
||||
conformance.home.k3s/safety-critical: "false"
|
||||
conformance.home.k3s/executable: "true"
|
||||
conformance.home.k3s/intent: >-
|
||||
The vuln-scanner Flux tier is wait:false, in neither gating tier, and nothing dependsOn it;
|
||||
ingress is exactly the Gateway to 4180, the CI runner to 8081 and Prometheus to 9090;
|
||||
egress reaches the API, DNS, the public internet on 443 only with cluster and LAN excluded,
|
||||
and the Keycloak path the sidecar needs. Proven by mutation.
|
||||
spec:
|
||||
steps:
|
||||
- name: assert-isolation-and-network-shape
|
||||
description: >-
|
||||
Parse the Flux wiring and networkpolicy.yaml, assert isolation and the exact network shape,
|
||||
and run the evaluators on mutated copies to prove they can fail.
|
||||
try:
|
||||
- script:
|
||||
content: bash ./check.sh
|
||||
+210
@@ -0,0 +1,210 @@
|
||||
#!/usr/bin/env bash
|
||||
# =============================================================================
|
||||
# L0 static check (offline, no cluster) — vuln-scanner-isolation
|
||||
# Serves: R-VULN-09 (docs/17 "Security of the scanner itself").
|
||||
#
|
||||
# GIVEN the repository as authored,
|
||||
# THEN (1) the Flux tier is wait:false, dependsOn platform-config, in neither gating tier, nothing
|
||||
# anywhere dependsOn it, and both overlays carry its SOPS decryption patch;
|
||||
# (2) ingress is EXACTLY: reserved:ingress to 4180, act-runner to 8081, prometheus to 9090;
|
||||
# (3) egress to the world is tcp/443 only through a CIDR rule excluding every private range,
|
||||
# and the only other destinations are the API entity, the Gateway entity on 443 and Keycloak
|
||||
# (the oauth2-proxy sidecar's OIDC path, the one deliberate addition to the spec's list);
|
||||
# no `world`, `all`, `cluster`, `host` or `remote-node` entity;
|
||||
# (4) the namespace ships a default-deny baseline and is not app-namespace=true (single owner);
|
||||
# AND (5) each evaluator is run on a mutated copy and must flag it (non-vacuity).
|
||||
# =============================================================================
|
||||
set -euo pipefail
|
||||
DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
ROOT="$(cd "${DIR}/../../../.." && pwd)"
|
||||
exec python3 - "$ROOT" <<'PY'
|
||||
import copy, re, sys
|
||||
from pathlib import Path
|
||||
import yaml
|
||||
|
||||
ROOT = Path(sys.argv[1])
|
||||
NAME = "vuln-scanner-isolation"
|
||||
FAILURES, CHECKS = [], 0
|
||||
FORBIDDEN_ENTITIES = {"world", "all", "cluster", "host", "remote-node"}
|
||||
PRIVATE_RANGES = {"10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16"}
|
||||
|
||||
|
||||
def check(cond, desc):
|
||||
global CHECKS
|
||||
CHECKS += 1
|
||||
if not cond:
|
||||
FAILURES.append(desc)
|
||||
|
||||
|
||||
def docs(path):
|
||||
return [d for d in yaml.safe_load_all(path.read_text()) if isinstance(d, dict)]
|
||||
|
||||
|
||||
fk_path = ROOT / "clusters" / "base" / "vuln-scanner.yaml"
|
||||
np_path = ROOT / "platform" / "vuln-scanner" / "networkpolicy.yaml"
|
||||
for p in (fk_path, np_path):
|
||||
if not p.is_file():
|
||||
print(f"CHECK FAIL ({NAME}): {p.relative_to(ROOT)} is absent", file=sys.stderr)
|
||||
raise SystemExit(1)
|
||||
|
||||
# (1) Flux wiring
|
||||
fk = next(d for d in docs(fk_path) if d.get("kind") == "Kustomization")
|
||||
check(fk["spec"].get("wait") is False, "(1) clusters/base/vuln-scanner.yaml must be wait: false")
|
||||
check([d.get("name") for d in fk["spec"].get("dependsOn") or []] == ["platform-config"],
|
||||
"(1) the tier dependsOn exactly platform-config")
|
||||
check(fk["spec"].get("path") == "./platform/vuln-scanner", "(1) the tier reconciles ./platform/vuln-scanner")
|
||||
check((fk["spec"].get("postBuild") or {}).get("substituteFrom"),
|
||||
"(1) postBuild.substituteFrom is present (the route/alerts use ${BASE_DOMAIN})")
|
||||
check("vuln-scanner.yaml" in (ROOT / "clusters" / "base" / "kustomization.yaml").read_text(),
|
||||
"(1) clusters/base/kustomization.yaml lists vuln-scanner.yaml")
|
||||
check("vuln-scanner" not in (ROOT / "platform" / "kustomization.yaml").read_text(),
|
||||
"(1) platform/kustomization.yaml must not include vuln-scanner (the gating tier)")
|
||||
for f in sorted((ROOT / "clusters").rglob("*.yaml")):
|
||||
for d in docs(f):
|
||||
for dep in (d.get("spec") or {}).get("dependsOn") or []:
|
||||
check(not (isinstance(dep, dict) and dep.get("name") == "vuln-scanner"),
|
||||
f"(1) {f.relative_to(ROOT)} {d.get('metadata', {}).get('name')} dependsOn vuln-scanner — "
|
||||
"a broken scanner must never stall anything")
|
||||
for env in ("dev", "prod"):
|
||||
text = (ROOT / "clusters" / env / "kustomization.yaml").read_text()
|
||||
check(re.search(r"name:\s*vuln-scanner\s*\n\s*namespace:\s*flux-system\s*\n\s*spec:\s*\n\s*decryption:",
|
||||
text) is not None,
|
||||
f"(1) clusters/{env}/kustomization.yaml carries the vuln-scanner SOPS decryption patch")
|
||||
|
||||
# Network shape
|
||||
np_docs = docs(np_path)
|
||||
dd = [d for d in np_docs if d.get("kind") == "NetworkPolicy" and d["metadata"]["name"] == "default-deny"]
|
||||
check(len(dd) == 1 and set(dd[0]["spec"]["policyTypes"]) == {"Ingress", "Egress"} and dd[0]["spec"].get("ingress") == [],
|
||||
"(4) default-deny baseline denies both directions with no ingress")
|
||||
ns = next(d for d in docs(ROOT / "platform" / "vuln-scanner" / "namespace.yaml") if d.get("kind") == "Namespace")
|
||||
check("homelab.keithjackson.dev/app-namespace" not in (ns["metadata"].get("labels") or {}),
|
||||
"(4) the namespace must not be app-namespace=true (the default-deny would be double-owned)")
|
||||
|
||||
|
||||
def cnp(name):
|
||||
return next((d for d in np_docs if d.get("kind") == "CiliumNetworkPolicy" and d["metadata"]["name"] == name), None)
|
||||
|
||||
|
||||
def ports_of(rule):
|
||||
return {(p.get("port"), p.get("protocol", "TCP")) for tp in rule.get("toPorts") or [] for p in tp.get("ports") or []}
|
||||
|
||||
|
||||
def eval_ingress(policy):
|
||||
bad = []
|
||||
rules = policy["spec"].get("ingress") or []
|
||||
if len(rules) != 3:
|
||||
bad.append(f"expected exactly 3 ingress rules, found {len(rules)}")
|
||||
seen = set()
|
||||
for r in rules:
|
||||
ports = ports_of(r)
|
||||
if r.get("fromEntities") == ["ingress"] and ports == {("4180", "TCP")}:
|
||||
seen.add("gateway")
|
||||
elif len(r.get("fromEndpoints") or []) == 1:
|
||||
sel = r["fromEndpoints"][0].get("matchLabels") or {}
|
||||
if (sel.get("io.kubernetes.pod.namespace") == "gitea-actions"
|
||||
and sel.get("app.kubernetes.io/name") == "act-runner" and ports == {("8081", "TCP")}):
|
||||
seen.add("ci")
|
||||
elif (sel.get("io.kubernetes.pod.namespace") == "monitoring"
|
||||
and sel.get("app.kubernetes.io/name") == "prometheus" and ports == {("9090", "TCP")}):
|
||||
seen.add("prometheus")
|
||||
else:
|
||||
bad.append(f"unexpected ingress rule {r}")
|
||||
else:
|
||||
bad.append(f"unexpected ingress rule {r}")
|
||||
missing = {"gateway", "ci", "prometheus"} - seen
|
||||
if missing:
|
||||
bad.append(f"ingress paths missing: {sorted(missing)}")
|
||||
return bad
|
||||
|
||||
|
||||
def eval_egress(policy):
|
||||
bad = []
|
||||
rules = policy["spec"].get("egress") or []
|
||||
if len(rules) != 4:
|
||||
bad.append(f"expected exactly 4 egress rules, found {len(rules)}")
|
||||
world = 0
|
||||
for r in rules:
|
||||
ents = set(r.get("toEntities") or [])
|
||||
if ents & FORBIDDEN_ENTITIES:
|
||||
bad.append(f"egress admits forbidden entity {sorted(ents & FORBIDDEN_ENTITIES)}")
|
||||
if ents == {"kube-apiserver"} and not r.get("toPorts"):
|
||||
continue
|
||||
if ents == {"ingress"} and ports_of(r) == {("443", "TCP")}:
|
||||
continue
|
||||
if r.get("toEndpoints"):
|
||||
sel = r["toEndpoints"][0].get("matchLabels") or {}
|
||||
if not (len(r["toEndpoints"]) == 1 and sel.get("io.kubernetes.pod.namespace") == "keycloak"
|
||||
and sel.get("app") == "keycloak"):
|
||||
bad.append(f"egress to unexpected endpoints {r}")
|
||||
continue
|
||||
if r.get("toCIDRSet"):
|
||||
world += 1
|
||||
sets = r["toCIDRSet"]
|
||||
if len(sets) != 1 or sets[0].get("cidr") != "0.0.0.0/0":
|
||||
bad.append(f"unexpected CIDR rule {r}")
|
||||
continue
|
||||
if not PRIVATE_RANGES.issubset(set(sets[0].get("except") or [])):
|
||||
bad.append(f"the world CIDR rule does not exclude {sorted(PRIVATE_RANGES - set(sets[0].get('except') or []))}")
|
||||
if ports_of(r) != {("443", "TCP")}:
|
||||
bad.append(f"the world CIDR rule is not tcp/443 only ({sorted(ports_of(r))})")
|
||||
continue
|
||||
if r.get("toCIDR"):
|
||||
bad.append(f"egress names a bare CIDR rule {r}")
|
||||
continue
|
||||
if not ents:
|
||||
bad.append(f"unexpected egress rule {r}")
|
||||
if world != 1:
|
||||
bad.append(f"expected exactly one world CIDR rule, found {world}")
|
||||
return bad
|
||||
|
||||
|
||||
ing, eg = cnp("vuln-scanner-allow-ingress"), cnp("vuln-scanner-allow-egress")
|
||||
check(ing is not None, "(2) the ingress CiliumNetworkPolicy exists")
|
||||
check(eg is not None, "(3) the egress CiliumNetworkPolicy exists")
|
||||
if ing is not None:
|
||||
for v in eval_ingress(ing):
|
||||
check(False, f"(2) {v}")
|
||||
if eg is not None:
|
||||
for v in eval_egress(eg):
|
||||
check(False, f"(3) {v}")
|
||||
check(len([d for d in np_docs if d.get("kind") == "CiliumNetworkPolicy"]) == 2,
|
||||
"(3) exactly two CiliumNetworkPolicies, so no other path is added silently")
|
||||
|
||||
# (5) non-vacuity
|
||||
if ing is not None and eg is not None:
|
||||
def m_ing_open(p): p["spec"]["ingress"].append({"fromEntities": ["world"], "toPorts": [{"ports": [{"port": "8081", "protocol": "TCP"}]}]})
|
||||
def m_ing_ns(p): p["spec"]["ingress"][1]["fromEndpoints"][0]["matchLabels"]["io.kubernetes.pod.namespace"] = "default"
|
||||
def m_ing_port(p): p["spec"]["ingress"][0]["toPorts"][0]["ports"][0]["port"] = "8080"
|
||||
for name, fn in (("extra world ingress", m_ing_open), ("CI rule moved to another namespace", m_ing_ns),
|
||||
("Gateway reaches the page port", m_ing_port)):
|
||||
m = copy.deepcopy(ing)
|
||||
fn(m)
|
||||
check(bool(eval_ingress(m)), f"(5) ingress evaluator missed mutant: {name}")
|
||||
def m_eg_world(p): p["spec"]["egress"].append({"toEntities": ["world"]})
|
||||
def m_eg_lan(p):
|
||||
for r in p["spec"]["egress"]:
|
||||
if r.get("toCIDRSet"):
|
||||
r["toCIDRSet"][0]["except"] = ["10.0.0.0/8"]
|
||||
def m_eg_port(p):
|
||||
for r in p["spec"]["egress"]:
|
||||
if r.get("toCIDRSet"):
|
||||
r["toPorts"][0]["ports"].append({"port": "80", "protocol": "TCP"})
|
||||
def m_eg_endpoint(p):
|
||||
for r in p["spec"]["egress"]:
|
||||
if r.get("toEndpoints"):
|
||||
r["toEndpoints"][0]["matchLabels"]["io.kubernetes.pod.namespace"] = "default"
|
||||
for name, fn in (("world entity added", m_eg_world), ("LAN not excluded", m_eg_lan),
|
||||
("port 80 added", m_eg_port), ("endpoints moved", m_eg_endpoint)):
|
||||
m = copy.deepcopy(eg)
|
||||
fn(m)
|
||||
check(bool(eval_egress(m)), f"(5) egress evaluator missed mutant: {name}")
|
||||
|
||||
if CHECKS < 18:
|
||||
FAILURES.append(f"only {CHECKS} assertions ran — the check degenerated")
|
||||
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
|
||||
@@ -0,0 +1,34 @@
|
||||
# =============================================================================
|
||||
# L0 STATIC — the CVE scanner's supply chain and privileges are what docs/17 "Security of the scanner
|
||||
# itself" says (R-VULN-09). Offline, no cluster.
|
||||
#
|
||||
# WHY SCOPE, NOT REQUIREMENTS: the property spans platform/vuln-scanner and renovate.json5. A
|
||||
# requirement-scoped selection would miss a Renovate edit that quietly lets the scanner auto-merge.
|
||||
# =============================================================================
|
||||
apiVersion: chainsaw.kyverno.io/v1alpha1
|
||||
kind: Test
|
||||
metadata:
|
||||
name: vuln-scanner-supply-chain
|
||||
labels:
|
||||
conformance.home.k3s/R-VULN-09: "true"
|
||||
annotations:
|
||||
conformance.home.k3s/requirements: "R-VULN-09"
|
||||
conformance.home.k3s/layer: "L0"
|
||||
conformance.home.k3s/scope: "whole-repo"
|
||||
conformance.home.k3s/safety-critical: "false"
|
||||
conformance.home.k3s/executable: "true"
|
||||
conformance.home.k3s/intent: >-
|
||||
The scanner image is pinned by digest and is not a banned Trivy version; Renovate never
|
||||
auto-merges it and proposes it only after 7 days; its RBAC holds get/list/watch only and no
|
||||
Secret; the pod is unprivileged with bounded resources and a size-limited cache. Each is
|
||||
proven by mutation.
|
||||
spec:
|
||||
steps:
|
||||
- name: assert-supply-chain-and-privileges
|
||||
description: >-
|
||||
Parse platform/vuln-scanner/deployment.yaml, rbac.yaml and renovate.json5, assert the
|
||||
digest/version/Renovate/RBAC/security-context contract, and run each evaluator on mutated
|
||||
copies to prove it can fail.
|
||||
try:
|
||||
- script:
|
||||
content: bash ./check.sh
|
||||
@@ -0,0 +1,298 @@
|
||||
#!/usr/bin/env bash
|
||||
# =============================================================================
|
||||
# L0 static check (offline, no cluster) — vuln-scanner-supply-chain
|
||||
# Serves: R-VULN-09 (docs/17 "Security of the scanner itself").
|
||||
#
|
||||
# GIVEN the repository as authored,
|
||||
# THEN (1) the Trivy image is `ghcr.io/aquasecurity/trivy:<ver>@sha256:<64hex>` — by digest, and a
|
||||
# tag so Renovate has something to follow;
|
||||
# (2) that version is not 0.69.4/0.69.5/0.69.6 (the compromised March 2026 releases);
|
||||
# (3) every pod image is digest-pinned except the oauth2-proxy sidecar, which follows the
|
||||
# per-app sidecar pattern so one Renovate branch moves them all;
|
||||
# (4) renovate.json5 keeps Trivy in the blast-radius list, automerge:false, minimumReleaseAge
|
||||
# of at least 7 days, and no later rule re-enables automerge for it;
|
||||
# (5) rbac.yaml: only ClusterRole/ClusterRoleBinding to the scanner SA, only get/list/watch, no
|
||||
# Secret, no wildcard, and only the resources the inventory needs;
|
||||
# (6) the pod: non-root, read-only rootfs, drop ALL, seccomp RuntimeDefault, no host
|
||||
# namespaces or hostPath, CPU and memory limits on every container, an ephemeral-storage
|
||||
# limit on the scanner, and a size-limited /cache emptyDir.
|
||||
# AND (7) each evaluator is run on a mutated copy and must flag it (non-vacuity).
|
||||
# =============================================================================
|
||||
set -euo pipefail
|
||||
DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
ROOT="$(cd "${DIR}/../../../.." && pwd)"
|
||||
exec python3 - "$ROOT" <<'PY'
|
||||
import copy, json, re, sys
|
||||
from pathlib import Path
|
||||
import yaml
|
||||
|
||||
ROOT = Path(sys.argv[1])
|
||||
NAME = "vuln-scanner-supply-chain"
|
||||
FAILURES, CHECKS = [], 0
|
||||
COMP = ROOT / "platform" / "vuln-scanner"
|
||||
BANNED = {"0.69.4", "0.69.5", "0.69.6"}
|
||||
DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$")
|
||||
ALLOWED_VERBS = {"get", "list", "watch"}
|
||||
ALLOWED_RESOURCES = {
|
||||
"": {"pods", "nodes"},
|
||||
"apps": {"replicasets", "deployments", "statefulsets", "daemonsets"},
|
||||
"batch": {"jobs", "cronjobs"},
|
||||
"kubevirt.io": {"virtualmachineinstances"},
|
||||
}
|
||||
|
||||
|
||||
def check(cond, desc):
|
||||
global CHECKS
|
||||
CHECKS += 1
|
||||
if not cond:
|
||||
FAILURES.append(desc)
|
||||
|
||||
|
||||
def docs(path):
|
||||
return [d for d in yaml.safe_load_all(path.read_text()) if isinstance(d, dict)]
|
||||
|
||||
|
||||
for f in ("deployment.yaml", "rbac.yaml"):
|
||||
if not (COMP / f).is_file():
|
||||
print(f"CHECK FAIL ({NAME}): platform/vuln-scanner/{f} is absent", file=sys.stderr)
|
||||
raise SystemExit(1)
|
||||
|
||||
|
||||
def split_ref(image):
|
||||
# A digest suffix, then a tag after the last ':' (only when it holds no '/', so a registry port
|
||||
# stays in the repository part).
|
||||
ref, _, digest = image.partition("@")
|
||||
head, colon, tail = ref.rpartition(":")
|
||||
if colon and "/" not in tail:
|
||||
return head, tail, digest or None
|
||||
return ref, None, digest or None
|
||||
|
||||
|
||||
def containers(pod):
|
||||
return (pod.get("initContainers") or []) + (pod.get("containers") or [])
|
||||
|
||||
|
||||
def eval_images(pod):
|
||||
bad = []
|
||||
trivy = [c for c in containers(pod) if split_ref(c["image"])[0] == "ghcr.io/aquasecurity/trivy"]
|
||||
if len(trivy) != 1:
|
||||
bad.append(f"expected one Trivy image, found {len(trivy)}")
|
||||
for c in containers(pod):
|
||||
repo, tag, digest = split_ref(c["image"])
|
||||
if repo == "quay.io/oauth2-proxy/oauth2-proxy":
|
||||
if not tag or tag == "latest":
|
||||
bad.append("oauth2-proxy sidecar has no version tag")
|
||||
continue
|
||||
if not digest or not DIGEST.match(digest):
|
||||
bad.append(f"{c['name']} image {c['image']} is not pinned by digest")
|
||||
if not tag:
|
||||
bad.append(f"{c['name']} image {c['image']} has no tag for Renovate to follow")
|
||||
if repo == "ghcr.io/aquasecurity/trivy" and tag in BANNED:
|
||||
bad.append(f"Trivy {tag} is a banned (compromised) version")
|
||||
return bad
|
||||
|
||||
|
||||
def strip_json5(text):
|
||||
# Drop // comments outside strings, then trailing commas.
|
||||
out, i, in_str, esc = [], 0, False, False
|
||||
while i < len(text):
|
||||
ch = text[i]
|
||||
if in_str:
|
||||
out.append(ch)
|
||||
if esc:
|
||||
esc = False
|
||||
elif ch == "\\":
|
||||
esc = True
|
||||
elif ch == '"':
|
||||
in_str = False
|
||||
i += 1
|
||||
continue
|
||||
if ch == '"':
|
||||
in_str = True
|
||||
out.append(ch)
|
||||
i += 1
|
||||
continue
|
||||
if text.startswith("//", i):
|
||||
while i < len(text) and text[i] != "\n":
|
||||
i += 1
|
||||
continue
|
||||
out.append(ch)
|
||||
i += 1
|
||||
return re.sub(r",(\s*[}\]])", r"\1", "".join(out))
|
||||
|
||||
|
||||
TRIVY_PKG = "ghcr.io/aquasecurity/trivy"
|
||||
VERSION_UPDATE_RULE_RE = re.compile(r"^(\d+)\s*(day|days)$")
|
||||
|
||||
|
||||
def eval_renovate(rules):
|
||||
bad = []
|
||||
trivy_rules = [r for r in rules if TRIVY_PKG in (r.get("matchPackageNames") or [])]
|
||||
if not trivy_rules:
|
||||
bad.append("no packageRule matches ghcr.io/aquasecurity/trivy (blast-radius list)")
|
||||
return bad
|
||||
if not any(r.get("automerge") is False and "blast-radius" in (r.get("addLabels") or [])
|
||||
for r in trivy_rules):
|
||||
bad.append("Trivy is not in a blast-radius rule with automerge:false")
|
||||
ages = [r["minimumReleaseAge"] for r in trivy_rules if "minimumReleaseAge" in r]
|
||||
if not ages or not any(VERSION_UPDATE_RULE_RE.match(a) and int(VERSION_UPDATE_RULE_RE.match(a).group(1)) >= 7
|
||||
for a in ages):
|
||||
bad.append(f"Trivy has no minimumReleaseAge of at least 7 days (found {ages})")
|
||||
# Renovate applies packageRules in order, later wins; the last rule that names automerge for
|
||||
# Trivy (or matches it generically by update type) must be false.
|
||||
last = None
|
||||
for r in rules:
|
||||
if "automerge" in r and (TRIVY_PKG in (r.get("matchPackageNames") or [])
|
||||
or set(r) <= {"description", "matchUpdateTypes", "automerge", "automergeType", "platformAutomerge"}):
|
||||
last = r
|
||||
if last is None or last.get("automerge") is not False:
|
||||
bad.append("the last packageRule deciding automerge for Trivy leaves it true")
|
||||
return bad
|
||||
|
||||
|
||||
def eval_rbac(rbac):
|
||||
bad = []
|
||||
croles = [d for d in rbac if d.get("kind") == "ClusterRole"]
|
||||
if len(croles) != 1:
|
||||
bad.append(f"expected one ClusterRole, found {len(croles)}")
|
||||
return bad
|
||||
for rule in croles[0].get("rules") or []:
|
||||
verbs = set(rule.get("verbs") or [])
|
||||
if verbs - ALLOWED_VERBS:
|
||||
bad.append(f"rule grants non-read verbs {sorted(verbs - ALLOWED_VERBS)}")
|
||||
if "*" in verbs:
|
||||
bad.append("rule grants the verb wildcard")
|
||||
for group in rule.get("apiGroups") or []:
|
||||
if group == "*" or group not in ALLOWED_RESOURCES:
|
||||
bad.append(f"rule names apiGroup {group!r}")
|
||||
continue
|
||||
for res in rule.get("resources") or []:
|
||||
if res == "*" or res not in ALLOWED_RESOURCES[group]:
|
||||
bad.append(f"rule grants resource {group}/{res}")
|
||||
if "secrets" in (rule.get("resources") or []):
|
||||
bad.append("rule grants Secret access")
|
||||
kinds = {d.get("kind") for d in rbac}
|
||||
if kinds - {"ServiceAccount", "ClusterRole", "ClusterRoleBinding"}:
|
||||
bad.append(f"unexpected RBAC kinds {sorted(kinds)}")
|
||||
for b in (d for d in rbac if d.get("kind") == "ClusterRoleBinding"):
|
||||
if (b.get("roleRef") or {}).get("name") != croles[0]["metadata"]["name"]:
|
||||
bad.append("the ClusterRoleBinding does not bind the scanner's ClusterRole")
|
||||
return bad
|
||||
|
||||
|
||||
def eval_pod(pod):
|
||||
bad = []
|
||||
psc = pod.get("securityContext") or {}
|
||||
if psc.get("runAsNonRoot") is not True:
|
||||
bad.append("pod runAsNonRoot is not true")
|
||||
if (psc.get("seccompProfile") or {}).get("type") != "RuntimeDefault":
|
||||
bad.append("pod seccompProfile is not RuntimeDefault")
|
||||
for f in ("hostNetwork", "hostPID", "hostIPC"):
|
||||
if pod.get(f):
|
||||
bad.append(f"pod sets {f}")
|
||||
if any("hostPath" in v for v in pod.get("volumes") or []):
|
||||
bad.append("pod mounts a hostPath volume")
|
||||
for c in containers(pod):
|
||||
sc = c.get("securityContext") or {}
|
||||
if sc.get("readOnlyRootFilesystem") is not True:
|
||||
bad.append(f"{c['name']} root filesystem is writable")
|
||||
if sc.get("allowPrivilegeEscalation") is not False:
|
||||
bad.append(f"{c['name']} allows privilege escalation")
|
||||
if "ALL" not in ((sc.get("capabilities") or {}).get("drop") or []):
|
||||
bad.append(f"{c['name']} does not drop ALL capabilities")
|
||||
limits = (c.get("resources") or {}).get("limits") or {}
|
||||
for lim in ("cpu", "memory"):
|
||||
if lim not in limits:
|
||||
bad.append(f"{c['name']} has no {lim} limit")
|
||||
if c["name"] == "scanner" and "ephemeral-storage" not in limits:
|
||||
bad.append("the scanner has no ephemeral-storage limit")
|
||||
vols = {v["name"]: v for v in pod.get("volumes") or []}
|
||||
if not (vols.get("cache", {}).get("emptyDir") or {}).get("sizeLimit"):
|
||||
bad.append("the /cache emptyDir has no sizeLimit (Longhorn or etcd must never hold it)")
|
||||
return bad
|
||||
|
||||
|
||||
dep = next(d for d in docs(COMP / "deployment.yaml") if d.get("kind") == "Deployment")
|
||||
pod = dep["spec"]["template"]["spec"]
|
||||
for v in eval_images(pod) + eval_pod(pod):
|
||||
check(False, f"(1-3,6) {v}")
|
||||
check(dep["spec"].get("replicas") == 1, "(6) one replica (one RWO report store)")
|
||||
check((dep["spec"].get("strategy") or {}).get("type") == "Recreate", "(6) Recreate strategy")
|
||||
|
||||
renovate = json.loads(strip_json5((ROOT / "renovate.json5").read_text()))
|
||||
rules = renovate.get("packageRules") or []
|
||||
check(len(rules) > 5, "(4) parsed renovate.json5 packageRules (parser degenerated)")
|
||||
for v in eval_renovate(rules):
|
||||
check(False, f"(4) {v}")
|
||||
|
||||
rbac = docs(COMP / "rbac.yaml")
|
||||
for v in eval_rbac(rbac):
|
||||
check(False, f"(5) {v}")
|
||||
|
||||
# (7) non-vacuity
|
||||
def mutated(fn, base):
|
||||
m = copy.deepcopy(base)
|
||||
fn(m)
|
||||
return m
|
||||
|
||||
# Image mutants
|
||||
def img_digest(p): p["initContainers"][0]["image"] = "ghcr.io/aquasecurity/trivy:0.76.1"
|
||||
def img_banned(p): p["initContainers"][0]["image"] = re.sub(r":[0-9.]+@", ":0.69.5@", p["initContainers"][0]["image"])
|
||||
def img_nopin(p): p["containers"][0]["image"] = "docker.io/library/python:3.14.9-alpine"
|
||||
for name, fn in (("digest stripped from Trivy", img_digest), ("banned Trivy version", img_banned),
|
||||
("python image unpinned", img_nopin)):
|
||||
check(bool(eval_images(mutated(fn, pod))), f"(7) image evaluator missed mutant: {name}")
|
||||
|
||||
# RBAC mutants
|
||||
def rbac_secret(r):
|
||||
role = next(d for d in r if d.get("kind") == "ClusterRole")
|
||||
role["rules"][0]["resources"].append("secrets")
|
||||
def rbac_write(r):
|
||||
role = next(d for d in r if d.get("kind") == "ClusterRole")
|
||||
role["rules"][0]["verbs"].append("delete")
|
||||
def rbac_wild(r):
|
||||
role = next(d for d in r if d.get("kind") == "ClusterRole")
|
||||
role["rules"][1]["resources"] = ["*"]
|
||||
for name, fn in (("Secret resource granted", rbac_secret), ("write verb granted", rbac_write),
|
||||
("resource wildcard granted", rbac_wild)):
|
||||
check(bool(eval_rbac(mutated(fn, rbac))), f"(7) RBAC evaluator missed mutant: {name}")
|
||||
|
||||
# Renovate mutants
|
||||
def ren_auto(r):
|
||||
for x in r:
|
||||
if TRIVY_PKG in (x.get("matchPackageNames") or []):
|
||||
x["automerge"] = True
|
||||
def ren_age(r):
|
||||
for x in r:
|
||||
if TRIVY_PKG in (x.get("matchPackageNames") or []) and "minimumReleaseAge" in x:
|
||||
x["minimumReleaseAge"] = "2 days"
|
||||
def ren_drop(r):
|
||||
r[:] = [x for x in r if TRIVY_PKG not in (x.get("matchPackageNames") or [])]
|
||||
def ren_later(r):
|
||||
r.append({"matchUpdateTypes": ["digest"], "automerge": True})
|
||||
for name, fn in (("Trivy automerge flipped true", ren_auto), ("minimumReleaseAge shortened", ren_age),
|
||||
("Trivy removed from Renovate rules", ren_drop), ("later auto-merge rule appended", ren_later)):
|
||||
check(bool(eval_renovate(mutated(fn, rules))), f"(7) Renovate evaluator missed mutant: {name}")
|
||||
|
||||
# Pod mutants
|
||||
def pod_root(p): p["securityContext"]["runAsNonRoot"] = False
|
||||
def pod_host(p): p["hostNetwork"] = True
|
||||
def pod_rw(p): p["containers"][0]["securityContext"]["readOnlyRootFilesystem"] = False
|
||||
def pod_nolim(p): del p["containers"][0]["resources"]["limits"]["memory"]
|
||||
def pod_nocap(p):
|
||||
for v in p["volumes"]:
|
||||
if v["name"] == "cache":
|
||||
v["emptyDir"].pop("sizeLimit", None)
|
||||
for name, fn in (("runAsNonRoot off", pod_root), ("hostNetwork on", pod_host), ("rw rootfs", pod_rw),
|
||||
("no memory limit", pod_nolim), ("no cache sizeLimit", pod_nocap)):
|
||||
check(bool(eval_pod(mutated(fn, pod))), f"(7) pod evaluator missed mutant: {name}")
|
||||
|
||||
if CHECKS < 15:
|
||||
FAILURES.append(f"only {CHECKS} assertions ran — the check degenerated")
|
||||
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
|
||||
@@ -0,0 +1,31 @@
|
||||
# =============================================================================
|
||||
# L0 STATIC — the CVE triage bot workflow fails loudly and is shaped like Renovate's (R-VULN-05,
|
||||
# R-VULN-07, docs/17 "The triage bot"). Offline, no cluster. Depends on .gitea/workflows/cve-triage.yaml
|
||||
# (the BOT's file); fails until that file is integrated.
|
||||
# =============================================================================
|
||||
apiVersion: chainsaw.kyverno.io/v1alpha1
|
||||
kind: Test
|
||||
metadata:
|
||||
name: vuln-triage-bot-workflow-guards
|
||||
labels:
|
||||
conformance.home.k3s/R-VULN-05: "true"
|
||||
conformance.home.k3s/R-VULN-07: "true"
|
||||
annotations:
|
||||
conformance.home.k3s/requirements: "R-VULN-05, R-VULN-07"
|
||||
conformance.home.k3s/layer: "L0"
|
||||
conformance.home.k3s/scope: "whole-repo"
|
||||
conformance.home.k3s/safety-critical: "false"
|
||||
conformance.home.k3s/executable: "true"
|
||||
conformance.home.k3s/intent: >-
|
||||
The bot workflow runs on the 07:15 cron, has a workflow_dispatch dry-run input, never runs
|
||||
concurrently with itself, and a step fails the job hard when RENOVATE_TOKEN is empty.
|
||||
Proven by mutation.
|
||||
spec:
|
||||
steps:
|
||||
- name: assert-workflow-guards
|
||||
description: >-
|
||||
Parse .gitea/workflows/cve-triage.yaml and assert the trigger, concurrency and hard
|
||||
credential failure, then run the evaluator on mutated copies to prove it can fail.
|
||||
try:
|
||||
- script:
|
||||
content: bash ./check.sh
|
||||
@@ -0,0 +1,102 @@
|
||||
#!/usr/bin/env bash
|
||||
# =============================================================================
|
||||
# L0 static check (offline, no cluster) — vuln-triage-bot-workflow-guards
|
||||
# Serves: R-VULN-05 (runs on a schedule, fails loudly), R-VULN-07 (bot conventions, docs/17).
|
||||
#
|
||||
# GIVEN .gitea/workflows/cve-triage.yaml,
|
||||
# THEN (1) it runs on `15 7 * * *` and has a workflow_dispatch `dry_run` input;
|
||||
# (2) concurrency is set (the workflow never runs against itself);
|
||||
# (3) a step fails the job HARD when RENOVATE_TOKEN is empty (exit non-zero, not skip);
|
||||
# AND (4) the evaluator is run on mutated copies and must flag each (non-vacuity).
|
||||
# A missing workflow file is a FAIL: this is whole-repo scoped and has no skip path.
|
||||
# =============================================================================
|
||||
set -euo pipefail
|
||||
DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
ROOT="$(cd "${DIR}/../../../.." && pwd)"
|
||||
exec python3 - "$ROOT" <<'PY'
|
||||
import copy, re, sys
|
||||
from pathlib import Path
|
||||
import yaml
|
||||
|
||||
ROOT = Path(sys.argv[1])
|
||||
NAME = "vuln-triage-bot-workflow-guards"
|
||||
FAILURES, CHECKS = [], 0
|
||||
WF = ROOT / ".gitea" / "workflows" / "cve-triage.yaml"
|
||||
|
||||
|
||||
def check(cond, desc):
|
||||
global CHECKS
|
||||
CHECKS += 1
|
||||
if not cond:
|
||||
FAILURES.append(desc)
|
||||
|
||||
|
||||
if not WF.is_file():
|
||||
print(f"CHECK FAIL ({NAME}): .gitea/workflows/cve-triage.yaml is absent (the BOT's deliverable); "
|
||||
f"whether it fails on a missing credential cannot be asserted without it", file=sys.stderr)
|
||||
raise SystemExit(1)
|
||||
doc = yaml.safe_load(WF.read_text())
|
||||
|
||||
|
||||
def evaluate(doc):
|
||||
bad = []
|
||||
triggers = doc.get(True) if True in doc else doc.get("on") # YAML reads `on:` as boolean True
|
||||
if not isinstance(triggers, dict):
|
||||
return ["no triggers parsed"]
|
||||
if "15 7 * * *" not in [s.get("cron") for s in triggers.get("schedule") or [] if isinstance(s, dict)]:
|
||||
bad.append("no schedule with cron 15 7 * * *")
|
||||
inputs = ((triggers.get("workflow_dispatch") or {}).get("inputs") or {})
|
||||
if "dry_run" not in inputs:
|
||||
bad.append("workflow_dispatch has no dry_run input")
|
||||
conc = doc.get("concurrency")
|
||||
jobs = doc.get("jobs") or {}
|
||||
if not conc and not any(isinstance(j, dict) and j.get("concurrency") for j in jobs.values()):
|
||||
bad.append("no concurrency set (the bot must not run against itself)")
|
||||
hard_fail = False
|
||||
for j in jobs.values():
|
||||
for step in (j.get("steps") or []) if isinstance(j, dict) else []:
|
||||
run = str(step.get("run", ""))
|
||||
if ("RENOVATE_TOKEN" in run and re.search(r"(-z|!\s*-n)[^\n]*RENOVATE_TOKEN|RENOVATE_TOKEN[^\n]*(-z)", run)
|
||||
and re.search(r"exit\s+[1-9]", run)):
|
||||
hard_fail = True
|
||||
if not hard_fail:
|
||||
bad.append("no step fails hard (exit non-zero) when RENOVATE_TOKEN is empty")
|
||||
return bad
|
||||
|
||||
|
||||
for v in evaluate(doc):
|
||||
check(False, f"(1-3) {v}")
|
||||
check(isinstance(doc.get("jobs"), dict) and doc["jobs"], "(1) the workflow declares jobs")
|
||||
|
||||
# (4) non-vacuity
|
||||
def mutate(fn):
|
||||
m = copy.deepcopy(doc)
|
||||
fn(m)
|
||||
return m
|
||||
|
||||
def trig(m):
|
||||
return m.get(True) if True in m else m.get("on")
|
||||
|
||||
def no_cron(m): trig(m).pop("schedule", None)
|
||||
def no_dry(m): trig(m)["workflow_dispatch"] = {}
|
||||
def no_conc(m):
|
||||
m.pop("concurrency", None)
|
||||
for j in m.get("jobs", {}).values():
|
||||
if isinstance(j, dict):
|
||||
j.pop("concurrency", None)
|
||||
def no_guard(m):
|
||||
for j in m.get("jobs", {}).values():
|
||||
for step in j.get("steps") or []:
|
||||
if "RENOVATE_TOKEN" in str(step.get("run", "")):
|
||||
step["run"] = 'echo "::warning::RENOVATE_TOKEN unset, skipping"'
|
||||
for name, fn in (("cron removed", no_cron), ("dry_run input removed", no_dry),
|
||||
("concurrency removed", no_conc), ("credential check softened", no_guard)):
|
||||
check(bool(evaluate(mutate(fn))), f"(4) evaluator missed mutant: {name}")
|
||||
|
||||
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
|
||||
@@ -0,0 +1,35 @@
|
||||
# =============================================================================
|
||||
# L2 BEHAVIORAL SCAFFOLD — the internal API is unreachable from an unrelated namespace (R-VULN-09)
|
||||
# Catalog: docs/17 "Acceptance gates" (L2).
|
||||
#
|
||||
# A pod in an unrelated namespace cannot reach vuln-scanner :8081 or :9090, while the CI runner identity can reach :8081.
|
||||
#
|
||||
# STATUS: scaffold only. It asserts nothing and DOES NOT EXECUTE IN CI: vars.CLUSTER_AVAILABLE is unset,
|
||||
# so L2/L3 conformance does not run. Coverage from this file is a named gap, not evidence.
|
||||
# =============================================================================
|
||||
apiVersion: chainsaw.kyverno.io/v1alpha1
|
||||
kind: Test
|
||||
metadata:
|
||||
name: vuln-internal-api-isolated
|
||||
labels:
|
||||
conformance.home.k3s/R-VULN-09: "true"
|
||||
annotations:
|
||||
conformance.home.k3s/requirements: "R-VULN-09"
|
||||
conformance.home.k3s/layer: "L2"
|
||||
conformance.home.k3s/safety-critical: "false"
|
||||
conformance.home.k3s/intent: >-
|
||||
A pod in an unrelated namespace cannot reach vuln-scanner :8081 or :9090, while the CI runner identity can reach :8081.
|
||||
spec:
|
||||
steps:
|
||||
- name: given-an-unrelated-namespace
|
||||
try:
|
||||
- script:
|
||||
content: 'echo "TODO: create a throwaway namespace and probe pod"'
|
||||
- name: when-connecting
|
||||
try:
|
||||
- script:
|
||||
content: 'echo "TODO: attempt tcp to vuln-scanner:8081 and :9090"'
|
||||
- name: then-denied
|
||||
try:
|
||||
- script:
|
||||
content: 'echo "TODO: assert a timeout on both, and success from the act-runner identity on :8081"'
|
||||
@@ -0,0 +1,36 @@
|
||||
# =============================================================================
|
||||
# L2 BEHAVIORAL SCAFFOLD — new workload digest appears in the report (R-VULN-01, R-VULN-02)
|
||||
# Catalog: docs/17 "Acceptance gates" (L2).
|
||||
#
|
||||
# A newly started workload's digest appears in report.json, mapped to that workload, within the 2-hour window.
|
||||
#
|
||||
# STATUS: scaffold only. It asserts nothing and DOES NOT EXECUTE IN CI: vars.CLUSTER_AVAILABLE is unset,
|
||||
# so L2/L3 conformance does not run. Coverage from this file is a named gap, not evidence.
|
||||
# =============================================================================
|
||||
apiVersion: chainsaw.kyverno.io/v1alpha1
|
||||
kind: Test
|
||||
metadata:
|
||||
name: vuln-new-digest-reported
|
||||
labels:
|
||||
conformance.home.k3s/R-VULN-01: "true"
|
||||
conformance.home.k3s/R-VULN-02: "true"
|
||||
annotations:
|
||||
conformance.home.k3s/requirements: "R-VULN-01, R-VULN-02"
|
||||
conformance.home.k3s/layer: "L2"
|
||||
conformance.home.k3s/safety-critical: "false"
|
||||
conformance.home.k3s/intent: >-
|
||||
A newly started workload's digest appears in report.json, mapped to that workload, within the 2-hour window.
|
||||
spec:
|
||||
steps:
|
||||
- name: given-a-new-workload
|
||||
try:
|
||||
- script:
|
||||
content: 'echo "TODO: start a pod from an image digest not yet in the report"'
|
||||
- name: when-the-scanner-ticks
|
||||
try:
|
||||
- script:
|
||||
content: 'echo "TODO: wait, bounded by the 2h SLO, polling GET /report.json from the CI runner identity"'
|
||||
- name: then-digest-is-reported
|
||||
try:
|
||||
- script:
|
||||
content: 'echo "TODO: assert a target with that digest and the workload mapping, state not pending past 2h"'
|
||||
@@ -0,0 +1,39 @@
|
||||
# =============================================================================
|
||||
# L2 BEHAVIORAL SCAFFOLD — the CVE page requires a platform-admins login (R-VULN-04)
|
||||
# Catalog: docs/17 "Acceptance gates" (L2).
|
||||
#
|
||||
# An unauthenticated request to cve.<domain> is redirected to Keycloak, and a logged-in user outside platform-admins gets 403.
|
||||
#
|
||||
# STATUS: scaffold only. It asserts nothing and DOES NOT EXECUTE IN CI: vars.CLUSTER_AVAILABLE is unset,
|
||||
# so L2/L3 conformance does not run. Coverage from this file is a named gap, not evidence.
|
||||
# =============================================================================
|
||||
apiVersion: chainsaw.kyverno.io/v1alpha1
|
||||
kind: Test
|
||||
metadata:
|
||||
name: vuln-page-requires-admin
|
||||
labels:
|
||||
conformance.home.k3s/R-VULN-04: "true"
|
||||
annotations:
|
||||
conformance.home.k3s/requirements: "R-VULN-04"
|
||||
conformance.home.k3s/layer: "L2"
|
||||
conformance.home.k3s/safety-critical: "false"
|
||||
conformance.home.k3s/intent: >-
|
||||
An unauthenticated request to cve.<domain> is redirected to Keycloak, and a logged-in user outside platform-admins gets 403.
|
||||
spec:
|
||||
steps:
|
||||
- name: given-the-page-route
|
||||
try:
|
||||
- script:
|
||||
content: 'echo "TODO: resolve cve.<domain> through the Gateway"'
|
||||
- name: when-unauthenticated
|
||||
try:
|
||||
- script:
|
||||
content: 'echo "TODO: request the page with no session"'
|
||||
- name: then-redirected-to-keycloak
|
||||
try:
|
||||
- script:
|
||||
content: 'echo "TODO: assert a 302 to the homelab realm"'
|
||||
- name: then-non-admin-forbidden
|
||||
try:
|
||||
- script:
|
||||
content: 'echo "TODO: assert 403 for a session outside platform-admins"'
|
||||
@@ -0,0 +1,40 @@
|
||||
# =============================================================================
|
||||
# L3 FAULT-INJECTION SCAFFOLD — breaking registry access produces failed targets and the coverage alert (R-VULN-03, R-VULN-06)
|
||||
# Catalog: docs/17 "Acceptance gates" (L3).
|
||||
#
|
||||
# With registry access broken for one image, that target becomes failed with its reason, prior findings are retained, and VulnCoverageGap fires after 24h.
|
||||
#
|
||||
# STATUS: scaffold only. It asserts nothing and DOES NOT EXECUTE IN CI: vars.CLUSTER_AVAILABLE is unset,
|
||||
# so L2/L3 conformance does not run. Coverage from this file is a named gap, not evidence.
|
||||
# =============================================================================
|
||||
apiVersion: chainsaw.kyverno.io/v1alpha1
|
||||
kind: Test
|
||||
metadata:
|
||||
name: vuln-registry-failure
|
||||
labels:
|
||||
conformance.home.k3s/R-VULN-03: "true"
|
||||
conformance.home.k3s/R-VULN-06: "true"
|
||||
annotations:
|
||||
conformance.home.k3s/requirements: "R-VULN-03, R-VULN-06"
|
||||
conformance.home.k3s/layer: "L3"
|
||||
conformance.home.k3s/safety-critical: "false"
|
||||
conformance.home.k3s/intent: >-
|
||||
With registry access broken for one image, that target becomes failed with its reason, prior findings are retained, and VulnCoverageGap fires after 24h.
|
||||
spec:
|
||||
steps:
|
||||
- name: given-a-scanned-target
|
||||
try:
|
||||
- script:
|
||||
content: 'echo "TODO: record a scanned target and its findings"'
|
||||
- name: when-registry-access-is-broken
|
||||
try:
|
||||
- script:
|
||||
content: 'echo "TODO: inject a fault blocking that registry"'
|
||||
- name: then-target-failed
|
||||
try:
|
||||
- script:
|
||||
content: 'echo "TODO: assert state failed, reason present, findings retained, run continues"'
|
||||
- name: then-coverage-alert-fires
|
||||
try:
|
||||
- script:
|
||||
content: 'echo "TODO: assert VulnCoverageGap fires once the failure passes 24h"'
|
||||
Reference in new issue
Block a user