jakman85andClaude Opus 5.5
d8b25a67fe
Review findings on the integrated scanner: - An OOM kill no longer wedges the loop: inFlight is persisted before each scan, and on restart that image is marked failed and sits out full runs for 20h. - One bad upstream string can no longer freeze the report: safe_url requires an exact lowercase scheme, acceptance reasons are bounded, /triage urls use the same rule, and report.build repairs or drops any entry the schema would refuse. lastCompleteRunAt advances only after a report is actually published. - The private registry reports credential-required, not failed: VULN_PRIVATE_REGISTRIES names it, and Trivy is not invoked. - loadError clears after a full run against a good inventory. - Every image under platform/vuln-scanner stays out of Renovate auto-merge. - The page labels status and target state per line, and mixed acceptance reads as mixed. The Keycloak client is still an operator action: its key must go in an existing encrypted file that an agent must not modify. platform/vuln-scanner/README.md has the exact steps. Co-Authored-By: Claude Opus 5.5 <[email protected]>
506 lines
24 KiB
Python
506 lines
24 KiB
Python
"""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 _state_label(target):
|
||
return "target " + e(target["state"]) + (" (" + e(target["reason"]) + ")" if target.get("reason") else "")
|
||
|
||
|
||
def _status_label(finding):
|
||
"""One finding's own status, with its own acceptance: never another image's."""
|
||
label = "status " + e(finding["status"])
|
||
acceptance = finding.get("acceptance")
|
||
if acceptance:
|
||
label += " (" + e(acceptance["reason"]) + ", until " + e(acceptance["expires"]) + ")"
|
||
return label
|
||
|
||
|
||
def _workload_lines(target, finding=None):
|
||
"""One line per workload, each carrying the target state and, in a CVE expansion, the status."""
|
||
suffix = " — " + _state_label(target) + (" — " + _status_label(finding) if finding else "")
|
||
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") + suffix + "</li>")
|
||
return "<ul>" + "".join(items) + "</ul>"
|
||
|
||
|
||
def _status_summary(statuses):
|
||
"""The row-level status. A CVE accepted on one image and open on another reads as mixed."""
|
||
counts = {}
|
||
for status in statuses:
|
||
counts[status] = counts.get(status, 0) + 1
|
||
if len(counts) == 1:
|
||
return e(next(iter(counts)))
|
||
return "mixed: " + ", ".join(e(counts[s]) + " " + e(s) for s in sorted(counts, key=STATUS_RANK.get))
|
||
|
||
|
||
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"]) + " — " + _state_label(target) + " — " + _status_label(finding)
|
||
+ " — " + e(finding["package"]) + " " + e(finding["installedVersion"]) + " → "
|
||
+ e(finding["fixedVersion"] or "no fix") + _workload_lines(target, finding) + "</li>")
|
||
status_text = _status_summary(f["status"] for f, _ in row["items"])
|
||
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"]) + " — " + _status_label(f)
|
||
+ " — " + e(f["targetId"]) + " — " + _state_label(targets[f["targetId"]]) + "</li>"
|
||
for f in sorted(row["findings"].values(), key=lambda f: (SEV_RANK[f["severity"]], f["id"], f["targetId"])))
|
||
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) + " — " + _state_label(targets[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>")
|