"""Prometheus text exposition: aggregates only (R-VULN-06). The label allowlist is state, severity, fixable.""" from datetime import timedelta import report as report_mod from state import parse_ts _CRITICAL_GRACE = timedelta(hours=72) _SEVERITIES = ("CRITICAL", "HIGH", "MEDIUM", "LOW", "UNKNOWN") def _sample(lines, name, value, labels=""): lines.append(name + labels + " " + format(value, ".0f" if float(value).is_integer() else "g")) def render(report, state, now): lines = [] if report is not None: _sample(lines, "vuln_scanner_report_generated_timestamp_seconds", parse_ts(report["generatedAt"]).timestamp()) _sample(lines, "vuln_scanner_database_updated_timestamp_seconds", parse_ts(report["database"]["updatedAt"]).timestamp()) counts = {name: 0 for name in report_mod.STATES} for target in report["targets"]: counts[target["state"]] += 1 for name in report_mod.STATES: _sample(lines, "vuln_scanner_targets", counts[name], '{state="' + name + '"}') findings = {(sev, fix): 0 for sev in _SEVERITIES for fix in ("true", "false")} overdue = 0 for finding in report["findings"]: if finding["status"] == "accepted": continue findings[(finding["severity"], "true" if finding["fixedVersion"] else "false")] += 1 if finding["severity"] == "CRITICAL" and finding["fixedVersion"]: first_seen = parse_ts(finding["firstSeenAt"]) if first_seen and now - first_seen > _CRITICAL_GRACE: overdue += 1 for (severity, fixable), value in sorted(findings.items()): _sample(lines, "vuln_scanner_findings", value, '{severity="' + severity + '",fixable="' + fixable + '"}') _sample(lines, "vuln_scanner_critical_fix_overdue", overdue) _sample(lines, "vuln_scanner_failed_targets_over_24h", report_mod.failed_over_24h(state, now)) for name, key in (("vuln_scanner_last_complete_run_timestamp_seconds", "lastCompleteRunAt"), ("vuln_scanner_triage_heartbeat_timestamp_seconds", "triageHeartbeatAt")): value = parse_ts(state.get(key)) if value is not None: _sample(lines, name, value.timestamp()) return "\n".join(lines) + "\n"