feat(vuln): scanner service and fixture-driven L1 tests
Stdlib-only scanner per docs/17: Kubernetes API inventory, Trivy runner, persisted state, schema-validated report.json, strict acceptances parser, server-rendered page, aggregate metrics, internal API, and a scan loop that stays live while scanning. L1 tests under conformance/tests/l1-rendering/vuln-scanner-* run the actual modules against recorded fixtures with a fake Trivy runner. Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
jakman85andClaude Opus 5.5 committed
1 parent
4c3a284cad
commit
50ec2297ee
32 files changed
+3951
No files matched your search
@@ -0,0 +1,482 @@
|
||||
"""The CVE page: server-rendered HTML with no script at all.
|
||||
|
||||
Every string that reaches the document goes through e(), the single escape path; links additionally
|
||||
go through safe_url(). The style block is the only inline resource and is allowed by hash.
|
||||
"""
|
||||
import base64
|
||||
import hashlib
|
||||
import html
|
||||
import math
|
||||
from datetime import timedelta
|
||||
from urllib.parse import parse_qs, urlencode
|
||||
|
||||
import report as report_mod
|
||||
import trivy
|
||||
from state import parse_ts
|
||||
|
||||
REPORT_STALE = timedelta(hours=26)
|
||||
DB_STALE = timedelta(hours=48)
|
||||
SEVERITIES = ("CRITICAL", "HIGH", "MEDIUM", "LOW", "UNKNOWN")
|
||||
SEV_RANK = {name: i for i, name in enumerate(SEVERITIES)}
|
||||
STATUS_RANK = {"fix-available": 0, "open": 1, "accepted": 2}
|
||||
STATE_RANK = {name: i for i, name in enumerate(report_mod.STATES)}
|
||||
PAGE_SIZES = (25, 50, 100, 200, 500)
|
||||
DEFAULT_SIZE = 100
|
||||
_MAX_PARAM = 200
|
||||
_EXPANSION_LIMIT = 200 # per-target CVE list in the inventory view; the remainder is linked
|
||||
|
||||
STYLE = """
|
||||
body{font:14px/1.45 system-ui,sans-serif;margin:0;background:#fff;color:#1a1a1a}
|
||||
main{padding:0 16px 32px}
|
||||
header{padding:12px 16px;background:#f4f4f6;border-bottom:1px solid #ccc}
|
||||
h1{font-size:18px;margin:0 0 8px}
|
||||
nav a{margin-right:12px}
|
||||
.warn{background:#fff0d6;border:2px solid #b35900;padding:8px 12px;margin:8px 0;font-weight:600}
|
||||
.status{font-weight:700}
|
||||
.tiles span{display:inline-block;margin:2px 10px 2px 0}
|
||||
table{border-collapse:collapse;width:100%;margin-top:8px}
|
||||
th,td{border-bottom:1px solid #ddd;padding:4px 6px;text-align:left;vertical-align:top}
|
||||
th a{text-decoration:none}
|
||||
.sev-CRITICAL{color:#a00;font-weight:700}
|
||||
.sev-HIGH{color:#b35900;font-weight:700}
|
||||
form{margin:8px 0}
|
||||
form label{margin-right:8px}
|
||||
code{word-break:break-all}
|
||||
@media (prefers-color-scheme:dark){body{background:#16181c;color:#e6e6e6}
|
||||
header{background:#22252b;border-color:#3a3d44}
|
||||
th,td{border-color:#33363c}
|
||||
a{color:#7fb4ff}
|
||||
.warn{background:#3d2a12;border-color:#c98422}}
|
||||
"""
|
||||
STYLE_HASH = base64.b64encode(hashlib.sha256(STYLE.encode("utf-8")).digest()).decode("ascii")
|
||||
CSP = ("default-src 'none'; style-src 'sha256-" + STYLE_HASH + "'; base-uri 'none'; "
|
||||
"form-action 'self'; frame-ancestors 'none'")
|
||||
SECURITY_HEADERS = {
|
||||
"Content-Security-Policy": CSP,
|
||||
"X-Content-Type-Options": "nosniff",
|
||||
"Content-Type": "text/html; charset=utf-8",
|
||||
"Cache-Control": "no-store",
|
||||
}
|
||||
|
||||
|
||||
def e(value):
|
||||
"""The one escape path for anything placed in the document."""
|
||||
return "" if value is None else html.escape(str(value), quote=True)
|
||||
|
||||
|
||||
def safe_url(value):
|
||||
return trivy.safe_url(value) if isinstance(value, str) else ""
|
||||
|
||||
|
||||
def _link(url, label):
|
||||
url = safe_url(url)
|
||||
return '<a href="' + e(url) + '" rel="noopener noreferrer">' + e(label) + "</a>" if url else e(label)
|
||||
|
||||
|
||||
def _parse(query):
|
||||
raw = parse_qs(query or "", keep_blank_values=False)
|
||||
params = {k: v[0][:_MAX_PARAM] for k, v in raw.items() if v}
|
||||
view = params.get("view", "cves")
|
||||
params["view"] = view if view in ("cves", "workloads", "inventory") else "cves"
|
||||
try:
|
||||
params["page"] = max(1, int(params.get("page", "1")))
|
||||
except ValueError:
|
||||
params["page"] = 1
|
||||
try:
|
||||
size = int(params.get("size", str(DEFAULT_SIZE)))
|
||||
except ValueError:
|
||||
size = DEFAULT_SIZE
|
||||
params["size"] = size if size in PAGE_SIZES else DEFAULT_SIZE
|
||||
return params
|
||||
|
||||
|
||||
def _qs(params, **overrides):
|
||||
merged = dict(params)
|
||||
merged.update(overrides)
|
||||
pairs = sorted((k, str(v)) for k, v in merged.items() if v not in (None, ""))
|
||||
return "?" + urlencode(pairs)
|
||||
|
||||
|
||||
def _age(moment, now):
|
||||
if moment is None:
|
||||
return "unknown"
|
||||
seconds = int((now - moment).total_seconds())
|
||||
if seconds < 0:
|
||||
return "in the future"
|
||||
hours, remainder = divmod(seconds, 3600)
|
||||
minutes = remainder // 60
|
||||
return str(hours) + "h " + str(minutes) + "m ago"
|
||||
|
||||
|
||||
def _finding_text(finding, target):
|
||||
parts = [finding["id"], finding["package"], finding["title"], target["id"], target["image"]]
|
||||
for workload in target["workloads"]:
|
||||
parts.extend((workload["namespace"], workload["name"], *workload["containers"]))
|
||||
return " ".join(parts).lower()
|
||||
|
||||
|
||||
def _match_workload(target, params):
|
||||
namespace, workload = params.get("namespace"), params.get("workload")
|
||||
if not namespace and not workload:
|
||||
return True
|
||||
for item in target["workloads"]:
|
||||
if namespace and item["namespace"] != namespace:
|
||||
continue
|
||||
if workload and workload not in (item["name"], item["kind"] + "/" + item["name"]):
|
||||
continue
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _match_finding(finding, target, params):
|
||||
if params.get("severity") and finding["severity"] != params["severity"].upper():
|
||||
return False
|
||||
if params.get("status") and finding["status"] != params["status"]:
|
||||
return False
|
||||
fix = params.get("fix")
|
||||
if fix == "yes" and not finding["fixedVersion"]:
|
||||
return False
|
||||
if fix == "no" and finding["fixedVersion"]:
|
||||
return False
|
||||
if not _match_workload(target, params):
|
||||
return False
|
||||
text = params.get("q", "").lower()
|
||||
return not text or text in _finding_text(finding, target)
|
||||
|
||||
|
||||
def _paginate(rows, params):
|
||||
total = len(rows)
|
||||
size = params["size"]
|
||||
pages = max(1, math.ceil(total / size))
|
||||
page = min(params["page"], pages)
|
||||
start = (page - 1) * size
|
||||
return rows[start:start + size], total, page, pages, start
|
||||
|
||||
|
||||
def _pager(params, total, page, pages, start, shown, noun):
|
||||
first = start + 1 if total else 0
|
||||
summary = "Showing " + str(first) + "–" + str(start + shown) + " of " + str(total) + " " + noun \
|
||||
+ " (page " + str(page) + " of " + str(pages) + ")"
|
||||
links = []
|
||||
if page > 1:
|
||||
links.append('<a href="' + e(_qs(params, page=page - 1)) + '">previous</a>')
|
||||
if page < pages:
|
||||
links.append('<a href="' + e(_qs(params, page=page + 1)) + '">next</a>')
|
||||
return "<p>" + e(summary) + " " + " ".join(links) + "</p>"
|
||||
|
||||
|
||||
def _th(params, label, key):
|
||||
direction = "desc" if params.get("sort") == key and params.get("dir", "asc") == "asc" else "asc"
|
||||
return '<th><a href="' + e(_qs(params, sort=key, dir=direction, page=1)) + '">' + e(label) + "</a></th>"
|
||||
|
||||
|
||||
def _table(head, rows, empty):
|
||||
if not rows:
|
||||
return "<p>" + e(empty) + "</p>"
|
||||
return "<table><thead><tr>" + "".join(head) + "</tr></thead><tbody>" + "".join(rows) + "</tbody></table>"
|
||||
|
||||
|
||||
def _sev(severity):
|
||||
return '<span class="sev-' + e(severity) + '">' + e(severity) + "</span>"
|
||||
|
||||
|
||||
def _workload_lines(target):
|
||||
items = []
|
||||
for workload in target["workloads"]:
|
||||
items.append("<li>" + e(workload["namespace"]) + " / " + e(workload["kind"]) + " / " + e(workload["name"])
|
||||
+ " — containers: " + e(", ".join(workload["containers"]) or "none") + "</li>")
|
||||
return "<ul>" + "".join(items) + "</ul>"
|
||||
|
||||
|
||||
def _remediation_cell(images, remediation):
|
||||
if not images:
|
||||
return ""
|
||||
out = []
|
||||
for image in sorted(images):
|
||||
entry = remediation.get(image)
|
||||
if entry is None:
|
||||
out.append(e(image) + ": no data")
|
||||
elif entry["state"] == "pr-open" and safe_url(entry.get("url")):
|
||||
out.append(e(image) + ": " + _link(entry["url"], "pr-open"))
|
||||
else:
|
||||
out.append(e(image) + ": " + e(entry["state"]))
|
||||
return "<br>".join(out)
|
||||
|
||||
|
||||
# --- views ------------------------------------------------------------------------------------------
|
||||
|
||||
def _cves(report, targets, params, remediation):
|
||||
groups = {}
|
||||
for finding in report["findings"]:
|
||||
target = targets[finding["targetId"]]
|
||||
if not _match_finding(finding, target, params):
|
||||
continue
|
||||
group = groups.setdefault(finding["id"], {
|
||||
"id": finding["id"], "severity": finding["severity"], "cvss": None, "packages": set(), "fixed": set(),
|
||||
"firstSeen": finding["firstSeenAt"], "statuses": set(), "url": "", "title": "", "items": [],
|
||||
})
|
||||
if SEV_RANK[finding["severity"]] < SEV_RANK[group["severity"]]:
|
||||
group["severity"] = finding["severity"]
|
||||
if finding["cvss"] is not None and (group["cvss"] is None or finding["cvss"] > group["cvss"]):
|
||||
group["cvss"] = finding["cvss"]
|
||||
group["packages"].add(finding["package"])
|
||||
if finding["fixedVersion"]:
|
||||
group["fixed"].add(finding["fixedVersion"])
|
||||
group["firstSeen"] = min(group["firstSeen"], finding["firstSeenAt"])
|
||||
group["statuses"].add(finding["status"])
|
||||
group["url"] = group["url"] or finding["url"]
|
||||
group["title"] = group["title"] or finding["title"]
|
||||
group["items"].append((finding, target))
|
||||
rows = []
|
||||
for group in groups.values():
|
||||
group["status"] = min(group["statuses"], key=STATUS_RANK.get)
|
||||
group["imageCount"] = len({t["id"] for _, t in group["items"]})
|
||||
group["workloadCount"] = len({(w["namespace"], w["kind"], w["name"]) for _, t in group["items"] for w in t["workloads"]})
|
||||
group["package"] = ", ".join(sorted(group["packages"]))
|
||||
group["fix"] = ", ".join(sorted(group["fixed"]))
|
||||
rows.append(group)
|
||||
sorts = {
|
||||
"id": lambda r: r["id"], "severity": lambda r: SEV_RANK[r["severity"]], "cvss": lambda r: r["cvss"] if r["cvss"] is not None else -1,
|
||||
"package": lambda r: r["package"], "fixed": lambda r: r["fix"], "firstSeen": lambda r: r["firstSeen"],
|
||||
"status": lambda r: STATUS_RANK[r["status"]], "images": lambda r: r["imageCount"], "workloads": lambda r: r["workloadCount"],
|
||||
}
|
||||
rows.sort(key=lambda r: r["id"])
|
||||
rows.sort(key=sorts.get(params.get("sort"), sorts["severity"]), reverse=params.get("dir") == "desc")
|
||||
page_rows, total, page, pages, start = _paginate(rows, params)
|
||||
html_rows = []
|
||||
for row in page_rows:
|
||||
lines = []
|
||||
fix_images = set()
|
||||
for finding, target in row["items"]:
|
||||
if finding["status"] == "fix-available":
|
||||
fix_images.add(target["image"])
|
||||
lines.append("<li>" + e(target["id"]) + " [" + e(target["state"]) + "] — " + e(finding["package"]) + " "
|
||||
+ e(finding["installedVersion"]) + " → " + e(finding["fixedVersion"] or "no fix")
|
||||
+ _workload_lines(target) + "</li>")
|
||||
acceptance = next((f.get("acceptance") for f, _ in row["items"] if f.get("acceptance")), None)
|
||||
status_text = e(row["status"])
|
||||
if acceptance:
|
||||
status_text += " (" + e(acceptance["reason"]) + ", until " + e(acceptance["expires"]) + ")"
|
||||
html_rows.append(
|
||||
"<tr><td>" + _link(row["url"], row["id"]) + "</td><td>" + _sev(row["severity"]) + "</td><td>"
|
||||
+ e("" if row["cvss"] is None else row["cvss"]) + "</td><td>" + e(row["package"]) + "</td><td>" + e(row["fix"])
|
||||
+ "</td><td>" + e(row["firstSeen"]) + "</td><td>" + status_text + "</td><td>" + _remediation_cell(fix_images, remediation)
|
||||
+ "</td><td><details><summary>" + e(row["imageCount"]) + " images, " + e(row["workloadCount"])
|
||||
+ " workloads</summary><p>" + e(row["title"]) + "</p><ul>" + "".join(lines) + "</ul></details></td></tr>")
|
||||
head = [_th(params, "CVE", "id"), _th(params, "Severity", "severity"), _th(params, "CVSS", "cvss"), _th(params, "Package", "package"),
|
||||
_th(params, "Fixed in", "fixed"), _th(params, "First seen", "firstSeen"), _th(params, "Status", "status"),
|
||||
"<th>Remediation</th>", _th(params, "Affected", "images")]
|
||||
return _pager(params, total, page, pages, start, len(page_rows), "CVEs") \
|
||||
+ _table(head, html_rows, "No CVE rows match these filters.")
|
||||
|
||||
|
||||
def _workload_key(item):
|
||||
return item["namespace"], item["kind"], item["name"]
|
||||
|
||||
|
||||
def _workloads(report, targets, params):
|
||||
by_target = {}
|
||||
for finding in report["findings"]:
|
||||
target = targets[finding["targetId"]]
|
||||
if _match_finding(finding, target, params):
|
||||
by_target.setdefault(target["id"], []).append(finding)
|
||||
rows = {}
|
||||
finding_filter = params.get("severity") or params.get("status") or params.get("fix")
|
||||
needle = params.get("q", "").lower()
|
||||
for target in report["targets"]:
|
||||
matching = by_target.get(target["id"], [])
|
||||
for workload in target["workloads"]:
|
||||
key = _workload_key(workload)
|
||||
scope = {"namespace": params.get("namespace"), "workload": params.get("workload")}
|
||||
if not _match_workload({"workloads": [workload]}, scope):
|
||||
continue
|
||||
text = " ".join((workload["namespace"], workload["name"], *workload["containers"], target["image"])).lower()
|
||||
if not matching and (finding_filter or (needle and needle not in text)):
|
||||
continue
|
||||
row = rows.setdefault(key, {"key": key, "containers": set(), "images": set(), "findings": {}, "counts": dict.fromkeys(SEVERITIES, 0)})
|
||||
row["containers"].update(workload["containers"])
|
||||
row["images"].add(target["id"])
|
||||
for finding in matching:
|
||||
row["findings"].setdefault((finding["id"], target["id"], finding["package"]), finding)
|
||||
for row in rows.values():
|
||||
for finding in row["findings"].values():
|
||||
row["counts"][finding["severity"]] += 1
|
||||
items = list(rows.values())
|
||||
sorts = {
|
||||
"namespace": lambda r: r["key"][0], "kind": lambda r: r["key"][1], "name": lambda r: r["key"][2],
|
||||
"images": lambda r: len(r["images"]), "total": lambda r: -len(r["findings"]),
|
||||
}
|
||||
for severity in SEVERITIES:
|
||||
sorts[severity.lower()] = (lambda s: lambda r: -r["counts"][s])(severity)
|
||||
items.sort(key=lambda r: r["key"])
|
||||
if params.get("sort") in sorts:
|
||||
items.sort(key=sorts[params["sort"]], reverse=params.get("dir") == "desc")
|
||||
else:
|
||||
items.sort(key=lambda r: tuple(-r["counts"][s] for s in SEVERITIES))
|
||||
page_rows, total, page, pages, start = _paginate(items, params)
|
||||
html_rows = []
|
||||
for row in page_rows:
|
||||
lines = "".join(
|
||||
"<li>" + _link(f["url"], f["id"]) + " " + _sev(f["severity"]) + " " + e(f["package"]) + " — " + e(f["status"]) + "</li>"
|
||||
for f in sorted(row["findings"].values(), key=lambda f: (SEV_RANK[f["severity"]], f["id"])))
|
||||
counts = "".join("<td>" + e(row["counts"][s]) + "</td>" for s in SEVERITIES)
|
||||
html_rows.append(
|
||||
"<tr><td>" + e(row["key"][0]) + "</td><td>" + e(row["key"][1]) + "</td><td>" + e(row["key"][2]) + "</td><td>"
|
||||
+ e(", ".join(sorted(row["containers"])) or "none") + "</td><td>" + "<br>".join(e(i) for i in sorted(row["images"])) + "</td>"
|
||||
+ counts + "<td><details><summary>" + e(len(row["findings"])) + " findings</summary><ul>" + lines + "</ul></details></td></tr>")
|
||||
head = [_th(params, "Namespace", "namespace"), _th(params, "Kind", "kind"), _th(params, "Workload", "name"), "<th>Containers</th>",
|
||||
_th(params, "Images", "images")] + [_th(params, s, s.lower()) for s in SEVERITIES] + [_th(params, "Findings", "total")]
|
||||
return _pager(params, total, page, pages, start, len(page_rows), "workloads") \
|
||||
+ _table(head, html_rows, "No workload rows match these filters.")
|
||||
|
||||
|
||||
def _inventory(report, params):
|
||||
counts = {}
|
||||
by_target = {}
|
||||
for finding in report["findings"]:
|
||||
counts[finding["targetId"]] = counts.get(finding["targetId"], 0) + 1
|
||||
by_target.setdefault(finding["targetId"], []).append(finding)
|
||||
needle = params.get("q", "").lower()
|
||||
wanted_state = params.get("status")
|
||||
rows = []
|
||||
for target in report["targets"]:
|
||||
if wanted_state and target["state"] != wanted_state:
|
||||
continue
|
||||
if not _match_workload(target, params):
|
||||
continue
|
||||
if needle:
|
||||
text = " ".join((target["id"], target["image"], target["state"], *[w["namespace"] + " " + w["name"] for w in target["workloads"]])).lower()
|
||||
if needle not in text:
|
||||
continue
|
||||
rows.append(target)
|
||||
sorts = {
|
||||
"id": lambda t: t["id"], "type": lambda t: t["type"], "state": lambda t: STATE_RANK[t["state"]],
|
||||
"lastScanned": lambda t: t["lastScannedAt"] or "", "findings": lambda t: counts.get(t["id"], 0),
|
||||
}
|
||||
rows.sort(key=lambda t: t["id"])
|
||||
rows.sort(key=sorts.get(params.get("sort"), sorts["state"]), reverse=params.get("dir") == "desc")
|
||||
page_rows, total, page, pages, start = _paginate(rows, params)
|
||||
html_rows = []
|
||||
for target in page_rows:
|
||||
found = sorted(by_target.get(target["id"], []), key=lambda f: (SEV_RANK[f["severity"]], f["id"]))
|
||||
listed = "".join("<li>" + _link(f["url"], f["id"]) + " " + _sev(f["severity"]) + " " + e(f["package"]) + " — " + e(f["status"]) + "</li>"
|
||||
for f in found[:_EXPANSION_LIMIT])
|
||||
if len(found) > _EXPANSION_LIMIT:
|
||||
listed += '<li><a href="' + e(_qs({"view": "cves", "q": target["image"]})) + '">' + e(len(found) - _EXPANSION_LIMIT) + " more in the CVE view</a></li>"
|
||||
html_rows.append(
|
||||
"<tr><td><code>" + e(target["id"]) + "</code></td><td>" + e(target["type"]) + "</td><td>" + e(target["state"]) + "</td><td>"
|
||||
+ e(target["reason"]) + "</td><td>" + e(target["lastScannedAt"] or "never") + "</td><td>" + e(counts.get(target["id"], 0))
|
||||
+ "</td><td><details><summary>" + e(len(target["workloads"])) + " workloads, " + e(len(found)) + " findings</summary>"
|
||||
+ _workload_lines(target) + ("<ul>" + listed + "</ul>" if listed else "") + "</details></td></tr>")
|
||||
head = [_th(params, "Target", "id"), _th(params, "Type", "type"), _th(params, "State", "state"), "<th>Reason</th>",
|
||||
_th(params, "Last scanned", "lastScanned"), _th(params, "Findings", "findings"), "<th>Workloads</th>"]
|
||||
return _pager(params, total, page, pages, start, len(page_rows), "targets") \
|
||||
+ _table(head, html_rows, "No targets match these filters.")
|
||||
|
||||
|
||||
# --- header -----------------------------------------------------------------------------------------
|
||||
|
||||
def _header(report, now):
|
||||
generated = parse_ts(report["generatedAt"])
|
||||
database = parse_ts(report["database"]["updatedAt"])
|
||||
states = dict.fromkeys(report_mod.STATES, 0)
|
||||
for target in report["targets"]:
|
||||
states[target["state"]] += 1
|
||||
open_counts = dict.fromkeys(SEVERITIES, 0)
|
||||
accepted = 0
|
||||
for finding in report["findings"]:
|
||||
if finding["status"] == "accepted":
|
||||
accepted += 1
|
||||
else:
|
||||
open_counts[finding["severity"]] += 1
|
||||
warnings = []
|
||||
report_stale = generated is None or now - generated > REPORT_STALE
|
||||
if report_stale:
|
||||
warnings.append("STALE REPORT: generated " + _age(generated, now) + "; the expected refresh is within 26 hours.")
|
||||
if report["run"]["status"] == "degraded":
|
||||
warnings.append("DEGRADED RUN: the last run did not complete normally (database, acceptances file or inventory problem). Do not treat these results as current.")
|
||||
db_epoch = report["database"]["version"] == "unavailable"
|
||||
if db_epoch:
|
||||
warnings.append("DATABASE UNAVAILABLE: no vulnerability database has been downloaded.")
|
||||
elif database is None or now - database > DB_STALE:
|
||||
warnings.append("DATABASE STALE: the vulnerability database is older than 48 hours.")
|
||||
total_targets = len(report["targets"])
|
||||
not_scanned = total_targets - states["scanned"]
|
||||
if total_targets == 0:
|
||||
warnings.append("NO TARGETS: the inventory is empty; nothing has been evaluated.")
|
||||
if not_scanned:
|
||||
warnings.append("INCOMPLETE COVERAGE: " + str(not_scanned) + " of " + str(total_targets)
|
||||
+ " targets are not in the scanned state; findings for them are missing or may be out of date.")
|
||||
open_total = sum(open_counts.values())
|
||||
if warnings:
|
||||
status = "Status: ATTENTION NEEDED"
|
||||
elif open_total or accepted:
|
||||
status = "Status: CURRENT, findings present"
|
||||
else:
|
||||
status = "Status: CURRENT"
|
||||
tiles = "".join("<span>" + e(name) + ": <strong>" + e(states[name]) + "</strong></span>" for name in report_mod.STATES)
|
||||
sev_tiles = "".join("<span>" + _sev(s) + ": <strong>" + e(open_counts[s]) + "</strong></span>" for s in SEVERITIES)
|
||||
claim = ""
|
||||
if not warnings and not report["findings"] and total_targets > 0 and not_scanned == 0:
|
||||
claim = "<p><strong>Clean: all " + e(total_targets) + " targets were evaluated within 24 hours with zero findings.</strong></p>"
|
||||
return ("<header><h1>CVE visibility</h1>"
|
||||
"<p class=\"status\">" + e(status) + "</p>"
|
||||
+ "".join('<p class="warn">' + e(w) + "</p>" for w in warnings) + claim
|
||||
+ "<p>Report age: " + e(_age(generated, now)) + " · Database age: "
|
||||
+ e("unavailable" if db_epoch else _age(database, now)) + " · Run: " + e(report["run"]["status"]) + "</p>"
|
||||
+ '<p class="tiles">Targets (' + e(total_targets) + "): " + tiles + "</p>"
|
||||
+ '<p class="tiles">Open findings by severity: ' + sev_tiles + "<span>accepted: <strong>" + e(accepted) + "</strong></span></p>"
|
||||
+ "</header>")
|
||||
|
||||
|
||||
def _nav(params):
|
||||
links = []
|
||||
for view, label in (("cves", "CVEs"), ("workloads", "Workloads"), ("inventory", "Inventory")):
|
||||
marker = ' aria-current="page"' if params["view"] == view else ""
|
||||
target = {k: v for k, v in params.items() if k not in ("page", "sort", "dir")}
|
||||
target["view"] = view
|
||||
links.append('<a href="' + e(_qs(target)) + '"' + marker + ">" + e(label) + "</a>")
|
||||
return "<nav>" + "".join(links) + "</nav>"
|
||||
|
||||
|
||||
def _filters(params):
|
||||
def field(name, label):
|
||||
return "<label>" + e(label) + ' <input name="' + e(name) + '" value="' + e(params.get(name, "")) + '"></label>'
|
||||
selected = params.get("fix", "")
|
||||
fix_options = "".join(
|
||||
'<option value="' + e(v) + '"' + (" selected" if selected == v else "") + ">" + e(t) + "</option>"
|
||||
for v, t in (("", "any"), ("yes", "yes"), ("no", "no")))
|
||||
return ('<form method="get"><input type="hidden" name="view" value="' + e(params["view"]) + '">'
|
||||
+ field("q", "Search") + field("severity", "Severity") + field("namespace", "Namespace")
|
||||
+ field("workload", "Workload") + field("status", "Status")
|
||||
+ '<label>Fix available <select name="fix">' + fix_options + "</select></label>"
|
||||
+ '<button type="submit">Filter</button></form>')
|
||||
|
||||
|
||||
def _document(body):
|
||||
return ("<!doctype html><html lang=\"en\"><head><meta charset=\"utf-8\"><title>CVE visibility</title>"
|
||||
'<meta name="viewport" content="width=device-width, initial-scale=1"><style>' + STYLE + "</style></head>"
|
||||
"<body>" + body + "</body></html>").encode("utf-8")
|
||||
|
||||
|
||||
def render(report, query, now):
|
||||
"""(status code, header dict, body bytes) for one page request."""
|
||||
if report is None:
|
||||
body = ("<main><h1>CVE visibility</h1><p class=\"warn\">NO REPORT YET: nothing has been evaluated. "
|
||||
"Nothing can be said about this cluster until the first report is published.</p></main>")
|
||||
return 503, dict(SECURITY_HEADERS), _document(body)
|
||||
params = _parse(query)
|
||||
targets = {t["id"]: t for t in report["targets"]}
|
||||
remediation = {r["image"]: r for r in report["remediation"]}
|
||||
view = params["view"]
|
||||
if view == "workloads":
|
||||
content = _workloads(report, targets, params)
|
||||
elif view == "inventory":
|
||||
content = _inventory(report, params)
|
||||
else:
|
||||
content = _cves(report, targets, params, remediation)
|
||||
return 200, dict(SECURITY_HEADERS), _document(_header(report, now) + "<main>" + _nav(params) + _filters(params) + content + "</main>")
|
||||
|
||||
|
||||
def not_found():
|
||||
return 404, dict(SECURITY_HEADERS), _document("<main><p>Not found.</p></main>")
|
||||
Reference in new issue
Block a user