feat(vuln): CVE triage bot workflow, script and L1 tests
Scheduled Gitea Actions bot in the Renovate shape: maintains a counts-only CVE Dashboard issue, derives remediation states from Renovate's dashboard and PRs, ticks approve rows for exact-matched images with an unaccepted CRITICAL with a fix, and posts the heartbeat. Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
jakman85andClaude Opus 5.5 committed
1 parent
4c3a284cad
commit
c3c72c47bb
12 files changed
+2101
No files matched your search
Executable
+820
@@ -0,0 +1,820 @@
|
||||
#!/usr/bin/env python3
|
||||
"""CVE triage bot (docs/17 "The triage bot", ADR-0027; R-VULN-05, R-VULN-07).
|
||||
|
||||
Runs in .gitea/workflows/cve-triage.yaml, once a day, before Renovate. Each run:
|
||||
1. reads report.json from the scanner's internal API;
|
||||
2. creates or updates the single `CVE Dashboard` issue (aggregates only);
|
||||
3. reads Renovate's Dependency Dashboard and open PRs, derives each image's remediation state;
|
||||
4. ticks `approve-branch` boxes on that dashboard, for packages matching an image with an
|
||||
unaccepted CRITICAL that has a fix;
|
||||
5. POSTs the heartbeat and the remediation states to the scanner.
|
||||
|
||||
WHAT IT MUST NEVER DO. Renovate is the only author of version changes (R-UPD-02): this script
|
||||
never edits a version, opens a bump PR, merges anything, or posts `/approve-override`. It never
|
||||
ticks a `rebase-branch` row. The dashboard issue carries no CVE id, package, image or workload,
|
||||
because the repository is readable without authentication, and no label the in-cluster PM acts
|
||||
on (platform/pm-agent/configmap.yaml PM_REQUIRE_LABEL = `agent`).
|
||||
|
||||
OUTPUT IS COUNTS ONLY. Job logs and the summary are as visible as the issue, so nothing here
|
||||
prints a report string. Every report value is untrusted: strings are validated, and only counts
|
||||
and timestamps that passed RFC 3339 validation ever reach markdown.
|
||||
|
||||
HOW A JOB CONTAINER REACHES THE SCANNER, AND WHAT IS NOT PROVEN
|
||||
---------------------------------------------------------------
|
||||
Jobs are containers on a per-job docker bridge network inside the DinD sidecar of the act-runner
|
||||
pod (apps/wave-1/gitea-actions.yaml). The runner config sets no `container.network` and dockerd no
|
||||
`--dns`, so name resolution is docker's embedded resolver forwarding to the resolvers of the
|
||||
dockerd container, which are that pod's: cluster DNS. Egress is masqueraded to the pod IP
|
||||
(measured 2026-10-29, platform/buildkit/networkpolicy.yaml header), so a ClusterIP is routed and
|
||||
policy-matched as the act-runner pod in gitea-actions. images.yaml relies on the same path to
|
||||
reach buildkitd.buildkit.svc.cluster.local, which is indirect evidence only.
|
||||
UNVERIFIED: in-cluster DNS from a DinD job container.
|
||||
UNVERIFIED: ClusterIP reachability from a DinD job container to vuln-scanner:8081.
|
||||
Hence VULN_API_FALLBACK_URL (the Service's pinned ClusterIP), used only when the name fails to
|
||||
resolve. Anything else, including refused and timed-out connections, fails the job: an
|
||||
unreachable API is a red run, never a skip.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import http.client
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import socket
|
||||
import sys
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Callable, Iterable, NamedTuple
|
||||
|
||||
DASHBOARD_TITLE = "CVE Dashboard"
|
||||
DASHBOARD_LABEL = "cve-dashboard"
|
||||
DASHBOARD_LABEL_COLOR = "b60205"
|
||||
# Labels the in-cluster PM dispatcher selects work by. Never applied to anything here.
|
||||
FORBIDDEN_LABELS = frozenset({"agent"})
|
||||
RENOVATE_DASHBOARD_TITLE = "Dependency Dashboard"
|
||||
|
||||
SEVERITIES = ("CRITICAL", "HIGH", "MEDIUM", "LOW", "UNKNOWN")
|
||||
FINDING_STATUSES = ("open", "fix-available", "accepted")
|
||||
TARGET_STATES = ("scanned", "pending", "stale", "failed", "credential-required", "by-tag", "not-covered")
|
||||
TARGET_TYPES = ("image", "vm", "node")
|
||||
RUN_STATUSES = ("complete", "degraded")
|
||||
REMEDIATION_STATES = ("pr-open", "approval-requested", "no-update-identified", "not-tracked")
|
||||
|
||||
MAX_REPORT_BYTES = 128 * 1024 * 1024
|
||||
MAX_API_BYTES = 8 * 1024 * 1024
|
||||
MAX_FIELD_LEN = 4096
|
||||
MAX_IMAGE_LEN = 512
|
||||
MAX_ISSUE_BODY = 2 * 1024 * 1024
|
||||
MAX_DASHBOARD_ROWS = 2000
|
||||
MAX_DETECTED_DEPS = 20000
|
||||
MAX_REMEDIATION = 5000
|
||||
MAX_PAGES = 200
|
||||
PAGE_SIZE = 50
|
||||
REPORT_STALE_HOURS = 26
|
||||
DATABASE_STALE_HOURS = 48
|
||||
|
||||
HTTP_TIMEOUT = 60
|
||||
|
||||
RFC3339_RE = re.compile(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d{1,9})?(Z|[+-]\d{2}:\d{2})")
|
||||
IMAGE_RE = re.compile(r"[a-z0-9]([a-z0-9._:/-]*[a-z0-9])?")
|
||||
HTTP_URL_RE = re.compile(r"https?://[A-Za-z0-9.-]+(:\d{1,5})?(/[A-Za-z0-9._~%/-]*)?")
|
||||
REPO_RE = re.compile(r"[A-Za-z0-9._-]+/[A-Za-z0-9._-]+")
|
||||
|
||||
# Renovate's own row shape (lib/workers/repository/dependency-dashboard.ts, getListItem). The kind
|
||||
# is captured whole so `approvePr-branch` can never read as `approve-branch`.
|
||||
ROW_RE = re.compile(r" - \[(?P<mark>[ xX])\] <!-- (?P<kind>[A-Za-z]+)-branch=(?P<branch>\S+) -->(?P<rest>.*)")
|
||||
LINKED_TITLE_RE = re.compile(r"\[(?P<title>.*)\]\((?P<target>[^()\s]*)\)(?P<tail>.*)")
|
||||
PR_LINK_RE = re.compile(r"(?:^|/)pulls?/(\d{1,9})$")
|
||||
PACKAGE_LIST_RE = re.compile(r"\s\(((?:`[^`\s]+`(?:, )?)+)\)\s*$")
|
||||
TITLE_PACKAGE_RE = re.compile(r"(?:^|\s)(?:Update|Pin)\s+(\S+)\s+Docker\s+(?:tag|digest)\b")
|
||||
SUMMARY_RE = re.compile(r"<details><summary>([^<]*?)(?: \(\d+\))?</summary>")
|
||||
DETECTED_RE = re.compile(r" - `(?P<name>[^\s`]+)(?: [^`]*)?`(?P<updates>\s*→ \[Updates:.*\])?\s*")
|
||||
|
||||
# Detected-dependency managers whose names are charts or actions, never container images.
|
||||
NON_IMAGE_MANAGERS = frozenset({"flux", "helmv3", "github-actions"})
|
||||
|
||||
|
||||
class BotError(Exception):
|
||||
"""A failure that ends the run. The message must be safe to print: no report text, no secret."""
|
||||
|
||||
|
||||
class DnsFailure(Exception):
|
||||
"""The hostname did not resolve. The only condition that selects the fallback address."""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- pure: normalisation + report
|
||||
|
||||
|
||||
def normalise_image(value: object) -> str:
|
||||
"""Canonical form for exact comparison, or "" when the value is not an image reference.
|
||||
|
||||
Lowercase; drop digest and tag; drop the Docker Hub registry and the `library/` namespace.
|
||||
`library/` is only Hub's implicit namespace, so ghcr.io/library/x keeps it. Comparison of the
|
||||
result is always string equality, never a substring test.
|
||||
"""
|
||||
if not isinstance(value, str) or not value or len(value) > MAX_IMAGE_LEN:
|
||||
return ""
|
||||
name = value.strip().lower().split("@", 1)[0]
|
||||
head, sep, last = name.rpartition("/")
|
||||
name = f"{head}{sep}{last.split(':', 1)[0]}"
|
||||
for hub in ("docker.io/", "index.docker.io/", "registry-1.docker.io/"):
|
||||
if name.startswith(hub):
|
||||
name = name[len(hub):]
|
||||
break
|
||||
if name.startswith("library/"):
|
||||
name = name[len("library/"):]
|
||||
return name if IMAGE_RE.fullmatch(name) else ""
|
||||
|
||||
|
||||
def parse_timestamp(value: object, field: str, nullable: bool = False) -> datetime | None:
|
||||
if value is None and nullable:
|
||||
return None
|
||||
if not isinstance(value, str) or not RFC3339_RE.fullmatch(value):
|
||||
raise BotError(f"report: {field} is not an RFC 3339 timestamp")
|
||||
text = re.sub(r"\.\d+", "", value).replace("Z", "+00:00")
|
||||
try:
|
||||
return datetime.fromisoformat(text)
|
||||
except ValueError as exc:
|
||||
raise BotError(f"report: {field} is not a valid timestamp") from exc
|
||||
|
||||
|
||||
def _object(value: object, field: str) -> dict:
|
||||
if not isinstance(value, dict):
|
||||
raise BotError(f"report: {field} is not an object")
|
||||
return value
|
||||
|
||||
|
||||
def _array(value: object, field: str) -> list:
|
||||
if not isinstance(value, list):
|
||||
raise BotError(f"report: {field} is not an array")
|
||||
return value
|
||||
|
||||
|
||||
def _string(value: object, field: str, nullable: bool = False) -> str | None:
|
||||
if value is None and nullable:
|
||||
return None
|
||||
if not isinstance(value, str) or len(value) > MAX_FIELD_LEN:
|
||||
raise BotError(f"report: {field} is not a bounded string")
|
||||
return value
|
||||
|
||||
|
||||
def _member(value: object, allowed: Iterable[str], field: str) -> str:
|
||||
if not isinstance(value, str) or value not in allowed:
|
||||
raise BotError(f"report: {field} has a value outside its enumeration")
|
||||
return value
|
||||
|
||||
|
||||
def parse_report(raw: bytes) -> dict:
|
||||
"""Decode and validate report.json into the only shape the rest of this file reads."""
|
||||
try:
|
||||
doc = json.loads(raw)
|
||||
except (ValueError, RecursionError) as exc:
|
||||
raise BotError("report: body is not valid JSON") from exc
|
||||
doc = _object(doc, "document")
|
||||
if doc.get("schemaVersion") != 1 or isinstance(doc.get("schemaVersion"), bool):
|
||||
raise BotError("report: unsupported schemaVersion")
|
||||
parse_timestamp(doc.get("generatedAt"), "generatedAt")
|
||||
database = _object(doc.get("database"), "database")
|
||||
run = _object(doc.get("run"), "run")
|
||||
parse_timestamp(database.get("updatedAt"), "database.updatedAt", nullable=True)
|
||||
parse_timestamp(run.get("startedAt"), "run.startedAt", nullable=True)
|
||||
parse_timestamp(run.get("finishedAt"), "run.finishedAt", nullable=True)
|
||||
if run.get("status") is not None:
|
||||
_member(run.get("status"), RUN_STATUSES, "run.status")
|
||||
|
||||
targets = []
|
||||
known_ids = {}
|
||||
for entry in _array(doc.get("targets"), "targets"):
|
||||
entry = _object(entry, "target")
|
||||
target_id = _string(entry.get("id"), "target.id")
|
||||
kind = _member(entry.get("type"), TARGET_TYPES, "target.type")
|
||||
state = _member(entry.get("state"), TARGET_STATES, "target.state")
|
||||
image = _string(entry.get("image"), "target.image", nullable=True)
|
||||
if kind == "image" and not normalise_image(image):
|
||||
raise BotError("report: an image target has an invalid image reference")
|
||||
known_ids[target_id] = image if kind == "image" else None
|
||||
targets.append({"id": target_id, "type": kind, "state": state, "image": image})
|
||||
|
||||
findings = []
|
||||
for entry in _array(doc.get("findings"), "findings"):
|
||||
entry = _object(entry, "finding")
|
||||
target_id = _string(entry.get("targetId"), "finding.targetId")
|
||||
if target_id not in known_ids:
|
||||
raise BotError("report: a finding references an unknown target")
|
||||
findings.append({
|
||||
"targetId": target_id,
|
||||
"image": known_ids[target_id],
|
||||
"severity": _member(entry.get("severity"), SEVERITIES, "finding.severity"),
|
||||
"status": _member(entry.get("status"), FINDING_STATUSES, "finding.status"),
|
||||
"fixedVersion": _string(entry.get("fixedVersion"), "finding.fixedVersion", nullable=True),
|
||||
})
|
||||
return {
|
||||
"generatedAt": doc["generatedAt"],
|
||||
"database": {"updatedAt": database.get("updatedAt")},
|
||||
"run": {"finishedAt": run.get("finishedAt"), "status": run.get("status")},
|
||||
"targets": targets,
|
||||
"findings": findings,
|
||||
}
|
||||
|
||||
|
||||
def qualifies(finding: dict) -> bool:
|
||||
"""An unaccepted CRITICAL with a fix: the only thing that may cause a tick."""
|
||||
return (
|
||||
finding["severity"] == "CRITICAL"
|
||||
and finding["status"] == "fix-available"
|
||||
and bool(finding["fixedVersion"])
|
||||
)
|
||||
|
||||
|
||||
def qualifying_images(report: dict) -> set[str]:
|
||||
return {
|
||||
norm
|
||||
for finding in report["findings"]
|
||||
if qualifies(finding) and (norm := normalise_image(finding["image"]))
|
||||
}
|
||||
|
||||
|
||||
def aggregate(report: dict) -> dict:
|
||||
severity = {sev: dict.fromkeys(FINDING_STATUSES, 0) for sev in SEVERITIES}
|
||||
for finding in report["findings"]:
|
||||
severity[finding["severity"]][finding["status"]] += 1
|
||||
coverage = dict.fromkeys(TARGET_STATES, 0)
|
||||
for target in report["targets"]:
|
||||
coverage[target["state"]] += 1
|
||||
return {
|
||||
"severity": severity,
|
||||
"coverage": coverage,
|
||||
"targets_total": len(report["targets"]),
|
||||
"critical_with_fix": sum(1 for f in report["findings"] if qualifies(f)),
|
||||
"generatedAt": report["generatedAt"],
|
||||
"databaseUpdatedAt": report["database"]["updatedAt"],
|
||||
"runFinishedAt": report["run"]["finishedAt"],
|
||||
"runStatus": report["run"]["status"],
|
||||
}
|
||||
|
||||
|
||||
def render_dashboard(agg: dict, page_url: str, now: datetime) -> str:
|
||||
"""The CVE Dashboard issue body: a fixed template over counts and validated timestamps."""
|
||||
if not HTTP_URL_RE.fullmatch(page_url):
|
||||
raise BotError("config: VULN_PAGE_URL is not a plain http(s) URL")
|
||||
warnings = []
|
||||
generated = parse_timestamp(agg["generatedAt"], "generatedAt")
|
||||
if now - generated > timedelta(hours=REPORT_STALE_HOURS):
|
||||
warnings.append(f"The report is older than {REPORT_STALE_HOURS} hours.")
|
||||
updated = parse_timestamp(agg["databaseUpdatedAt"], "database.updatedAt", nullable=True)
|
||||
if updated is None or now - updated > timedelta(hours=DATABASE_STALE_HOURS):
|
||||
warnings.append(f"The vulnerability database is older than {DATABASE_STALE_HOURS} hours or unknown.")
|
||||
if agg["runStatus"] != "complete":
|
||||
warnings.append("The latest scan run did not complete cleanly.")
|
||||
|
||||
lines = [
|
||||
"Maintained by the CVE triage bot, once a day. **Counts only**: this repository is readable",
|
||||
"without authentication, so which CVE lands where is only on the",
|
||||
f"[CVE page]({page_url}) (SSO, platform-admins).",
|
||||
"",
|
||||
]
|
||||
lines += [f"> **WARNING:** {text}" for text in warnings]
|
||||
if warnings:
|
||||
lines.append("")
|
||||
lines += [
|
||||
"## Freshness",
|
||||
"",
|
||||
"| | |",
|
||||
"|---|---|",
|
||||
f"| Report generated | {agg['generatedAt']} |",
|
||||
f"| Database updated | {agg['databaseUpdatedAt'] or 'unknown'} |",
|
||||
f"| Last run finished | {agg['runFinishedAt'] or 'none yet'} |",
|
||||
f"| Last run status | {agg['runStatus'] or 'none yet'} |",
|
||||
f"| Dashboard rendered | {now.strftime('%Y-%m-%dT%H:%M:%SZ')} |",
|
||||
"",
|
||||
"## Findings by severity",
|
||||
"",
|
||||
"| Severity | Open | Fix available | Accepted |",
|
||||
"|---|---:|---:|---:|",
|
||||
]
|
||||
for sev in SEVERITIES:
|
||||
row = agg["severity"][sev]
|
||||
lines.append(f"| {sev} | {row['open']} | {row['fix-available']} | {row['accepted']} |")
|
||||
lines += [
|
||||
"",
|
||||
f"Unaccepted CRITICAL with a fix: **{agg['critical_with_fix']}**",
|
||||
"",
|
||||
"## Coverage",
|
||||
"",
|
||||
f"{agg['targets_total']} targets. A count in any state other than `scanned` is something"
|
||||
" that was not freshly examined, and is not a clean result.",
|
||||
"",
|
||||
"| State | Targets |",
|
||||
"|---|---:|",
|
||||
]
|
||||
lines += [f"| {state} | {agg['coverage'][state]} |" for state in TARGET_STATES]
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- pure: Renovate's dashboard
|
||||
|
||||
|
||||
class Row(NamedTuple):
|
||||
kind: str
|
||||
mark: str
|
||||
branch: str
|
||||
packages: tuple[str, ...]
|
||||
pr_number: int | None
|
||||
offset: int # offset in the body of the first character of the line
|
||||
|
||||
|
||||
def _title_package(title: str) -> str:
|
||||
match = TITLE_PACKAGE_RE.search(title)
|
||||
return normalise_image(match.group(1)) if match else ""
|
||||
|
||||
|
||||
def _row_details(rest: str) -> tuple[tuple[str, ...], int | None]:
|
||||
"""Packages and PR number of a row. No recognisable package means no packages, so no tick."""
|
||||
rest = rest.rstrip("\r")
|
||||
number = None
|
||||
title = rest
|
||||
tail = rest
|
||||
linked = LINKED_TITLE_RE.fullmatch(rest)
|
||||
if linked:
|
||||
title, tail = linked.group("title"), linked.group("tail")
|
||||
link = PR_LINK_RE.search(linked.group("target"))
|
||||
number = int(link.group(1)) if link else None
|
||||
listed = PACKAGE_LIST_RE.search(tail)
|
||||
if listed:
|
||||
names = re.findall(r"`([^`]+)`", listed.group(1))
|
||||
packages = tuple(n for n in map(normalise_image, names) if n)
|
||||
else:
|
||||
package = _title_package(title)
|
||||
packages = (package,) if package else ()
|
||||
return packages, number
|
||||
|
||||
|
||||
def _is_detected_heading(line: str) -> bool:
|
||||
return line.startswith("## ") and line[3:].strip().lower().startswith("detected dependencies")
|
||||
|
||||
|
||||
def parse_rows(body: str) -> list[Row]:
|
||||
"""Checkbox rows above `## Detected Dependencies`. Only a whole-line match counts."""
|
||||
if len(body) > MAX_ISSUE_BODY:
|
||||
raise BotError("renovate dashboard: body exceeds its bound")
|
||||
rows: list[Row] = []
|
||||
offset = 0
|
||||
for line in body.split("\n"):
|
||||
if _is_detected_heading(line):
|
||||
break
|
||||
match = ROW_RE.fullmatch(line.rstrip("\r"))
|
||||
if match:
|
||||
packages, number = _row_details(match.group("rest"))
|
||||
rows.append(Row(match.group("kind"), match.group("mark"), match.group("branch"),
|
||||
packages, number, offset))
|
||||
if len(rows) > MAX_DASHBOARD_ROWS:
|
||||
raise BotError("renovate dashboard: too many rows")
|
||||
offset += len(line) + 1
|
||||
return rows
|
||||
|
||||
|
||||
def parse_detected(body: str) -> dict[str, bool]:
|
||||
"""{normalised depName: has a pending update} from `## Detected Dependencies`."""
|
||||
tracked: dict[str, bool] = {}
|
||||
lines = body.split("\n")
|
||||
in_section = False
|
||||
manager = ""
|
||||
count = 0
|
||||
for index, raw in enumerate(lines):
|
||||
line = raw.rstrip("\r")
|
||||
if line.startswith("## "):
|
||||
in_section = _is_detected_heading(line)
|
||||
continue
|
||||
if not in_section:
|
||||
continue
|
||||
summary = SUMMARY_RE.fullmatch(line)
|
||||
if summary:
|
||||
# A manager heading is the summary directly followed by <blockquote>; a file heading
|
||||
# is not.
|
||||
if index + 1 < len(lines) and lines[index + 1].strip() == "<blockquote>":
|
||||
manager = summary.group(1)
|
||||
continue
|
||||
match = DETECTED_RE.fullmatch(line)
|
||||
if not match or manager in NON_IMAGE_MANAGERS:
|
||||
continue
|
||||
count += 1
|
||||
if count > MAX_DETECTED_DEPS:
|
||||
raise BotError("renovate dashboard: too many detected dependencies")
|
||||
norm = normalise_image(match.group("name"))
|
||||
if norm:
|
||||
tracked[norm] = tracked.get(norm, False) or bool(match.group("updates"))
|
||||
return tracked
|
||||
|
||||
|
||||
def row_matches(row: Row, qualifying: set[str]) -> bool:
|
||||
return any(package in qualifying for package in row.packages)
|
||||
|
||||
|
||||
def select_ticks(rows: list[Row], qualifying: set[str]) -> list[Row]:
|
||||
return [
|
||||
row for row in rows
|
||||
if row.kind == "approve" and row.mark == " " and row_matches(row, qualifying)
|
||||
]
|
||||
|
||||
|
||||
def apply_ticks(body: str, ticks: list[Row]) -> str:
|
||||
"""Replace only the `[ ]` of each row, re-verifying the row where it stands.
|
||||
|
||||
Every byte other than the three of each marker is carried over unchanged.
|
||||
"""
|
||||
pieces = []
|
||||
cursor = 0
|
||||
for row in sorted(ticks, key=lambda r: r.offset):
|
||||
expected = f" - [ ] <!-- approve-branch={row.branch} -->"
|
||||
if row.kind != "approve" or not body.startswith(expected, row.offset):
|
||||
raise BotError("renovate dashboard: a row changed between parse and write")
|
||||
marker = row.offset + len(" - ")
|
||||
pieces += [body[cursor:marker], "[x]"]
|
||||
cursor = marker + len("[ ]")
|
||||
pieces.append(body[cursor:])
|
||||
return "".join(pieces)
|
||||
|
||||
|
||||
def derive_remediation(
|
||||
images: Iterable[str],
|
||||
rows: list[Row],
|
||||
open_prs: list[dict],
|
||||
tracked: dict[str, bool],
|
||||
pr_url: Callable[[int], str],
|
||||
now_text: str,
|
||||
) -> list[dict]:
|
||||
"""One entry per report image.
|
||||
|
||||
pr-open: an open-PR row or an open Renovate PR for that package. approval-requested: an
|
||||
approve row. no-update-identified: Renovate tracks it and lists no update. not-tracked:
|
||||
everything else, which deliberately includes "Renovate knows an update but no PR or approval
|
||||
row exists" (awaiting schedule, rate limited): saying no update was identified would be
|
||||
false, and the enumeration has no closer value, so it lands on the state that draws attention.
|
||||
"""
|
||||
by_branch = {pr["branch"]: pr["number"] for pr in open_prs}
|
||||
result = []
|
||||
for image in sorted(set(images)):
|
||||
norm = normalise_image(image)
|
||||
state = "not-tracked"
|
||||
url = None
|
||||
rebase = next((r for r in rows if r.kind == "rebase" and norm in r.packages), None)
|
||||
pr = next((p for p in open_prs if p["package"] == norm), None)
|
||||
if rebase is not None or pr is not None:
|
||||
state = "pr-open"
|
||||
number = None
|
||||
if rebase is not None:
|
||||
number = rebase.pr_number or by_branch.get(rebase.branch)
|
||||
if number is None and pr is not None:
|
||||
number = pr["number"]
|
||||
url = pr_url(number) if number is not None else None
|
||||
elif any(r.kind == "approve" and norm in r.packages for r in rows):
|
||||
state = "approval-requested"
|
||||
elif norm in tracked and not tracked[norm]:
|
||||
state = "no-update-identified"
|
||||
result.append({"image": image, "state": state, "url": url, "updatedAt": now_text})
|
||||
return result
|
||||
|
||||
|
||||
def count_states(remediation: list[dict]) -> dict[str, int]:
|
||||
counts = dict.fromkeys(REMEDIATION_STATES, 0)
|
||||
for entry in remediation:
|
||||
counts[entry["state"]] += 1
|
||||
return counts
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- I/O
|
||||
|
||||
|
||||
class _NoRedirect(urllib.request.HTTPRedirectHandler):
|
||||
"""A redirect would carry the Authorization header to wherever it points."""
|
||||
|
||||
def redirect_request(self, *args, **kwargs):
|
||||
return None
|
||||
|
||||
|
||||
class UrllibTransport:
|
||||
"""(method, url, headers, body, max_bytes) -> (status, bytes); never uses a proxy."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), _NoRedirect())
|
||||
|
||||
def __call__(self, method: str, url: str, headers: dict, body: bytes | None, max_bytes: int):
|
||||
request = urllib.request.Request(url, data=body, method=method, headers=headers)
|
||||
try:
|
||||
with self._opener.open(request, timeout=HTTP_TIMEOUT) as response:
|
||||
data = response.read(max_bytes + 1)
|
||||
return response.status, data
|
||||
except urllib.error.HTTPError as exc:
|
||||
return exc.code, b""
|
||||
except urllib.error.URLError as exc:
|
||||
if isinstance(exc.reason, socket.gaierror):
|
||||
raise DnsFailure() from exc
|
||||
raise BotError(f"{method} {urllib.parse.urlsplit(url).path}: {type(exc.reason).__name__}") from exc
|
||||
except (OSError, http.client.HTTPException) as exc:
|
||||
raise BotError(f"{method} {urllib.parse.urlsplit(url).path}: {type(exc).__name__}") from exc
|
||||
|
||||
|
||||
class GiteaClient:
|
||||
def __init__(self, base_url: str, repo: str, token: str, transport) -> None:
|
||||
self._api = f"{base_url.rstrip('/')}/api/v1/repos/{repo}"
|
||||
self._web = f"{base_url.rstrip('/')}/{repo}"
|
||||
self._token = token
|
||||
self._transport = transport
|
||||
|
||||
def _call(self, method: str, path: str, query: dict | None = None, payload=None, expect=(200,)):
|
||||
url = self._api + path + (f"?{urllib.parse.urlencode(query)}" if query else "")
|
||||
headers = {"Authorization": f"token {self._token}", "Accept": "application/json"}
|
||||
data = None
|
||||
if payload is not None:
|
||||
data = json.dumps(payload).encode()
|
||||
headers["Content-Type"] = "application/json"
|
||||
try:
|
||||
status, raw = self._transport(method, url, headers, data, MAX_API_BYTES)
|
||||
except DnsFailure as exc:
|
||||
raise BotError(f"gitea: {method} {path}: name resolution failed") from exc
|
||||
if status not in expect:
|
||||
raise BotError(f"gitea: {method} {path} -> HTTP {status}")
|
||||
if len(raw) > MAX_API_BYTES:
|
||||
raise BotError(f"gitea: {method} {path}: response exceeds its bound")
|
||||
if not raw:
|
||||
return None
|
||||
try:
|
||||
return json.loads(raw)
|
||||
except ValueError as exc:
|
||||
raise BotError(f"gitea: {method} {path}: response is not JSON") from exc
|
||||
|
||||
def _list(self, path: str, query: dict | None = None) -> list:
|
||||
items: list = []
|
||||
for page in range(1, MAX_PAGES + 1):
|
||||
batch = self._call("GET", path, {**(query or {}), "limit": PAGE_SIZE, "page": page})
|
||||
if not isinstance(batch, list):
|
||||
raise BotError(f"gitea: GET {path}: expected a list")
|
||||
items += batch
|
||||
if len(batch) < PAGE_SIZE:
|
||||
return items
|
||||
raise BotError(f"gitea: GET {path}: more than {MAX_PAGES} pages")
|
||||
|
||||
def pr_url(self, number: int) -> str:
|
||||
return f"{self._web}/pulls/{int(number)}"
|
||||
|
||||
def labels(self) -> list:
|
||||
return self._list("/labels")
|
||||
|
||||
def create_label(self, name: str, color: str) -> dict:
|
||||
return self._call("POST", "/labels", payload={"name": name, "color": color}, expect=(201,))
|
||||
|
||||
def issues_with_label(self, label: str) -> list:
|
||||
return self._list("/issues", {"state": "all", "type": "issues", "labels": label})
|
||||
|
||||
def issues_matching(self, text: str) -> list:
|
||||
return self._list("/issues", {"state": "all", "type": "issues", "q": text})
|
||||
|
||||
def create_issue(self, title: str, body: str, label_id: int) -> dict:
|
||||
return self._call("POST", "/issues", payload={"title": title, "body": body, "labels": [label_id]}, expect=(201,))
|
||||
|
||||
def update_issue(self, number: int, fields: dict) -> None:
|
||||
self._call("PATCH", f"/issues/{int(number)}", payload=fields, expect=(200, 201))
|
||||
|
||||
def add_issue_label(self, number: int, label_id: int) -> None:
|
||||
self._call("POST", f"/issues/{int(number)}/labels", payload={"labels": [label_id]}, expect=(200,))
|
||||
|
||||
def get_issue(self, number: int) -> dict:
|
||||
issue = self._call("GET", f"/issues/{int(number)}")
|
||||
if not isinstance(issue, dict):
|
||||
raise BotError("gitea: issue response is not an object")
|
||||
return issue
|
||||
|
||||
def open_renovate_prs(self) -> list[dict]:
|
||||
result = []
|
||||
for pr in self._list("/pulls", {"state": "open"}):
|
||||
head = pr.get("head") if isinstance(pr, dict) else None
|
||||
ref = head.get("ref") if isinstance(head, dict) else None
|
||||
number = pr.get("number") if isinstance(pr, dict) else None
|
||||
title = pr.get("title") if isinstance(pr, dict) else None
|
||||
if not isinstance(ref, str) or not isinstance(number, int) or isinstance(number, bool):
|
||||
raise BotError("gitea: a pull request has an unexpected shape")
|
||||
if ref.startswith("renovate/"):
|
||||
package = _title_package(title) if isinstance(title, str) else ""
|
||||
result.append({"number": number, "branch": ref, "package": package})
|
||||
return result
|
||||
|
||||
|
||||
class ScannerClient:
|
||||
"""The scanner's internal API. Falls back to the pinned ClusterIP on a DNS failure only."""
|
||||
|
||||
def __init__(self, primary: str, fallback: str, transport) -> None:
|
||||
self._bases = [base.rstrip("/") for base in (primary, fallback) if base]
|
||||
self._transport = transport
|
||||
self.used_fallback = False
|
||||
|
||||
def _call(self, method: str, path: str, body: bytes | None, headers: dict, max_bytes: int):
|
||||
last = len(self._bases) - 1
|
||||
for index, base in enumerate(self._bases):
|
||||
try:
|
||||
result = self._transport(method, base + path, headers, body, max_bytes)
|
||||
except DnsFailure as exc:
|
||||
if index == last:
|
||||
raise BotError(f"scanner: {method} {path}: name resolution failed") from exc
|
||||
continue
|
||||
if index:
|
||||
self.used_fallback = True
|
||||
self._bases = self._bases[index:]
|
||||
return result
|
||||
raise BotError("scanner: no address configured")
|
||||
|
||||
def report(self) -> dict:
|
||||
status, raw = self._call("GET", "/report.json", None, {"Accept": "application/json"}, MAX_REPORT_BYTES)
|
||||
if status != 200:
|
||||
raise BotError(f"scanner: GET /report.json -> HTTP {status}")
|
||||
if len(raw) > MAX_REPORT_BYTES:
|
||||
raise BotError("scanner: report exceeds its bound")
|
||||
return parse_report(raw)
|
||||
|
||||
def triage(self, heartbeat: str, remediation: list[dict]) -> None:
|
||||
if len(remediation) > MAX_REMEDIATION:
|
||||
raise BotError("scanner: more remediation entries than POST /triage accepts")
|
||||
body = json.dumps({"heartbeatAt": heartbeat, "remediation": remediation}).encode()
|
||||
status, _ = self._call("POST", "/triage", body, {"Content-Type": "application/json"}, 4096)
|
||||
if status != 204:
|
||||
raise BotError(f"scanner: POST /triage -> HTTP {status}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- orchestration
|
||||
|
||||
|
||||
def _label_names(issue: dict) -> set:
|
||||
return {label.get("name") for label in (issue.get("labels") or []) if isinstance(label, dict)}
|
||||
|
||||
|
||||
def find_dashboard_issues(gitea: GiteaClient) -> list[dict]:
|
||||
"""Every issue that is the dashboard: by label, and by exact title in case a label was removed."""
|
||||
found: dict[int, dict] = {}
|
||||
for issue in gitea.issues_with_label(DASHBOARD_LABEL) + gitea.issues_matching(DASHBOARD_TITLE):
|
||||
if (
|
||||
isinstance(issue, dict)
|
||||
and issue.get("title") == DASHBOARD_TITLE
|
||||
and not issue.get("pull_request")
|
||||
and isinstance(issue.get("number"), int)
|
||||
):
|
||||
found[issue["number"]] = issue
|
||||
return list(found.values())
|
||||
|
||||
|
||||
def sync_dashboard_issue(gitea: GiteaClient, body: str, dry_run: bool) -> dict:
|
||||
"""Create or update the one dashboard issue. Returns counts of what it did or would do."""
|
||||
if DASHBOARD_LABEL in FORBIDDEN_LABELS:
|
||||
raise BotError("config: the dashboard label is one the PM agent acts on")
|
||||
plan = {"label_created": 0, "issue_created": 0, "issue_updated": 0, "issue_reopened": 0}
|
||||
existing = find_dashboard_issues(gitea)
|
||||
if len(existing) > 1:
|
||||
# Closing an issue on a guess is worse than a red run naming the numbers to look at.
|
||||
numbers = sorted(issue["number"] for issue in existing)
|
||||
raise BotError(f"dashboard: {len(numbers)} issues titled {DASHBOARD_TITLE!r}: {numbers}")
|
||||
|
||||
label = next((x for x in gitea.labels() if isinstance(x, dict) and x.get("name") == DASHBOARD_LABEL), None)
|
||||
if label is None:
|
||||
plan["label_created"] = 1
|
||||
if not dry_run:
|
||||
label = gitea.create_label(DASHBOARD_LABEL, DASHBOARD_LABEL_COLOR)
|
||||
|
||||
if not existing:
|
||||
plan["issue_created"] = 1
|
||||
if not dry_run:
|
||||
created = gitea.create_issue(DASHBOARD_TITLE, body, label["id"])
|
||||
if not isinstance(created, dict) or not isinstance(created.get("number"), int):
|
||||
raise BotError("gitea: issue creation returned an unexpected shape")
|
||||
return plan
|
||||
|
||||
issue = existing[0]
|
||||
fields: dict = {}
|
||||
if issue.get("body") != body:
|
||||
fields["body"] = body
|
||||
if issue.get("state") != "open":
|
||||
fields["state"] = "open"
|
||||
plan["issue_reopened"] = 1
|
||||
if fields:
|
||||
plan["issue_updated"] = 1
|
||||
if not dry_run:
|
||||
if DASHBOARD_LABEL not in _label_names(issue):
|
||||
gitea.add_issue_label(issue["number"], label["id"])
|
||||
if fields:
|
||||
gitea.update_issue(issue["number"], fields)
|
||||
return plan
|
||||
|
||||
|
||||
def _renovate_issue(gitea: GiteaClient, number: int) -> dict:
|
||||
issue = gitea.get_issue(number)
|
||||
body = issue.get("body")
|
||||
if issue.get("title") != RENOVATE_DASHBOARD_TITLE or issue.get("state") != "open" or not isinstance(body, str):
|
||||
raise BotError(f"renovate dashboard: issue {number} is not the open Dependency Dashboard")
|
||||
return issue
|
||||
|
||||
|
||||
def report_images(report: dict) -> list[str]:
|
||||
return [t["image"] for t in report["targets"] if t["type"] == "image"]
|
||||
|
||||
|
||||
def run(cfg: dict, gitea: GiteaClient, scanner: ScannerClient, now: datetime, emit: Callable[[str], None]) -> dict:
|
||||
"""One triage pass. Any failure raises, and the heartbeat is the last thing it writes."""
|
||||
dry_run = cfg["dry_run"]
|
||||
report = scanner.report()
|
||||
agg = aggregate(report)
|
||||
now_text = now.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
|
||||
summary: dict = {"dry_run": dry_run, "targets": agg["targets_total"], "critical_with_fix": agg["critical_with_fix"]}
|
||||
summary["dashboard"] = sync_dashboard_issue(gitea, render_dashboard(agg, cfg["page_url"], now), dry_run)
|
||||
|
||||
number = cfg["renovate_issue"]
|
||||
body = _renovate_issue(gitea, number)["body"]
|
||||
remediation = derive_remediation(
|
||||
report_images(report), parse_rows(body), gitea.open_renovate_prs(), parse_detected(body),
|
||||
gitea.pr_url, now_text,
|
||||
)
|
||||
summary["remediation"] = count_states(remediation)
|
||||
|
||||
# Re-read immediately before writing: Renovate rewrites this body, and a tick must apply to
|
||||
# the body as it is now.
|
||||
fresh = _renovate_issue(gitea, number)["body"]
|
||||
ticks = select_ticks(parse_rows(fresh), qualifying_images(report))
|
||||
summary["ticks"] = len(ticks)
|
||||
if ticks and not dry_run:
|
||||
gitea.update_issue(number, {"body": apply_ticks(fresh, ticks)})
|
||||
|
||||
if not dry_run:
|
||||
scanner.triage(now_text, remediation)
|
||||
summary["heartbeat"] = 0 if dry_run else 1
|
||||
summary["fallback_used"] = scanner.used_fallback
|
||||
emit(render_summary(summary))
|
||||
return summary
|
||||
|
||||
|
||||
def render_summary(summary: dict) -> str:
|
||||
verb = "would be" if summary["dry_run"] else "were"
|
||||
dash = summary["dashboard"]
|
||||
states = ", ".join(f"{name} {count}" for name, count in summary["remediation"].items())
|
||||
return "\n".join([
|
||||
f"### CVE triage ({'DRY RUN, no writes' if summary['dry_run'] else 'live'})",
|
||||
"",
|
||||
f"- targets in report: {summary['targets']}",
|
||||
f"- unaccepted CRITICAL with a fix: {summary['critical_with_fix']}",
|
||||
f"- dashboard label created: {dash['label_created']}; issue created: {dash['issue_created']};"
|
||||
f" updated: {dash['issue_updated']}; reopened: {dash['issue_reopened']} ({verb} applied)",
|
||||
f"- remediation states: {states}",
|
||||
f"- approval boxes that {verb} ticked: {summary['ticks']}",
|
||||
f"- heartbeat posted: {summary['heartbeat']}",
|
||||
f"- scanner reached through the ClusterIP fallback: {summary['fallback_used']}",
|
||||
"",
|
||||
])
|
||||
|
||||
|
||||
def load_config(env: dict) -> dict:
|
||||
token = env.get("RENOVATE_TOKEN", "")
|
||||
if not token:
|
||||
raise BotError("RENOVATE_TOKEN is not set")
|
||||
repo = env.get("GITEA_REPOSITORY", "Home-Network/home-network-k3s-impl")
|
||||
if not REPO_RE.fullmatch(repo):
|
||||
raise BotError("config: GITEA_REPOSITORY is malformed")
|
||||
urls = {
|
||||
"gitea_url": env.get("GITEA_URL", "https://gitea.home.keithjackson.dev"),
|
||||
"vuln_url": env.get("VULN_API_URL", "http://vuln-scanner.vuln-scanner.svc.cluster.local:8081"),
|
||||
"page_url": env.get("VULN_PAGE_URL", "https://cve.home.keithjackson.dev"),
|
||||
}
|
||||
fallback = env.get("VULN_API_FALLBACK_URL", "")
|
||||
for name, value in {**urls, **({"fallback": fallback} if fallback else {})}.items():
|
||||
if not HTTP_URL_RE.fullmatch(value):
|
||||
raise BotError(f"config: {name} is not a plain http(s) URL")
|
||||
dry_run = env.get("DRY_RUN", "false")
|
||||
if dry_run not in ("true", "false"):
|
||||
raise BotError("config: DRY_RUN must be true or false")
|
||||
issue = env.get("RENOVATE_DASHBOARD_ISSUE", "91")
|
||||
if not issue.isdigit():
|
||||
raise BotError("config: RENOVATE_DASHBOARD_ISSUE is not a number")
|
||||
return {**urls, "token": token, "repo": repo, "fallback": fallback,
|
||||
"dry_run": dry_run == "true", "renovate_issue": int(issue)}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
def emit(text: str) -> None:
|
||||
print(text)
|
||||
path = os.environ.get("GITHUB_STEP_SUMMARY")
|
||||
if path:
|
||||
with open(path, "a", encoding="utf-8") as handle:
|
||||
handle.write(text + "\n")
|
||||
|
||||
try:
|
||||
cfg = load_config(dict(os.environ))
|
||||
transport = UrllibTransport()
|
||||
gitea = GiteaClient(cfg["gitea_url"], cfg["repo"], cfg["token"], transport)
|
||||
scanner = ScannerClient(cfg["vuln_url"], cfg["fallback"], transport)
|
||||
run(cfg, gitea, scanner, datetime.now(timezone.utc), emit)
|
||||
except BotError as exc:
|
||||
print(f"::error::CVE triage failed: {exc}", file=sys.stderr)
|
||||
emit(f"### CVE triage FAILED\n\n{exc}\n")
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in new issue
Block a user