Give the vulnerability filer a folder of its own, and leave its item's score to rank_todo.py
This commit is contained in:
@@ -1,170 +0,0 @@
|
||||
#!/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, and 0 on a clean report.
|
||||
|
||||
The item is written the way ~/.claude/bin/rank_todo.py writes one, which CI cannot run:
|
||||
todo.md's `Next ID = N` gives its ID, the release of the plan's top row (the next release)
|
||||
gives its release, and its Score is todo.md's `Score = …` formula over the scores below and
|
||||
the weight todo.md's goal table gives goal 1.1. 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.
|
||||
|
||||
A security item scores R, S and C 1, since the fix is a version bump, and A and G 8, since a
|
||||
known vulnerability in the binaries works against README goal 1.1: installing fejkdata the
|
||||
way one installs other tools, which image scanners then flag.
|
||||
"""
|
||||
|
||||
import ast
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
ROW = re.compile(r"^\| (\d+) \| ([^|]+?) \| [^|]*\| \*\*", re.M)
|
||||
WEIGHT = re.compile(r"^\| ([\d.]+) \| ([\d.]+) \|$", re.M)
|
||||
NEXT_ID = re.compile(r"`Next ID = (\d+)`")
|
||||
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 evaluate(node, values):
|
||||
"""The value of todo.md's score formula: numbers, letters, unary minus and + - * /."""
|
||||
match node:
|
||||
case ast.Constant(value=value):
|
||||
return value
|
||||
case ast.Name(id=name):
|
||||
return values[name]
|
||||
case ast.UnaryOp(op=ast.USub(), operand=operand):
|
||||
return -evaluate(operand, values)
|
||||
case ast.BinOp(left=left, op=ast.Add(), right=right):
|
||||
return evaluate(left, values) + evaluate(right, values)
|
||||
case ast.BinOp(left=left, op=ast.Sub(), right=right):
|
||||
return evaluate(left, values) - evaluate(right, values)
|
||||
case ast.BinOp(left=left, op=ast.Mult(), right=right):
|
||||
return evaluate(left, values) * evaluate(right, values)
|
||||
case ast.BinOp(left=left, op=ast.Div(), right=right):
|
||||
return evaluate(left, values) / evaluate(right, values)
|
||||
raise ValueError(f"todo.md's formula holds {ast.dump(node)}, which this script cannot evaluate")
|
||||
|
||||
|
||||
def score(todo: str) -> str:
|
||||
formula = re.search(r"`Score = (.+?)`", todo).group(1)
|
||||
values = dict(SCORES, W=float(dict(WEIGHT.findall(todo))[GOAL]))
|
||||
return f"{evaluate(ast.parse(formula, mode='eval').body, values):.1f}"
|
||||
|
||||
|
||||
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."""
|
||||
filed = []
|
||||
for module in modules:
|
||||
if all(osv in todo for osv in module.osvs):
|
||||
continue
|
||||
first = ROW.search(todo)
|
||||
if first is None:
|
||||
raise ValueError("todo.md has no item to take the next release from")
|
||||
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} | {first.group(2)} | security | **{line}** | {scores} | {GOAL} | {score(todo)} |\n"
|
||||
todo = todo[: first.start()] + row + todo[first.start() :]
|
||||
todo = todo.replace(f"`Next ID = {next_id}`", f"`Next ID = {next_id + 1}`", 1)
|
||||
todo = todo.replace("## Details\n\n", f"## Details\n\n### {next_id}. {line}\n\n{detail(module)}\n", 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:
|
||||
todo, filed = file_items(modules, f.read())
|
||||
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:]))
|
||||
@@ -1,177 +0,0 @@
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from contextlib import redirect_stdout
|
||||
|
||||
import file_vulnerabilities
|
||||
|
||||
TODO = """# todo
|
||||
|
||||
## Scoring
|
||||
|
||||
`Score = -R - S/4 + 2*A + 2*G*W`
|
||||
|
||||
`Bar = 9`
|
||||
|
||||
`Next ID = 7`
|
||||
|
||||
| Goal | W |
|
||||
|---|---|
|
||||
| 1 | 1.00 |
|
||||
| 1.1 | 0.90 |
|
||||
|
||||
## Items
|
||||
|
||||
| ID | Release | Exempt | Item | R | S | C | A | G | Goals | Score |
|
||||
|---|---|---|---|---|---|---|---|---|---|---|
|
||||
| 5 | 0.2.0 | | **Add a thing.** | 2 | 2 | 2 | 7 | 8 | 1 | 26.5 |
|
||||
|
||||
## Details
|
||||
|
||||
### 5. Add a thing.
|
||||
|
||||
Why.
|
||||
"""
|
||||
|
||||
OSVS = (
|
||||
{"id": "GO-2026-0002", "summary": "Denial of service in net/http", "aliases": ["GHSA-2222", "CVE-2026-2"]},
|
||||
{"id": "GO-2026-0001", "summary": "Cross-site scripting in html/template", "aliases": ["CVE-2026-1"]},
|
||||
{"id": "GO-2026-0003", "summary": "Panic in x/text"},
|
||||
{"id": "GO-2026-0004"},
|
||||
)
|
||||
STDLIB = (
|
||||
("GO-2026-0002", "stdlib", "v1.27.1", "v1.27.2"),
|
||||
("GO-2026-0001", "stdlib", "v1.27.1", "v1.27.10"),
|
||||
("GO-2026-0001", "stdlib", "v1.27.1", "v1.27.10"),
|
||||
)
|
||||
TEXT = (("GO-2026-0003", "golang.org/x/text", "v0.3.0", "v0.3.8"),)
|
||||
GO_TAGLINE = "Raise Go to 1.27.10, so the release binaries carry none of Go 1.27.1's known vulnerabilities."
|
||||
|
||||
|
||||
def report(*findings):
|
||||
messages = [{"config": {"go_version": "go1.27.1", "scan_level": "module"}}]
|
||||
messages += [{"osv": osv} for osv in OSVS]
|
||||
for osv, module, version, fixed in findings:
|
||||
finding = {"osv": osv, "trace": [{"module": module, "version": version}]}
|
||||
if fixed:
|
||||
finding["fixed_version"] = fixed
|
||||
messages.append({"finding": finding})
|
||||
return "\n".join(json.dumps(m, indent=2) for m in messages)
|
||||
|
||||
|
||||
def file(todo, *findings):
|
||||
return file_vulnerabilities.file_items(file_vulnerabilities.vulnerable_modules(report(*findings)), todo)
|
||||
|
||||
|
||||
def rows(todo):
|
||||
return [line for line in todo.splitlines() if "| **" in line]
|
||||
|
||||
|
||||
class FileItems(unittest.TestCase):
|
||||
def test_no_finding_files_nothing(self):
|
||||
self.assertEqual(file(TODO), (TODO, []))
|
||||
|
||||
def test_files_one_security_item_per_module_atop_the_next_release(self):
|
||||
todo, filed = file(TODO, *STDLIB)
|
||||
self.assertEqual(filed, [7])
|
||||
self.assertIn("`Next ID = 8`", todo)
|
||||
self.assertEqual(rows(todo)[0], f"| 7 | 0.2.0 | security | **{GO_TAGLINE}** | 1 | 1 | 1 | 8 | 8 | 1.1 | 29.1 |")
|
||||
self.assertTrue(rows(todo)[1].startswith("| 5 |"))
|
||||
self.assertTrue(
|
||||
todo.split("## Details\n\n", 1)[1].startswith(
|
||||
f"### 7. {GO_TAGLINE}\n\n"
|
||||
"govulncheck found these in Go 1.27.1:\n\n"
|
||||
"- GO-2026-0001 (CVE-2026-1): Cross-site scripting in html/template\n"
|
||||
"- GO-2026-0002 (GHSA-2222, CVE-2026-2): Denial of service in net/http\n\n"
|
||||
"### 5. Add a thing."
|
||||
)
|
||||
)
|
||||
|
||||
def test_a_required_module_gets_its_own_item(self):
|
||||
todo, filed = file(TODO, *STDLIB, *TEXT)
|
||||
self.assertEqual(filed, [7, 8])
|
||||
self.assertIn(
|
||||
"| 8 | 0.2.0 | security | **Raise `golang.org/x/text` to v0.3.8, so fejkdata requires no version"
|
||||
" of it with a known vulnerability.** |",
|
||||
todo,
|
||||
)
|
||||
self.assertIn("govulncheck found these in `golang.org/x/text` v0.3.0:\n\n- GO-2026-0003: Panic in x/text\n", todo)
|
||||
self.assertIn("`Next ID = 9`", todo)
|
||||
|
||||
def test_a_vulnerability_already_planned_is_not_filed_again(self):
|
||||
planned, _ = file(TODO, *STDLIB)
|
||||
self.assertEqual(file(planned, *STDLIB), (planned, []))
|
||||
|
||||
def test_a_module_newly_vulnerable_is_filed_beside_one_already_planned(self):
|
||||
planned, _ = file(TODO, *STDLIB)
|
||||
todo, filed = file(planned, *STDLIB, *TEXT)
|
||||
self.assertEqual(filed, [8])
|
||||
self.assertIn("| 8 | 0.2.0 | security | **Raise `golang.org/x/text`", todo)
|
||||
|
||||
def test_a_vulnerability_no_version_fixes_names_no_target(self):
|
||||
todo, _ = file(TODO, *STDLIB, ("GO-2026-0004", "stdlib", "v1.27.1", ""))
|
||||
self.assertIn("- GO-2026-0004\n", todo)
|
||||
self.assertIn(
|
||||
"| 7 | 0.2.0 | security | **Ship no release while fejkdata carries Go 1.27.1, some of whose known"
|
||||
" vulnerabilities no version fixes yet.** |",
|
||||
todo,
|
||||
)
|
||||
|
||||
def test_a_plan_with_no_items_is_refused(self):
|
||||
empty = TODO.split("| 5 |")[0].rstrip("\n") + "\n\n## Details\n"
|
||||
with self.assertRaises(ValueError):
|
||||
file(empty, *STDLIB)
|
||||
|
||||
def test_the_repositorys_plan_takes_an_item_in_its_own_shape(self):
|
||||
path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "todo.md")
|
||||
with open(path, encoding="utf-8") as f:
|
||||
plan = f.read()
|
||||
todo, filed = file(plan, *STDLIB)
|
||||
header = next(line for line in plan.splitlines() if line.startswith("| ID |"))
|
||||
row = rows(todo)[0]
|
||||
self.assertEqual(row.count("|"), header.count("|"))
|
||||
self.assertEqual(row.split(" | ")[1], rows(plan)[0].split(" | ")[1])
|
||||
self.assertRegex(row, rf"^\| {filed[0]} \| .* \| \d+\.\d \|$")
|
||||
self.assertIn(f"`Next ID = {filed[0] + 1}`", todo)
|
||||
|
||||
|
||||
class Main(unittest.TestCase):
|
||||
def setUp(self):
|
||||
cwd = os.getcwd()
|
||||
tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(tmp.cleanup)
|
||||
os.chdir(tmp.name)
|
||||
self.addCleanup(os.chdir, cwd)
|
||||
with open("todo.md", "w", encoding="utf-8") as f:
|
||||
f.write(TODO)
|
||||
|
||||
def run_main(self, *findings):
|
||||
with open("report.json", "w", encoding="utf-8") as f:
|
||||
f.write(report(*findings))
|
||||
out = io.StringIO()
|
||||
with redirect_stdout(out):
|
||||
code = file_vulnerabilities.main(["report.json", "todo.md"])
|
||||
with open("todo.md", encoding="utf-8") as f:
|
||||
return code, out.getvalue(), f.read()
|
||||
|
||||
def test_a_clean_report_passes_and_leaves_the_plan(self):
|
||||
self.assertEqual(self.run_main(), (0, "No known vulnerabilities.\n", TODO))
|
||||
|
||||
def test_a_finding_fails_and_files_it(self):
|
||||
code, out, todo = self.run_main(*STDLIB)
|
||||
self.assertEqual((code, out), (1, "2 known vulnerabilities; todo.md item 7 plans the fix.\n"))
|
||||
self.assertIn("| 7 | 0.2.0 | security |", todo)
|
||||
|
||||
def test_findings_in_two_modules_name_both_items(self):
|
||||
_, out, _ = self.run_main(*STDLIB, *TEXT)
|
||||
self.assertEqual(out, "3 known vulnerabilities; todo.md items 7, 8 plan the fix.\n")
|
||||
|
||||
def test_one_finding_already_planned_still_fails(self):
|
||||
self.run_main(*TEXT)
|
||||
self.assertEqual(self.run_main(*TEXT)[:2], (1, "1 known vulnerability, already planned in todo.md.\n"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user