fejkdata/vulnerability-filing/file_vulnerabilities.py

#!/usr/bin/env python3
"""File each module's known vulnerabilities as one security item atop todo.md's next release.

Usage: file_vulnerabilities.py <govulncheck -format json report> <todo.md>
Exits 1 while any vulnerability is known, filed now or before, 0 on a clean report, and 2 on
a todo.md it cannot read.

todo.md's `Next ID = N` gives the item's ID, and the release of the plan's top row, the next
release, gives its release. Its Score is 0.0 until rank_todo.py, which CI cannot run, scores
it; a security item ranks first in its release whatever it scores. The test that files into
the repository's own todo.md fails the gate when the plan's shape drifts from what this
script reads.

Scores, as todo.md's table spells them: R (risk), S (size) and C (difficulty) are 1, since
the fix is a version bump; A (value to the audience) and G (value to the goal) are 8, the goal
being README goal 1.1, installing fejkdata the way one installs other tools, which a flagged
binary works against.
"""

import json
import re
import sys
from dataclasses import dataclass, field

ROW = re.compile(r"^\| (?P<id>\d+) \| (?P<release>[^|]+?) \| [^|]*\| \*\*", re.M)
NEXT_ID = re.compile(r"`Next ID = (\d+)`")
DETAILS = "## Details\n\n"
SCORES = {"R": 1, "S": 1, "C": 1, "A": 8, "G": 8}
GOAL = "1.1"


@dataclass
class Module:
	path: str
	found_in: str
	fixed: list[str] = field(default_factory=list)
	unfixed: bool = False
	osvs: dict[str, dict] = field(default_factory=dict)

	@property
	def name(self) -> str:
		return "Go" if self.path == "stdlib" else f"`{self.path}`"

	def version(self, version: str) -> str:
		return version.removeprefix("v") if self.path == "stdlib" else version

	@property
	def carried(self) -> str:
		return f"{self.name} {self.version(self.found_in)}"


def messages(report: str):
	decoder, at = json.JSONDecoder(), 0
	while at < len(report):
		if report[at].isspace():
			at += 1
			continue
		message, at = decoder.raw_decode(report, at)
		yield message


def vulnerable_modules(report: str) -> list[Module]:
	"""Each module govulncheck found a vulnerability in, Go (stdlib) first, then by path."""
	osvs, modules = {}, {}
	for message in messages(report):
		if "osv" in message:
			osv = message["osv"]
			osvs[osv["id"]] = {"id": osv["id"], "aliases": osv.get("aliases", []), "summary": osv.get("summary", "")}
		elif "finding" in message:
			finding = message["finding"]
			frame = finding["trace"][0]
			module = modules.setdefault(frame["module"], Module(frame["module"], frame["version"]))
			module.osvs[finding["osv"]] = osvs[finding["osv"]]
			if "fixed_version" in finding:
				module.fixed.append(finding["fixed_version"])
			else:
				module.unfixed = True
	return sorted(modules.values(), key=lambda m: (m.path != "stdlib", m.path))


def version_key(version: str):
	return [int(part) for part in re.findall(r"\d+", version)]


def tagline(module: Module) -> str:
	if module.unfixed:
		return f"Ship no release while fejkdata carries {module.carried}, some of whose known vulnerabilities no version fixes yet."
	fixed = module.version(max(module.fixed, key=version_key))
	if module.path == "stdlib":
		return f"Raise Go to {fixed}, so the release binaries carry none of {module.carried}'s known vulnerabilities."
	return f"Raise {module.name} to {fixed}, so fejkdata requires no version of it with a known vulnerability."


def detail(module: Module) -> str:
	lines = []
	for osv in sorted(module.osvs.values(), key=lambda o: o["id"]):
		aliases = f" ({', '.join(osv['aliases'])})" if osv["aliases"] else ""
		summary = f": {osv['summary']}" if osv["summary"] else ""
		lines.append(f"- {osv['id']}{aliases}{summary}")
	return f"govulncheck found these in {module.carried}:\n\n" + "\n".join(lines) + "\n"


def file_items(modules: list[Module], todo: str) -> tuple[str, list[int]]:
	"""Return todo.md with an item for each module whose vulnerabilities it does not yet name, and the items' IDs."""
	first = ROW.search(todo)
	if first is None:
		raise ValueError("todo.md has no item to take the next release from")
	if NEXT_ID.search(todo) is None:
		raise ValueError("todo.md has no `Next ID = N` line to number the item by")
	if DETAILS not in todo:
		raise ValueError(f"todo.md has no {DETAILS.strip()!r} heading to put the item's detail under")
	release, rows_at = first.group("release"), first.start()
	details_at = todo.index(DETAILS) + len(DETAILS)
	filed = []
	for module in modules:
		if all(osv in todo for osv in module.osvs):
			continue
		next_id = int(NEXT_ID.search(todo).group(1))
		line = tagline(module)
		scores = " | ".join(str(SCORES[k]) for k in "RSCAG")
		row = f"| {next_id} | {release} | security | **{line}** | {scores} | {GOAL} | 0.0 |\n"
		text = f"### {next_id}. {line}\n\n{detail(module)}\n"
		todo = todo[:rows_at] + row + todo[rows_at:]
		rows_at += len(row)
		details_at += len(row)
		todo = todo[:details_at] + text + todo[details_at:]
		details_at += len(text)
		todo = NEXT_ID.sub(f"`Next ID = {next_id + 1}`", todo, count=1)
		filed.append(next_id)
	return todo, filed


def main(argv: list[str]) -> int:
	report_path, todo_path = argv
	with open(report_path, encoding="utf-8") as f:
		modules = vulnerable_modules(f.read())
	if not modules:
		print("No known vulnerabilities.")
		return 0
	count = sum(len(m.osvs) for m in modules)
	known = f"{count} known {'vulnerability' if count == 1 else 'vulnerabilities'}"
	with open(todo_path, encoding="utf-8") as f:
		try:
			todo, filed = file_items(modules, f.read())
		except ValueError as err:
			print(err, file=sys.stderr)
			return 2
	if not filed:
		print(f"{known}, already planned in todo.md.")
		return 1
	with open(todo_path, "w", encoding="utf-8") as f:
		f.write(todo)
	items = ", ".join(map(str, filed))
	print(f"{known}; todo.md {'item ' + items + ' plans' if len(filed) == 1 else 'items ' + items + ' plan'} the fix.")
	return 1


if __name__ == "__main__":
	sys.exit(main(sys.argv[1:]))