From 77d447c18b56558a89ea9b638e7a1515332a0d57 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 13:51:20 +0000 Subject: [PATCH 1/4] Pin Python 3.12, make ruff a real dev dependency, format the codebase .python-version pointed at 3.14 (a release candidate). With the locked pydantic, FastAPI fails to import there, so `patchowner serve` crashes and one test fails on a fresh checkout. Pin 3.12, which CLAUDE.md and requires-python already assume; 3.13 passes too. CLAUDE.md and the repo hook both call `uv run ruff`, but ruff was not a dependency, so the hook failed on every edit. Add ruff to the dev group with an explicit config in pyproject.toml, run check --fix and format once across the codebase, and replace the hook command (which relied on an empty CLAUDE_FILE_PATHS variable) with a small script that reads the edited path from the hook's stdin JSON. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_012ZCSLekTvZQ7X3PsAUJuXk --- .claude/hooks/lint_changed.py | 36 ++++++ .claude/hooks/test_gate.py | 13 +- .claude/settings.json | 4 +- .python-version | 2 +- patchowner/decide.py | 64 +++++++--- patchowner/engine.py | 29 ++++- patchowner/health.py | 223 ++++++++++++++++++++++++++-------- patchowner/inventory.py | 32 +++-- patchowner/matching.py | 20 ++- patchowner/state.py | 12 +- pyproject.toml | 13 ++ tests/test_decide.py | 14 ++- tests/test_engine.py | 26 ++-- tests/test_health.py | 46 +++++-- tests/test_matching.py | 73 +++++++---- tests/test_routing.py | 142 +++++++++++++++++----- tests/test_ssvc.py | 45 ++++--- tests/test_state.py | 48 ++++++-- uv.lock | 27 ++++ 19 files changed, 665 insertions(+), 204 deletions(-) create mode 100755 .claude/hooks/lint_changed.py mode change 100644 => 100755 .claude/hooks/test_gate.py diff --git a/.claude/hooks/lint_changed.py b/.claude/hooks/lint_changed.py new file mode 100755 index 0000000..c05c89a --- /dev/null +++ b/.claude/hooks/lint_changed.py @@ -0,0 +1,36 @@ +#!/usr/bin/env python3 +"""PostToolUse hook: lint and format the Python file Claude just edited. + +Claude Code hands the tool call to hooks as JSON on stdin. The edited path is +tool_input.file_path. Non-Python files and files outside the repo are ignored. +""" + +import json +import pathlib +import subprocess +import sys + + +def main() -> int: + try: + payload = json.load(sys.stdin) + except (ValueError, OSError): + return 0 + path = (payload.get("tool_input") or {}).get("file_path", "") + if not path or not path.endswith(".py"): + return 0 + file = pathlib.Path(path) + repo = pathlib.Path(__file__).resolve().parents[2] + if not file.exists() or repo not in file.resolve().parents: + return 0 + for cmd in (["uv", "run", "ruff", "check", "--fix", str(file)], ["uv", "run", "ruff", "format", str(file)]): + r = subprocess.run(cmd, capture_output=True, text=True, check=False) + if r.returncode != 0: + # Report the lint findings back to Claude rather than failing silently. + print((r.stdout + r.stderr).strip()[-1500:], file=sys.stderr) + return 2 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.claude/hooks/test_gate.py b/.claude/hooks/test_gate.py old mode 100644 new mode 100755 index ecef7a2..4f82261 --- a/.claude/hooks/test_gate.py +++ b/.claude/hooks/test_gate.py @@ -3,6 +3,7 @@ import subprocess import sys + def main(): try: input_data = json.load(sys.stdin) @@ -14,17 +15,11 @@ def main(): sys.exit(0) # Execute test suite via uv run pytest - result = subprocess.run( - ["uv", "run", "pytest"], capture_output=True, text=True, timeout=60 - ) + result = subprocess.run(["uv", "run", "pytest"], capture_output=True, text=True, timeout=60) if result.returncode != 0: # Extract last 1000 characters of test output for context - stderr_output = ( - result.stderr[-1000:] - if result.stderr - else result.stdout[-1000:] - ) + stderr_output = result.stderr[-1000:] if result.stderr else result.stdout[-1000:] output = { "decision": "block", "reason": f"Tests are failing. Fix assertions before completing:\n{stderr_output}", @@ -34,6 +29,6 @@ def main(): sys.exit(0) + if __name__ == "__main__": main() - diff --git a/.claude/settings.json b/.claude/settings.json index 7c5ca29..807c01c 100644 --- a/.claude/settings.json +++ b/.claude/settings.json @@ -6,8 +6,8 @@ "hooks": [ { "type": "command", - "command": "uv run ruff check --fix \"$CLAUDE_FILE_PATHS\" && uv run ruff format \"$CLAUDE_FILE_PATHS\"", - "timeout": 15 + "command": "python3 .claude/hooks/lint_changed.py", + "timeout": 30 } ] } diff --git a/.python-version b/.python-version index 6324d40..e4fba21 100644 --- a/.python-version +++ b/.python-version @@ -1 +1 @@ -3.14 +3.12 diff --git a/patchowner/decide.py b/patchowner/decide.py index fed3532..dd3c3cb 100644 --- a/patchowner/decide.py +++ b/patchowner/decide.py @@ -1,4 +1,5 @@ """Turn matches into decisions: SSVC outcome, urgency words, recipient, escalation, suppression.""" + from __future__ import annotations from collections import Counter, defaultdict @@ -26,6 +27,7 @@ @dataclass(frozen=True) class Delivery: """What an outcome means for people. This is the THEN half of a policy, and it lives on the leaf.""" + notify_oncall: bool acknowledge_within: str plan_within: str @@ -64,15 +66,15 @@ class Decision: recipient_email: str recipient_name: str recipient_is_fallback: bool - assessment: Assessment | None = None # None for "possible" matches: no path until a human confirms + assessment: Assessment | None = None # None for "possible" matches: no path until a human confirms outcome: str | None = None policy_row: int | None = None vector: str = "" - fixers: list[Recipient] = field(default_factory=list) # get the full card - accountable: Recipient | None = None # gets one status line - escalation: Recipient | None = None # hears only if unresolved + fixers: list[Recipient] = field(default_factory=list) # get the full card + accountable: Recipient | None = None # gets one status line + escalation: Recipient | None = None # hears only if unresolved delivery: Delivery = WATCH_DELIVERY - state: Action | None = None # what a person last did about it + state: Action | None = None # what a person last did about it history: list[Action] = field(default_factory=list) suppressed_reason: str | None = None facts: list[str] = field(default_factory=list) @@ -176,8 +178,14 @@ def _suppression(m: Match, today: date) -> str | None: return None -def decide(matches: list[Match], fallback_email: str, fallback_name: str = "Security team", - today: date | None = None, policy: Policy | None = None, state: StateStore | None = None) -> list[Decision]: +def decide( + matches: list[Match], + fallback_email: str, + fallback_name: str = "Security team", + today: date | None = None, + policy: Policy | None = None, + state: StateStore | None = None, +) -> list[Decision]: today = today or date.today() policy = policy or Policy.default() state = state or StateStore(None) @@ -189,8 +197,15 @@ def decide(matches: list[Match], fallback_email: str, fallback_name: str = "Secu else: email, name, fb = fallback_email, fallback_name, True - d = Decision(match=m, urgency=WATCH, urgency_reason="", recipient_email=email, recipient_name=name, - recipient_is_fallback=fb, suppressed_reason=_suppression(m, today)) + d = Decision( + match=m, + urgency=WATCH, + urgency_reason="", + recipient_email=email, + recipient_name=name, + recipient_is_fallback=fb, + suppressed_reason=_suppression(m, today), + ) if m.tier == "possible": d.urgency_reason = f"match is only possible: {m.reason}" else: @@ -225,7 +240,8 @@ def decide(matches: list[Match], fallback_email: str, fallback_name: str = "Secu d.facts.append(f"Federal remediation due date: {adv.due_date.isoformat()}.") d.estimates.append(f"Product match is {m.tier}: {m.reason}.") d.estimates.append( - f"Version {a.version} was not checked; the CISA feed does not list affected versions." if a.version + f"Version {a.version} was not checked; the CISA feed does not list affected versions." + if a.version else "No version recorded for this asset, so affected-version status is unknown." ) if fb: @@ -248,15 +264,24 @@ class Summary: unowned: int by_urgency: dict[str, int] by_recipient: dict[str, list[Decision]] - by_person: dict[str, dict] # email -> {name, fixer, status, escalation} + by_person: dict[str, dict] # email -> {name, fixer, status, escalation} status_lines: dict[str, list[Decision]] # accountable email -> notices - by_state: dict[str, int] # open | acknowledged | assigned | fixed + by_state: dict[str, int] # open | acknowledged | assigned | fixed over_budget: dict[str, int] warnings: list[str] -def summarize(decisions: list[Decision], *, days: int, catalog_version: str, policy_name: str, advisories_in_window: int, - assets: int, budget: int, warnings: list[str]) -> Summary: +def summarize( + decisions: list[Decision], + *, + days: int, + catalog_version: str, + policy_name: str, + advisories_in_window: int, + assets: int, + budget: int, + warnings: list[str], +) -> Summary: sent = [d for d in decisions if d.sent] by_rec: dict[str, list[Decision]] = defaultdict(list) for d in sent: @@ -282,9 +307,14 @@ def bump(r: Recipient, col: str) -> None: if d.escalation: bump(d.escalation, "escalation") return Summary( - days=days, catalog_version=catalog_version, policy_name=policy_name, advisories_in_window=advisories_in_window, - assets=assets, relevant_advisories=len({d.match.advisory.cve_id for d in decisions}), - sent=len(sent), suppressed=len(decisions) - len(sent), + days=days, + catalog_version=catalog_version, + policy_name=policy_name, + advisories_in_window=advisories_in_window, + assets=assets, + relevant_advisories=len({d.match.advisory.cve_id for d in decisions}), + sent=len(sent), + suppressed=len(decisions) - len(sent), unowned=sum(1 for d in sent if d.recipient_is_fallback), by_urgency={u: sum(1 for d in sent if d.urgency == u) for u in (ACT_NOW, UPDATE_SOON, PLAN_UPDATE, WATCH)}, by_recipient=dict(sorted(by_rec.items(), key=lambda kv: -len(kv[1]))), diff --git a/patchowner/engine.py b/patchowner/engine.py index 662290d..b033bd0 100644 --- a/patchowner/engine.py +++ b/patchowner/engine.py @@ -1,4 +1,5 @@ """One call that runs the whole replay: feed -> window -> match -> decide -> summarize -> html.""" + from __future__ import annotations from dataclasses import dataclass @@ -22,9 +23,19 @@ class Replay: html: str -def run_replay(csv_text: str, *, inventory_name: str, days: int = 90, fallback_email: str = "security@example.com", - fallback_name: str = "Security team", budget: int = 10, refresh: bool = False, - today: date | None = None, policy: Policy | None = None, state: StateStore | None = None) -> Replay: +def run_replay( + csv_text: str, + *, + inventory_name: str, + days: int = 90, + fallback_email: str = "security@example.com", + fallback_name: str = "Security team", + budget: int = 10, + refresh: bool = False, + today: date | None = None, + policy: Policy | None = None, + state: StateStore | None = None, +) -> Replay: assets = parse_inventory(csv_text) warnings = [f"Row {a.row} ({a.asset or 'blank'}): {w}" for a in assets for w in a.warnings] advisories, version = load_feed(refresh=refresh) @@ -32,7 +43,15 @@ def run_replay(csv_text: str, *, inventory_name: str, days: int = 90, fallback_e matches = match_all(window, assets) policy = policy or Policy.default() decisions = decide(matches, fallback_email, fallback_name, today, policy, state) - summary = summarize(decisions, days=days, catalog_version=version, policy_name=policy.name, advisories_in_window=len(window), - assets=len(assets), budget=budget, warnings=warnings) + summary = summarize( + decisions, + days=days, + catalog_version=version, + policy_name=policy.name, + advisories_in_window=len(window), + assets=len(assets), + budget=budget, + warnings=warnings, + ) health = assess_health(assets, decisions, summary, today=today) return Replay(summary, decisions, health, render_html(summary, decisions, policy, health, inventory_name=inventory_name)) diff --git a/patchowner/health.py b/patchowner/health.py index fb9660a..44c391a 100644 --- a/patchowner/health.py +++ b/patchowner/health.py @@ -3,6 +3,7 @@ Every check is a plain sentence, a count, and the one thing to do about it. Nothing here changes a decision; it tells the person running the replay where the inventory or the follow-through is thin. """ + from __future__ import annotations from dataclasses import dataclass, field @@ -13,31 +14,31 @@ GOOD, WARN, BAD = "good", "warn", "bad" GRADE_WORDS = {GOOD: "Ready", WARN: "Mostly ready", BAD: "Needs attention"} -FEED_STALE_AFTER = 7 # days: KEV is updated most working days -EXCEPTION_SOON = 30 # days: an exception ending this soon is worth a look now +FEED_STALE_AFTER = 7 # days: KEV is updated most working days +EXCEPTION_SOON = 30 # days: an exception ending this soon is worth a look now @dataclass class Check: - label: str # what good looks like: "Every active asset has an owner" - ok: int # how many pass - total: int # out of how many - status: str # good | warn | bad - detail: str # the failing ones, by name, or a reassuring sentence - fix: str = "" # the one thing to do; blank when nothing is needed + label: str # what good looks like: "Every active asset has an owner" + ok: int # how many pass + total: int # out of how many + status: str # good | warn | bad + detail: str # the failing ones, by name, or a reassuring sentence + fix: str = "" # the one thing to do; blank when nothing is needed names: list[str] = field(default_factory=list) @dataclass class Health: - grade: str # good | warn | bad - headline: str # one sentence for the top of the tab + grade: str # good | warn | bad + headline: str # one sentence for the top of the tab checks: list[Check] feed_date: date | None feed_age_days: int | None active_assets: int - handled: int # fixed or does-not-apply - unhandled_urgent: int # Act now or Update soon with nobody acting + handled: int # fixed or does-not-apply + unhandled_urgent: int # Act now or Update soon with nobody acting @property def word(self) -> str: @@ -81,62 +82,155 @@ def assess_health(assets: list[Asset], decisions: list[Decision], summary: Summa checks: list[Check] = [] # 1. Routing: is there a person for each role? Without these the notice goes to the fallback or nowhere. - checks.append(_coverage("Every active asset has an owner to fix it", active, [a for a in active if not a.has_owner], - "Add owner_email (and owner_name) to those rows. Until then the fallback contact gets their notices.")) - checks.append(_coverage("Every active asset has an accountable person", active, [a for a in active if not a.accountable], - "Add accountable. They get one status line per asset, never the technical detail.", bad_below=0.5)) + checks.append( + _coverage( + "Every active asset has an owner to fix it", + active, + [a for a in active if not a.has_owner], + "Add owner_email (and owner_name) to those rows. Until then the fallback contact gets their notices.", + ) + ) + checks.append( + _coverage( + "Every active asset has an accountable person", + active, + [a for a in active if not a.accountable], + "Add accountable. They get one status line per asset, never the technical detail.", + bad_below=0.5, + ) + ) exposed = [a for a in active if a.internet_exposed or a.exposure == "open"] - checks.append(_coverage("Every internet-facing asset has an on-call contact", exposed, [a for a in exposed if not a.oncall], - "Add oncall. On Act now, they join the owner; nobody else does.", bad_below=0.5)) - checks.append(_coverage("Every active asset has an escalation contact", active, [a for a in active if not a.escalate_to], - "Add escalate_to. They hear nothing unless the window passes.", bad_below=0.5)) + checks.append( + _coverage( + "Every internet-facing asset has an on-call contact", + exposed, + [a for a in exposed if not a.oncall], + "Add oncall. On Act now, they join the owner; nobody else does.", + bad_below=0.5, + ) + ) + checks.append( + _coverage( + "Every active asset has an escalation contact", + active, + [a for a in active if not a.escalate_to], + "Add escalate_to. They hear nothing unless the window passes.", + bad_below=0.5, + ) + ) # 2. Data quality: things the parser had to guess or ignore. warned = [a for a in assets if a.warnings] - checks.append(Check("Every row parsed cleanly", len(assets) - len(warned), len(assets), - GOOD if not warned else WARN, - "No warnings." if not warned else f"Warnings on {_names(warned)}. See the top of the Notices tab.", - "" if not warned else "Fix the flagged values; an ignored value falls back to a default that may be wrong.", - [a.asset for a in warned])) + checks.append( + Check( + "Every row parsed cleanly", + len(assets) - len(warned), + len(assets), + GOOD if not warned else WARN, + "No warnings." if not warned else f"Warnings on {_names(warned)}. See the top of the Notices tab.", + "" if not warned else "Fix the flagged values; an ignored value falls back to a default that may be wrong.", + [a.asset for a in warned], + ) + ) no_version = [a for a in active if not a.version] - checks.append(Check("Versions recorded", len(active) - len(no_version), len(active), - GOOD if not no_version else WARN, - "All of them." if not no_version else f"No version on {_names(no_version)}.", - "" if no_version == [] else "Add version. KEV carries no version data, so PatchOwner never claims one is affected; the owner still needs it to check.", - [a.asset for a in no_version])) + checks.append( + Check( + "Versions recorded", + len(active) - len(no_version), + len(active), + GOOD if not no_version else WARN, + "All of them." if not no_version else f"No version on {_names(no_version)}.", + "" + if no_version == [] + else "Add version. KEV carries no version data, so PatchOwner never claims one is affected; the owner still needs it to check.", + [a.asset for a in no_version], + ) + ) # 3. Exceptions: expired ones silently stop suppressing; ones ending soon deserve a look. expired = [a for a in active if a.exception_until and a.exception_until < today] ending = [a for a in active if a.exception_until and today <= a.exception_until <= today + timedelta(days=EXCEPTION_SOON)] if expired: - checks.append(Check("No exception has expired", 0, len(expired), BAD, - f"Expired on {_names(expired)}. Their notices are being sent again.", - "Extend exception_until with a fresh reason, or remove the row's exception and fix the asset.", - [a.asset for a in expired])) + checks.append( + Check( + "No exception has expired", + 0, + len(expired), + BAD, + f"Expired on {_names(expired)}. Their notices are being sent again.", + "Extend exception_until with a fresh reason, or remove the row's exception and fix the asset.", + [a.asset for a in expired], + ) + ) if ending: - checks.append(Check(f"No exception ends within {EXCEPTION_SOON} days", 0, len(ending), WARN, - f"Ending soon on {_names(ending)}.", - "Decide now whether to extend or to fix; the notice resumes the day after.", - [a.asset for a in ending])) + checks.append( + Check( + f"No exception ends within {EXCEPTION_SOON} days", + 0, + len(ending), + WARN, + f"Ending soon on {_names(ending)}.", + "Decide now whether to extend or to fix; the notice resumes the day after.", + [a.asset for a in ending], + ) + ) if not expired and not ending: on_file = [a for a in active if a.exception_until] - checks.append(Check("Exceptions are current", len(on_file), len(on_file), GOOD, - "None on file." if not on_file else f"{len(on_file)} on file, none expired or ending within {EXCEPTION_SOON} days.")) + checks.append( + Check( + "Exceptions are current", + len(on_file), + len(on_file), + GOOD, + "None on file." if not on_file else f"{len(on_file)} on file, none expired or ending within {EXCEPTION_SOON} days.", + ) + ) if retired: - checks.append(Check("Retired assets stay quiet", len(retired), len(retired), GOOD, - f"{_names(retired)}: retired, so nothing is sent and each suppression is listed with its reason.")) + checks.append( + Check( + "Retired assets stay quiet", + len(retired), + len(retired), + GOOD, + f"{_names(retired)}: retired, so nothing is sent and each suppression is listed with its reason.", + ) + ) # 4. The feed: KEV is updated most working days. feed_date = feed_date_from_version(summary.catalog_version) age = (today - feed_date).days if feed_date else None if age is None: - checks.append(Check("KEV catalog is fresh", 0, 1, WARN, f"Catalog version '{summary.catalog_version}' has no date in it.", - "Run with --refresh to download the feed again.")) + checks.append( + Check( + "KEV catalog is fresh", + 0, + 1, + WARN, + f"Catalog version '{summary.catalog_version}' has no date in it.", + "Run with --refresh to download the feed again.", + ) + ) elif age > FEED_STALE_AFTER: - checks.append(Check("KEV catalog is fresh", 0, 1, BAD, f"The cached catalog is {age} days old (version {summary.catalog_version}).", - "Run with --refresh. Anything CISA added since is missing from this replay.")) + checks.append( + Check( + "KEV catalog is fresh", + 0, + 1, + BAD, + f"The cached catalog is {age} days old (version {summary.catalog_version}).", + "Run with --refresh. Anything CISA added since is missing from this replay.", + ) + ) else: - checks.append(Check("KEV catalog is fresh", 1, 1, GOOD, f"Version {summary.catalog_version}, {age} day{'s' if age != 1 else ''} old on {today.isoformat()}.")) + checks.append( + Check( + "KEV catalog is fresh", + 1, + 1, + GOOD, + f"Version {summary.catalog_version}, {age} day{'s' if age != 1 else ''} old on {today.isoformat()}.", + ) + ) # 5. Follow-through: was anything sent, and is anyone acting on the urgent ones? sent = [d for d in decisions if d.sent] @@ -144,23 +238,44 @@ def assess_health(assets: list[Asset], decisions: list[Decision], summary: Summa urgent = [d for d in sent if d.urgency in {ACT_NOW, UPDATE_SOON}] idle = [d for d in urgent if d.state is None or d.state.action == "reopened"] if urgent: - checks.append(Check("Someone has acted on every urgent notice", len(urgent) - len(idle), len(urgent), - GOOD if not idle else (BAD if any(d.urgency == ACT_NOW for d in idle) else WARN), - "All of them." if not idle else "Nobody has touched " + ", ".join(f"{d.match.advisory.cve_id} on {d.match.asset.asset}" for d in idle[:4]) + (f" and {len(idle) - 4} more" if len(idle) > 4 else "") + ".", - "" if not idle else "Acknowledge is the first button on each card. Acknowledged is not fixed, but it is not silence.")) + checks.append( + Check( + "Someone has acted on every urgent notice", + len(urgent) - len(idle), + len(urgent), + GOOD if not idle else (BAD if any(d.urgency == ACT_NOW for d in idle) else WARN), + "All of them." + if not idle + else "Nobody has touched " + + ", ".join(f"{d.match.advisory.cve_id} on {d.match.asset.asset}" for d in idle[:4]) + + (f" and {len(idle) - 4} more" if len(idle) > 4 else "") + + ".", + "" if not idle else "Acknowledge is the first button on each card. Acknowledged is not fixed, but it is not silence.", + ) + ) else: checks.append(Check("Someone has acted on every urgent notice", 0, 0, GOOD, "Nothing urgent was sent in this window.")) if summary.over_budget: who = ", ".join(f"{e} ({n})" for e, n in summary.over_budget.items()) - checks.append(Check("Nobody is over their notice budget", 0, len(summary.over_budget), WARN, - f"Over budget: {who}.", "Route lower-confidence items to a weekly digest, or split ownership.")) + checks.append( + Check( + "Nobody is over their notice budget", + 0, + len(summary.over_budget), + WARN, + f"Over budget: {who}.", + "Route lower-confidence items to a weekly digest, or split ownership.", + ) + ) else: checks.append(Check("Nobody is over their notice budget", 1, 1, GOOD, "Nobody would have been flooded.")) grade = BAD if any(c.status == BAD for c in checks) else (WARN if any(c.status == WARN for c in checks) else GOOD) issues = sum(1 for c in checks if c.status != GOOD) if grade == GOOD: - headline = f"Every one of your {len(active)} active assets can be routed, the catalog is current, and nothing urgent is sitting untouched." + headline = ( + f"Every one of your {len(active)} active assets can be routed, the catalog is current, and nothing urgent is sitting untouched." + ) else: headline = f"{issues} thing{'s' if issues != 1 else ''} to fix, worst first. Each one says what to do." order = {BAD: 0, WARN: 1, GOOD: 2} diff --git a/patchowner/inventory.py b/patchowner/inventory.py index dd01022..cc0d8f5 100644 --- a/patchowner/inventory.py +++ b/patchowner/inventory.py @@ -1,4 +1,5 @@ """Inventory: the customer's list of technology they care about, from a CSV.""" + from __future__ import annotations import csv @@ -8,9 +9,21 @@ REQUIRED = ("asset", "vendor", "product") OPTIONAL = ( - "version", "environment", "internet_exposed", "criticality", - "owner_name", "owner_email", "team", "status", "exception_until", "exception_reason", - "exposure", "human_impact", "escalate_to", "accountable", "oncall", + "version", + "environment", + "internet_exposed", + "criticality", + "owner_name", + "owner_email", + "team", + "status", + "exception_until", + "exception_reason", + "exposure", + "human_impact", + "escalate_to", + "accountable", + "oncall", ) EXPOSURE_VALUES = {"small", "controlled", "open"} HUMAN_IMPACT_VALUES = {"low", "medium", "high", "very high"} @@ -32,11 +45,11 @@ class Asset: status: str = "active" exception_until: date | None = None exception_reason: str = "" - exposure: str = "" # SSVC System Exposure override: small | controlled | open - human_impact: str = "" # SSVC Human Impact override: low | medium | high | very high - escalate_to: str = "" # hears about it only if it is stuck - accountable: str = "" # gets a one-line status, never the technical detail - oncall: str = "" # joins the fixer on Act now + exposure: str = "" # SSVC System Exposure override: small | controlled | open + human_impact: str = "" # SSVC Human Impact override: low | medium | high | very high + escalate_to: str = "" # hears about it only if it is stuck + accountable: str = "" # gets a one-line status, never the technical detail + oncall: str = "" # joins the fixer on Act now row: int = 0 warnings: list[str] = field(default_factory=list) @@ -65,8 +78,7 @@ def parse_inventory(text: str) -> list[Asset]: missing = [c for c in REQUIRED if c not in headers] if missing: raise InventoryError( - f"Missing required column(s): {', '.join(missing)}. " - f"Required: {', '.join(REQUIRED)}. Optional: {', '.join(OPTIONAL)}." + f"Missing required column(s): {', '.join(missing)}. Required: {', '.join(REQUIRED)}. Optional: {', '.join(OPTIONAL)}." ) def get(rowd: dict, key: str) -> str: diff --git a/patchowner/matching.py b/patchowner/matching.py index 945b944..8fc62c6 100644 --- a/patchowner/matching.py +++ b/patchowner/matching.py @@ -3,6 +3,7 @@ Tiers: exact, likely, possible, none. KEV carries no version data, so a match never claims a specific version is affected; that is stated on every alert. """ + from __future__ import annotations import re @@ -79,8 +80,23 @@ def vendor_matches(inv_vendor: str, kev_vendor: str, kev_product: str) -> bool: GENERIC_PRODUCT_TOKENS = { - "server", "servers", "appliance", "appliances", "platform", "edition", "enterprise", "suite", - "os", "software", "product", "products", "service", "services", "system", "systems", "core", + "server", + "servers", + "appliance", + "appliances", + "platform", + "edition", + "enterprise", + "suite", + "os", + "software", + "product", + "products", + "service", + "services", + "system", + "systems", + "core", } diff --git a/patchowner/state.py b/patchowner/state.py index 02fdd53..2fe2a2a 100644 --- a/patchowner/state.py +++ b/patchowner/state.py @@ -2,18 +2,22 @@ Acknowledged is not remediated. Only 'fixed' and 'not_applicable' count as handled. """ + from __future__ import annotations import json from dataclasses import asdict, dataclass -from datetime import datetime, timezone +from datetime import UTC, datetime from pathlib import Path ACTIONS = ("acknowledged", "assigned", "not_applicable", "fixed", "reopened") HANDLED = {"fixed", "not_applicable"} WORDS = { - "acknowledged": "Acknowledged", "assigned": "Assigned", "not_applicable": "Does not apply", - "fixed": "Fixed", "reopened": "Reopened", + "acknowledged": "Acknowledged", + "assigned": "Assigned", + "not_applicable": "Does not apply", + "fixed": "Fixed", + "reopened": "Reopened", } @@ -52,7 +56,7 @@ def __init__(self, path: Path | str | None): def record(self, key: str, action: str, by: str, note: str = "", at: datetime | None = None) -> Action: if action not in ACTIONS: raise ValueError(f"'{action}' is not one of {', '.join(ACTIONS)}") - a = Action(action, by.strip() or "someone", note.strip(), (at or datetime.now(timezone.utc)).isoformat(timespec="seconds")) + a = Action(action, by.strip() or "someone", note.strip(), (at or datetime.now(UTC)).isoformat(timespec="seconds")) self.data.setdefault(key, []).append(a) self.save() return a diff --git a/pyproject.toml b/pyproject.toml index d1fce45..cba7de0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,6 +20,7 @@ patchowner = "patchowner.cli:main" dev = [ "httpx>=0.28.1", "pytest>=8", + "ruff>=0.6", ] [build-system] @@ -31,3 +32,15 @@ packages = ["patchowner"] [tool.pytest.ini_options] testpaths = ["tests"] + +[tool.ruff] +line-length = 140 +target-version = "py312" +extend-exclude = ["docs"] + +[tool.ruff.lint] +select = ["E", "F", "W", "I", "B", "UP"] +ignore = ["E501"] # long lines are a style choice here; the limit above is for ruff format + +[tool.ruff.lint.per-file-ignores] +"patchowner/web.py" = ["B008"] # File(...) / Form(...) in defaults is how FastAPI declares parameters diff --git a/tests/test_decide.py b/tests/test_decide.py index 475ec5a..a800d93 100644 --- a/tests/test_decide.py +++ b/tests/test_decide.py @@ -10,9 +10,17 @@ def adv(description="Bad thing over the network. More.", ransomware=False): - return Advisory(cve_id="CVE-2026-0001", vendor="Fortinet", product="FortiOS", name="n", description=description, - required_action="Apply update. Then verify.", date_added=date(2026, 9, 1), due_date=date(2026, 9, 22), - ransomware_known=ransomware) + return Advisory( + cve_id="CVE-2026-0001", + vendor="Fortinet", + product="FortiOS", + name="n", + description=description, + required_action="Apply update. Then verify.", + date_added=date(2026, 9, 1), + due_date=date(2026, 9, 22), + ransomware_known=ransomware, + ) def dec(tier="exact", policy=None, description="Bad thing over the network. More.", **kw): diff --git a/tests/test_engine.py b/tests/test_engine.py index 6564062..85360af 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -1,4 +1,5 @@ """End to end against the cached KEV feed. Skips if the feed has not been downloaded.""" + from datetime import date from pathlib import Path @@ -28,6 +29,7 @@ def test_example_inventory_replays_and_renders(): def test_card_ids_line_up_with_page_data(): import json import re + r = run_replay(EXAMPLE.read_text(), inventory_name="example", days=90, today=date(2026, 9, 10)) data = json.loads(re.search(r'', r.html, re.S).group(1)) for m in re.finditer(r'
' in scheduled.html # visible, not hidden + assert '

' in scheduled.html # visible, not hidden - deferred = run_replay(EXAMPLE.read_text(), inventory_name="example", days=90, today=date(2026, 9, 10), - policy=_policy_with_every_answer(tmp_path, "defer")) + deferred = run_replay( + EXAMPLE.read_text(), inventory_name="example", days=90, today=date(2026, 9, 10), policy=_policy_with_every_answer(tmp_path, "defer") + ) assert all(not d.sent for d in deferred.decisions if d.outcome == "defer") assert "Simulated recommendation: Defer, for testing the prioritization logic only" in deferred.html assert "was not sent because the policy" in deferred.html @@ -67,7 +77,7 @@ def test_slow_answers_carry_a_simulated_recommendation_caution(tmp_path): @pytest.mark.skipif(not DEFAULT_CACHE.exists(), reason="KEV feed not cached") def test_disclaimer_is_sitewide_and_on_about_tab(): r = run_replay(EXAMPLE.read_text(), inventory_name="example", days=90, today=date(2026, 9, 10)) - assert r.html.count("This site is for educational and testing purposes only") == 2 # About tab and footer - assert "Demo for testing the prioritization logic only" in r.html # banner + assert r.html.count("This site is for educational and testing purposes only") == 2 # About tab and footer + assert "Demo for testing the prioritization logic only" in r.html # banner assert 'id="tab-about"' in r.html and "Business Source License" in r.html - assert "Demo only, not security advice" in r.html # share text + assert "Demo only, not security advice" in r.html # share text diff --git a/tests/test_health.py b/tests/test_health.py index bc826ab..7373e74 100644 --- a/tests/test_health.py +++ b/tests/test_health.py @@ -1,7 +1,8 @@ """Health: routing coverage, exceptions, feed freshness, follow-through. No feed needed.""" + from datetime import date, timedelta -from patchowner.decide import ACT_NOW, Delivery, Decision, Recipient, summarize +from patchowner.decide import ACT_NOW, Decision, Delivery, Recipient, summarize from patchowner.health import BAD, GOOD, WARN, assess_health, feed_date_from_version from patchowner.inventory import parse_inventory from patchowner.kev import Advisory @@ -13,16 +14,42 @@ def _summary(decisions=(), catalog="2026.09.10", assets=1, warnings=(), budget=10): - return summarize(list(decisions), days=90, catalog_version=catalog, policy_name="p", advisories_in_window=10, - assets=assets, budget=budget, warnings=list(warnings)) + return summarize( + list(decisions), + days=90, + catalog_version=catalog, + policy_name="p", + advisories_in_window=10, + assets=assets, + budget=budget, + warnings=list(warnings), + ) def _decision(asset, urgency=ACT_NOW, state=None): - adv = Advisory(cve_id="CVE-2026-1", vendor="V", product="P", name="n", description="d", required_action="patch", - date_added=TODAY, due_date=None, ransomware_known=False) + adv = Advisory( + cve_id="CVE-2026-1", + vendor="V", + product="P", + name="n", + description="d", + required_action="patch", + date_added=TODAY, + due_date=None, + ransomware_known=False, + ) m = Match(advisory=adv, asset=asset, tier="exact", score=100, reason="") - return Decision(match=m, urgency=urgency, urgency_reason="", recipient_email="o@x", recipient_name="O", recipient_is_fallback=False, - fixers=[Recipient("o@x", "O", "fixer")], delivery=Delivery(True, "2 hours", "8 hours", "24 hours", False), state=state) + return Decision( + match=m, + urgency=urgency, + urgency_reason="", + recipient_email="o@x", + recipient_name="O", + recipient_is_fallback=False, + fixers=[Recipient("o@x", "O", "fixer")], + delivery=Delivery(True, "2 hours", "8 hours", "24 hours", False), + state=state, + ) def test_complete_inventory_is_ready(): @@ -68,7 +95,10 @@ def test_expired_and_ending_exceptions(): gone = (TODAY - timedelta(days=1)).isoformat() soon = (TODAY + timedelta(days=10)).isoformat() far = (TODAY + timedelta(days=200)).isoformat() - assets = parse_inventory(FULL + f"X,V,P,1.0,no,o@x,a@x,oc@x,e@x,active,{gone}\nY,V,P,1.0,no,o@x,a@x,oc@x,e@x,active,{soon}\nZ,V,P,1.0,no,o@x,a@x,oc@x,e@x,active,{far}\n") + assets = parse_inventory( + FULL + + f"X,V,P,1.0,no,o@x,a@x,oc@x,e@x,active,{gone}\nY,V,P,1.0,no,o@x,a@x,oc@x,e@x,active,{soon}\nZ,V,P,1.0,no,o@x,a@x,oc@x,e@x,active,{far}\n" + ) h = assess_health(assets, [], _summary(), today=TODAY) labels = {c.label: c for c in h.checks} assert labels["No exception has expired"].status == BAD and "X" in labels["No exception has expired"].detail diff --git a/tests/test_matching.py b/tests/test_matching.py index 46b0847..48dc0c1 100644 --- a/tests/test_matching.py +++ b/tests/test_matching.py @@ -8,8 +8,17 @@ def adv(vendor, product): - return Advisory(cve_id="CVE-2026-0001", vendor=vendor, product=product, name="", description="", required_action="", - date_added=date(2026, 9, 1), due_date=None, ransomware_known=False) + return Advisory( + cve_id="CVE-2026-0001", + vendor=vendor, + product=product, + name="", + description="", + required_action="", + date_added=date(2026, 9, 1), + due_date=None, + ransomware_known=False, + ) def asset(vendor, product, **kw): @@ -26,34 +35,46 @@ def test_normalize_drops_noise_and_parentheticals(): assert normalize("Cisco Systems, Inc.") == "cisco" -@pytest.mark.parametrize("kv,kp,iv,ip,expected", [ - ("Microsoft", "SharePoint", "Microsoft", "SharePoint Server", "exact"), - ("Microsoft", "SharePoint Server", "Microsoft", "SharePoint", "exact"), - ("Fortinet", "FortiOS", "Fortinet", "FortiOS", "exact"), - ("Fortinet", "FortiOS", "Fortinet", "Fortinet FortiOS", "exact"), - ("SonicWall", "SMA1000 Appliances", "SonicWall", "SMA 1000", "exact"), - ("Broadcom", "VMware vCenter", "VMware", "vCenter Server", "exact"), - ("Synacor", "Zimbra Collaboration Suite (ZCS)", "Zimbra", "Zimbra Collaboration Suite", "exact"), - ("Fortinet", "Multiple Products", "Fortinet", "FortiOS", "possible"), - ("Citrix", "NetScaler ADC and NetScaler Gateway", "Citrix", "NetScaler Gateway", "exact"), - ("Microsoft", "Windows", "Microsoft", "Windows Server", "exact"), - ("Cisco", "Secure Firewall Adaptive Security Appliance (ASA) and Secure Firewall Threat Defense (FTD) ", "Cisco", "ASA 5506", "likely"), - ("Microsoft", "Windows Ancillary Function Driver for WinSock", "Microsoft", "Windows Server", "exact"), - ("PaperCut", "NG/MF", "PaperCut", "PaperCut MF", "exact"), -]) +@pytest.mark.parametrize( + "kv,kp,iv,ip,expected", + [ + ("Microsoft", "SharePoint", "Microsoft", "SharePoint Server", "exact"), + ("Microsoft", "SharePoint Server", "Microsoft", "SharePoint", "exact"), + ("Fortinet", "FortiOS", "Fortinet", "FortiOS", "exact"), + ("Fortinet", "FortiOS", "Fortinet", "Fortinet FortiOS", "exact"), + ("SonicWall", "SMA1000 Appliances", "SonicWall", "SMA 1000", "exact"), + ("Broadcom", "VMware vCenter", "VMware", "vCenter Server", "exact"), + ("Synacor", "Zimbra Collaboration Suite (ZCS)", "Zimbra", "Zimbra Collaboration Suite", "exact"), + ("Fortinet", "Multiple Products", "Fortinet", "FortiOS", "possible"), + ("Citrix", "NetScaler ADC and NetScaler Gateway", "Citrix", "NetScaler Gateway", "exact"), + ("Microsoft", "Windows", "Microsoft", "Windows Server", "exact"), + ( + "Cisco", + "Secure Firewall Adaptive Security Appliance (ASA) and Secure Firewall Threat Defense (FTD) ", + "Cisco", + "ASA 5506", + "likely", + ), + ("Microsoft", "Windows Ancillary Function Driver for WinSock", "Microsoft", "Windows Server", "exact"), + ("PaperCut", "NG/MF", "PaperCut", "PaperCut MF", "exact"), + ], +) def test_positive_matches(kv, kp, iv, ip, expected): assert tier(kv, kp, iv, ip) == expected -@pytest.mark.parametrize("kv,kp,iv,ip", [ - ("Cisco", "IOS", "Fortinet", "FortiOS"), # vendor gate stops the substring coincidence - ("Microsoft", "Windows", "Intuit", "QuickBooks Desktop"), - ("Google", "Chromium V8", "Fortinet", "FortiOS"), - ("Apple", "macOS", "Microsoft", "Windows Server"), - ("Microsoft", "SQL Server", "Microsoft", "SharePoint Server"), # "server" alone is not a match - ("Microsoft", "SQL Server", "Microsoft", "Windows Server"), - ("Microsoft", "Active Directory Federation Services", "Microsoft", "SharePoint Server"), -]) +@pytest.mark.parametrize( + "kv,kp,iv,ip", + [ + ("Cisco", "IOS", "Fortinet", "FortiOS"), # vendor gate stops the substring coincidence + ("Microsoft", "Windows", "Intuit", "QuickBooks Desktop"), + ("Google", "Chromium V8", "Fortinet", "FortiOS"), + ("Apple", "macOS", "Microsoft", "Windows Server"), + ("Microsoft", "SQL Server", "Microsoft", "SharePoint Server"), # "server" alone is not a match + ("Microsoft", "SQL Server", "Microsoft", "Windows Server"), + ("Microsoft", "Active Directory Federation Services", "Microsoft", "SharePoint Server"), + ], +) def test_no_match(kv, kp, iv, ip): assert tier(kv, kp, iv, ip) is None diff --git a/tests/test_routing.py b/tests/test_routing.py index 88d1c94..7fb648a 100644 --- a/tests/test_routing.py +++ b/tests/test_routing.py @@ -3,6 +3,7 @@ Principle: send the detailed alert to the person who can fix it, a concise status to the person accountable for it, and an escalation only to the person who can remove a blocker. """ + from datetime import date from patchowner.decide import ACT_NOW, UPDATE_SOON, decide @@ -14,8 +15,17 @@ def adv(vendor, product, description="Remote code execution over the network."): - return Advisory(cve_id="CVE-2026-1", vendor=vendor, product=product, name="", description=description, - required_action="Apply update.", date_added=date(2026, 9, 1), due_date=None, ransomware_known=False) + return Advisory( + cve_id="CVE-2026-1", + vendor=vendor, + product=product, + name="", + description=description, + required_action="Apply update.", + date_added=date(2026, 9, 1), + due_date=None, + ransomware_known=False, + ) def route(asset, advisory): @@ -27,39 +37,80 @@ def emails(recipients): def test_actively_exploited_vpn(): - d = route(Asset(asset="VPN", vendor="Fortinet", product="FortiOS", internet_exposed=True, criticality="high", - owner_email="netops@x", owner_name="Network Operations", oncall="soc@x", - accountable="infra-owner@x", escalate_to="ciso@x"), adv("Fortinet", "FortiOS")) + d = route( + Asset( + asset="VPN", + vendor="Fortinet", + product="FortiOS", + internet_exposed=True, + criticality="high", + owner_email="netops@x", + owner_name="Network Operations", + oncall="soc@x", + accountable="infra-owner@x", + escalate_to="ciso@x", + ), + adv("Fortinet", "FortiOS"), + ) assert d.urgency == ACT_NOW - assert emails(d.fixers) == ["netops@x", "soc@x"] # detail: network operations + security on-call - assert d.accountable.email == "infra-owner@x" # status: infrastructure owner + assert emails(d.fixers) == ["netops@x", "soc@x"] # detail: network operations + security on-call + assert d.accountable.email == "infra-owner@x" # status: infrastructure owner assert d.escalation.email == "ciso@x" and d.delivery.escalate_after == "24 hours" # CISO only if unresolved def test_vulnerable_npm_dependency_in_production_service(): - d = route(Asset(asset="Checkout service", vendor="npm", product="lodash", criticality="high", - owner_email="repo-owner@x", accountable="eng-manager@x", escalate_to="champion@x"), - adv("npm", "lodash")) - assert d.urgency == UPDATE_SOON # internal, so not immediate under the default policy - assert emails(d.fixers) == ["repo-owner@x"] # on-call is not woken for out-of-cycle + d = route( + Asset( + asset="Checkout service", + vendor="npm", + product="lodash", + criticality="high", + owner_email="repo-owner@x", + accountable="eng-manager@x", + escalate_to="champion@x", + ), + adv("npm", "lodash"), + ) + assert d.urgency == UPDATE_SOON # internal, so not immediate under the default policy + assert emails(d.fixers) == ["repo-owner@x"] # on-call is not woken for out-of-cycle assert d.accountable.email == "eng-manager@x" assert d.escalation.email == "champion@x" and d.delivery.escalate_after == "7 days" def test_browser_patch_for_managed_laptops(): - d = route(Asset(asset="Managed laptops", vendor="Google", product="Chrome", criticality="medium", - owner_email="endpoint@x", accountable="itops-lead@x", escalate_to="helpdesk@x"), - adv("Google", "Chromium V8", "requires user interaction to visit a crafted page")) + d = route( + Asset( + asset="Managed laptops", + vendor="Google", + product="Chrome", + criticality="medium", + owner_email="endpoint@x", + accountable="itops-lead@x", + escalate_to="helpdesk@x", + ), + adv("Google", "Chromium V8", "requires user interaction to visit a crafted page"), + ) assert emails(d.fixers) == ["endpoint@x"] assert d.accountable.email == "itops-lead@x" - assert d.escalation.email == "helpdesk@x" # help desk only if it becomes stuck - assert d.questions # the text hint asks a person, never downgrades + assert d.escalation.email == "helpdesk@x" # help desk only if it becomes stuck + assert d.questions # the text hint asks a person, never downgrades def test_vulnerable_payment_system(): - d = route(Asset(asset="Payments", vendor="Adobe", product="Magento", internet_exposed=True, human_impact="very high", - owner_email="app-owner@x", oncall="infra-sec@x", accountable="finance-owner@x", escalate_to="cfo@x"), - adv("Adobe", "Commerce and Magento")) + d = route( + Asset( + asset="Payments", + vendor="Adobe", + product="Magento", + internet_exposed=True, + human_impact="very high", + owner_email="app-owner@x", + oncall="infra-sec@x", + accountable="finance-owner@x", + escalate_to="cfo@x", + ), + adv("Adobe", "Commerce and Magento"), + ) assert d.urgency == ACT_NOW assert emails(d.fixers) == ["app-owner@x", "infra-sec@x"] assert d.accountable.email == "finance-owner@x" @@ -67,17 +118,39 @@ def test_vulnerable_payment_system(): def test_cloud_control_plane(): - d = route(Asset(asset="Kubernetes control plane", vendor="Kubernetes", product="kube-apiserver", exposure="controlled", - criticality="critical", owner_email="cloud-platform@x", oncall="security@x", - accountable="cloud-owner@x", escalate_to="cto@x"), adv("Kubernetes", "kube-apiserver")) + d = route( + Asset( + asset="Kubernetes control plane", + vendor="Kubernetes", + product="kube-apiserver", + exposure="controlled", + criticality="critical", + owner_email="cloud-platform@x", + oncall="security@x", + accountable="cloud-owner@x", + escalate_to="cto@x", + ), + adv("Kubernetes", "kube-apiserver"), + ) assert d.assessment.values == ("active", "controlled", "yes", "very high") assert d.accountable.email == "cloud-owner@x" and d.escalation.email == "cto@x" assert "cloud-platform@x" in emails(d.fixers) def test_accountable_status_line_has_no_technical_detail(): - d = route(Asset(asset="VPN", vendor="Fortinet", product="FortiOS", internet_exposed=True, criticality="high", - owner_email="netops@x", owner_name="Dana", accountable="marco@x"), adv("Fortinet", "FortiOS")) + d = route( + Asset( + asset="VPN", + vendor="Fortinet", + product="FortiOS", + internet_exposed=True, + criticality="high", + owner_email="netops@x", + owner_name="Dana", + accountable="marco@x", + ), + adv("Fortinet", "FortiOS"), + ) line = d.status_line assert line.startswith("VPN: act now. Assigned to Dana. Acknowledge within 2 hours, plan within 8 hours.") assert "CVE" not in line and "Remote code" not in line @@ -91,10 +164,21 @@ def test_no_owner_means_fallback_fixes_and_nobody_is_accountable_or_escalated(): def test_accountable_gets_one_status_line_per_asset_not_per_advisory(): from patchowner.decide import summarize - a = Asset(asset="VPN", vendor="Fortinet", product="FortiOS", internet_exposed=True, criticality="high", - owner_email="netops@x", accountable="marco@x") - ds = decide([Match(adv("Fortinet", "FortiOS"), a, "exact", 100, "r"), Match(adv("Fortinet", "FortiOS"), a, "exact", 100, "r")], - fallback_email="security@x", today=TODAY) + + a = Asset( + asset="VPN", + vendor="Fortinet", + product="FortiOS", + internet_exposed=True, + criticality="high", + owner_email="netops@x", + accountable="marco@x", + ) + ds = decide( + [Match(adv("Fortinet", "FortiOS"), a, "exact", 100, "r"), Match(adv("Fortinet", "FortiOS"), a, "exact", 100, "r")], + fallback_email="security@x", + today=TODAY, + ) s = summarize(ds, days=90, catalog_version="v", policy_name="p", advisories_in_window=2, assets=1, budget=10, warnings=[]) assert len(s.status_lines["marco@x"]) == 1 and s.by_person["marco@x"]["status"] == 1 assert s.by_person["netops@x"]["fixer"] == 2 diff --git a/tests/test_ssvc.py b/tests/test_ssvc.py index ebe5169..fc01ffc 100644 --- a/tests/test_ssvc.py +++ b/tests/test_ssvc.py @@ -8,8 +8,17 @@ def adv(description="Remote code execution over the network."): - return Advisory(cve_id="CVE-2026-0001", vendor="V", product="P", name="", description=description, - required_action="Apply update.", date_added=date(2026, 9, 1), due_date=None, ransomware_known=False) + return Advisory( + cve_id="CVE-2026-0001", + vendor="V", + product="P", + name="", + description=description, + required_action="Apply update.", + date_added=date(2026, 9, 1), + due_date=None, + ransomware_known=False, + ) def test_default_policy_is_complete(): @@ -20,13 +29,16 @@ def test_default_policy_is_complete(): assert len(p.rows) == expected == 72 -@pytest.mark.parametrize("values,outcome,row", [ - (("none", "small", "no", "low"), "defer", 0), - (("active", "small", "no", "low"), "scheduled", 48), - (("active", "open", "yes", "high"), "immediate", 70), - (("active", "open", "no", "very high"), "immediate", 67), - (("active", "controlled", "yes", "medium"), "out-of-cycle", 61), -]) +@pytest.mark.parametrize( + "values,outcome,row", + [ + (("none", "small", "no", "low"), "defer", 0), + (("active", "small", "no", "low"), "scheduled", 48), + (("active", "open", "yes", "high"), "immediate", 70), + (("active", "open", "no", "very high"), "immediate", 67), + (("active", "controlled", "yes", "medium"), "out-of-cycle", 61), + ], +) def test_default_policy_rows(values, outcome, row): assert Policy.default().outcome_for(values) == (outcome, row) @@ -75,12 +87,15 @@ def test_a_person_can_confirm_it_needs_help(): assert b.values[2] == "no" and b.steps[2].fact and b.questions == [] -@pytest.mark.parametrize("text,expected", [ - ("allows an authenticated attacker to execute code", "authenticated attacker"), - ("requires user interaction to open a crafted file", "user interaction"), - ("allows a remote attacker to execute arbitrary code via crafted packets", ""), - ("could allow a remote unauthenticated attacker to cause requests", ""), -]) +@pytest.mark.parametrize( + "text,expected", + [ + ("allows an authenticated attacker to execute code", "authenticated attacker"), + ("requires user interaction to open a crafted file", "user interaction"), + ("allows a remote attacker to execute arbitrary code via crafted packets", ""), + ("could allow a remote unauthenticated attacker to cause requests", ""), + ], +) def test_automatable_hint(text, expected): assert automatable_hint(text) == expected diff --git a/tests/test_state.py b/tests/test_state.py index a6c53d9..381dcc7 100644 --- a/tests/test_state.py +++ b/tests/test_state.py @@ -1,4 +1,4 @@ -from datetime import date, datetime, timezone +from datetime import UTC, date, datetime import pytest @@ -12,20 +12,38 @@ def adv(cve="CVE-2026-1"): - return Advisory(cve_id=cve, vendor="Fortinet", product="FortiOS", name="", description="Remote code execution.", - required_action="Apply update.", date_added=date(2026, 9, 1), due_date=None, ransomware_known=False) + return Advisory( + cve_id=cve, + vendor="Fortinet", + product="FortiOS", + name="", + description="Remote code execution.", + required_action="Apply update.", + date_added=date(2026, 9, 1), + due_date=None, + ransomware_known=False, + ) def asset(**kw): - return Asset(asset="VPN", vendor="Fortinet", product="FortiOS", internet_exposed=True, criticality="high", - owner_email="dana@x", owner_name="Dana", accountable="marco@x", **kw) + return Asset( + asset="VPN", + vendor="Fortinet", + product="FortiOS", + internet_exposed=True, + criticality="high", + owner_email="dana@x", + owner_name="Dana", + accountable="marco@x", + **kw, + ) def test_store_persists_and_reports_current_state(tmp_path): path = tmp_path / "state.json" s = StateStore(path) key = notice_key("CVE-2026-1", "VPN") - s.record(key, "acknowledged", "Dana", at=datetime(2026, 9, 10, 9, 0, tzinfo=timezone.utc)) + s.record(key, "acknowledged", "Dana", at=datetime(2026, 9, 10, 9, 0, tzinfo=UTC)) s.record(key, "assigned", "Dana", "Priya") again = StateStore(path) assert again.current(key).action == "assigned" and again.current(key).note == "Priya" @@ -34,8 +52,10 @@ def test_store_persists_and_reports_current_state(tmp_path): def test_reopen_clears_current_state_but_keeps_history(tmp_path): - s = StateStore(tmp_path / "s.json"); key = "k" - s.record(key, "fixed", "Dana"); s.record(key, "reopened", "Marco", "vendor patch was pulled") + s = StateStore(tmp_path / "s.json") + key = "k" + s.record(key, "fixed", "Dana") + s.record(key, "reopened", "Marco", "vendor patch was pulled") assert s.current(key) is None and len(s.history(key)) == 2 @@ -53,7 +73,8 @@ def test_not_applicable_suppresses_next_replay_with_the_persons_reason(tmp_path) def test_status_line_reflects_state(tmp_path): - s = StateStore(tmp_path / "s.json"); key = notice_key("CVE-2026-1", "VPN") + s = StateStore(tmp_path / "s.json") + key = notice_key("CVE-2026-1", "VPN") fresh = decide([Match(adv(), asset(), "exact", 100, "r")], fallback_email="sec@x", today=TODAY, state=s)[0] assert "Assigned to Dana. Acknowledge within 2 hours" in fresh.status_line s.record(key, "acknowledged", "Dana") @@ -68,8 +89,12 @@ def test_status_line_reflects_state(tmp_path): def test_summary_counts_states(tmp_path): s = StateStore(tmp_path / "s.json") s.record(notice_key("CVE-2026-2", "VPN"), "acknowledged", "Dana") - ds = decide([Match(adv("CVE-2026-1"), asset(), "exact", 100, "r"), Match(adv("CVE-2026-2"), asset(), "exact", 100, "r")], - fallback_email="sec@x", today=TODAY, state=s) + ds = decide( + [Match(adv("CVE-2026-1"), asset(), "exact", 100, "r"), Match(adv("CVE-2026-2"), asset(), "exact", 100, "r")], + fallback_email="sec@x", + today=TODAY, + state=s, + ) sm = summarize(ds, days=90, catalog_version="v", policy_name="p", advisories_in_window=2, assets=1, budget=10, warnings=[]) assert sm.by_state == {"open": 1, "acknowledged": 1, "assigned": 0, "fixed": 0} @@ -78,6 +103,7 @@ def test_web_act_endpoint_records_and_rejects(tmp_path): from fastapi.testclient import TestClient from patchowner import web + web.configure(tmp_path / "state.json") c = TestClient(web.app) r = c.post("/act", json={"key": "CVE-2026-1|VPN", "action": "acknowledged", "by": "Dana"}) diff --git a/uv.lock b/uv.lock index d5ca2da..ee6a03f 100644 --- a/uv.lock +++ b/uv.lock @@ -231,6 +231,7 @@ dependencies = [ dev = [ { name = "httpx" }, { name = "pytest" }, + { name = "ruff" }, ] [package.metadata] @@ -246,6 +247,7 @@ requires-dist = [ dev = [ { name = "httpx", specifier = ">=0.28.1" }, { name = "pytest", specifier = ">=8" }, + { name = "ruff", specifier = ">=0.6" }, ] [[package]] @@ -471,6 +473,31 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/fb/04/a0b0e6324b6384d1ab40feb4d16400af3b3101d38cbd15957edd9d17cbe0/rapidfuzz-3.14.6-cp315-cp315t-win_arm64.whl", hash = "sha256:07c7aa0b1e4b9999a54f9e73317d6743ff85442c8ef7b7fbbe6b190fd37d9e75", size = 1243815, upload-time = "2026-08-30T21:45:31.187Z" }, ] +[[package]] +name = "ruff" +version = "0.16.10" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/c4/49/23802c45f093eb14bde54b141d2b2f058edfa63a06db7beed047308cc08f/ruff-0.16.10.tar.gz", hash = "sha256:eff4728c4eaae93f0955cd264d24b2ab348e74bf59986ccf282ba6dc16b3b017", size = 4958724, upload-time = "2026-10-01T18:03:21.697Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/2f/21/ebce22e1d90cdb2cd691026b9c6e9bec6499f0b396089481755b5d49efff/ruff-0.16.10-py3-none-linux_armv6l.whl", hash = "sha256:488b0fe3f3574210e5cf80d9f59b9e3ab17a127a8155de3f307b392589cfb511", size = 10095557, upload-time = "2026-10-01T18:02:36.072Z" }, + { url = "https://files.pythonhosted.org/packages/cb/98/a54de85876a8b2612bfa0d84c7b9abfb39c6a3354aee7800b09c1649b9e3/ruff-0.16.10-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:e748ff95c934c4e978783b8e687bc174e7bd84e8ad24e3243e1ecfcda5e0282d", size = 10340577, upload-time = "2026-10-01T18:02:39.258Z" }, + { url = "https://files.pythonhosted.org/packages/9f/16/1a5a4a2657effe4806110f2b907313802f1367fcdbb29e8122766e407fab/ruff-0.16.10-py3-none-macosx_11_0_arm64.whl", hash = "sha256:3031a4a2e8e7b8a46f70be45f198c35a11ece509a94b80334d8d397a33c67550", size = 9774282, upload-time = "2026-10-01T18:02:41.79Z" }, + { url = "https://files.pythonhosted.org/packages/6e/fb/470085af734da396e68cd80fb0a3e7c109a459716ae59588c0f6fab8a17d/ruff-0.16.10-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:494401c86df4c4c25f69b9419605d944467ee98c42fb6ad405ef4fa40b8fb67d", size = 9920895, upload-time = "2026-10-01T18:02:44.458Z" }, + { url = "https://files.pythonhosted.org/packages/57/de/f10cffe4f88a37ea6615ec460f01bf76f0bc1477737a35d9ee61a630a78b/ruff-0.16.10-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:d203abc0ff2b773ee33d00ab8df0bb67046fbc7c07b119332f08b9b345cf8221", size = 9902707, upload-time = "2026-10-01T18:02:47.041Z" }, + { url = "https://files.pythonhosted.org/packages/ff/44/3fdcedf83ae60ef837dd239170e480dce606a9142cfa2db911afdf855fd9/ruff-0.16.10-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:bd83d1235a5258d318477bc5b576303974cbdef5df0c01a1bff14efcc23bd12a", size = 10619287, upload-time = "2026-10-01T18:02:49.485Z" }, + { url = "https://files.pythonhosted.org/packages/c1/62/02e76a5574002153618eafbb70e159468addc72a5d2c488e65cbb4639d2d/ruff-0.16.10-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:bc2610fb269fa56dd8a68669ae470fa6272902668c0fc2ebc3aa112b2633d5b8", size = 11339942, upload-time = "2026-10-01T18:02:52.008Z" }, + { url = "https://files.pythonhosted.org/packages/6b/c4/cde27d47ad8d4126c587e608c67c46e7feac11763f606d261a1bc8a489e6/ruff-0.16.10-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:3e70175e29cc94c26ea296c80e470180b744b7419026898e58f520c6ab32578e", size = 10934316, upload-time = "2026-10-01T18:02:54.811Z" }, + { url = "https://files.pythonhosted.org/packages/e4/03/17234145f302a645a123e8c3bb2411ecf4669fbc350de3b0d803230a1729/ruff-0.16.10-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:f33f43a864a8483eebd160e713336c8bab02c934feaff0a33cf5ccb41546d09a", size = 10387968, upload-time = "2026-10-01T18:02:57.497Z" }, + { url = "https://files.pythonhosted.org/packages/71/29/2493af60240ee7644b38a4b821f5f9c3b5a4fa3178770fe1d0c685217395/ruff-0.16.10-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:1dfc6f0088149fb6a362c1c446bcbb3fd2157b3852fe2fa68409276eab9ad9b3", size = 10537906, upload-time = "2026-10-01T18:03:00.006Z" }, + { url = "https://files.pythonhosted.org/packages/2e/9a/f56b28f3b143bb9e518c34fb89ab191b626bca8b70dbf8af0d2e2d473572/ruff-0.16.10-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:6553498afc35f580f036030795810b9e6bcea31604b0fd9e8d352795473042e3", size = 10012938, upload-time = "2026-10-01T18:03:03.006Z" }, + { url = "https://files.pythonhosted.org/packages/c2/c4/fca37362848ea4d4d80712df13632e645e7c7cfb1bdedf140699ac7a090b/ruff-0.16.10-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:a3b8471dea115d37f123882be852bed13403746d3a76c11de4a19ec5f9ff5a03", size = 9897945, upload-time = "2026-10-01T18:03:05.818Z" }, + { url = "https://files.pythonhosted.org/packages/7d/c7/e0bc57664d6e0c61fd22f620af260fe9662d7e1ddb320ea7f160e76ef165/ruff-0.16.10-py3-none-musllinux_1_2_i686.whl", hash = "sha256:92e59a70bcbd9d3a5483656da906ec28edfdacfce00afd99edb8b4e9d15644be", size = 10333134, upload-time = "2026-10-01T18:03:08.25Z" }, + { url = "https://files.pythonhosted.org/packages/21/aa/5c9f3b68737c0e4a33a1db5dfab44f7b786d96ca91a7233112ddab1e6dc2/ruff-0.16.10-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:7ae7375f803b5520dc9f546bed7e3a0acb70b91812e9bb4b19927de22f25b77d", size = 10741702, upload-time = "2026-10-01T18:03:10.775Z" }, + { url = "https://files.pythonhosted.org/packages/78/fa/0f9c2020dc316c53d983be011720b8157cc052b97e3f4dfa7db6d0f880a6/ruff-0.16.10-py3-none-win32.whl", hash = "sha256:2a12e01cb9156c10c466f63b46eaae5ecea28dfbd21b5836353ae498e7d1349a", size = 10139176, upload-time = "2026-10-01T18:03:13.267Z" }, + { url = "https://files.pythonhosted.org/packages/99/29/cfb0df9448d4d4ad48c2de029ada9ebd71baa6da983a6c77ee6c6cd0fe82/ruff-0.16.10-py3-none-win_amd64.whl", hash = "sha256:97f2015c92aa97105b0eab19eb5d224884399281cfc5da86a92db4ab5e7fb2ca", size = 10584734, upload-time = "2026-10-01T18:03:16.006Z" }, + { url = "https://files.pythonhosted.org/packages/fc/05/c16957eb287c3fc062e032619a25868d93d408a844b725bcf514f7a378ff/ruff-0.16.10-py3-none-win_arm64.whl", hash = "sha256:25a65fe998c4e6861ec079ada5826a2fc605e6cbccbe9dcd7fac1f54e791621b", size = 10366440, upload-time = "2026-10-01T18:03:19.04Z" }, +] + [[package]] name = "starlette" version = "1.6.0" From 23eb9450951e772c7bb0beee6019b3bc8647a144 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 13:51:20 +0000 Subject: [PATCH 2/4] Fail plainly when the KEV feed is unavailable; bound web uploads; escape report data - kev.load_feed raises FeedError with a sentence for the person running the tool instead of a raw URLError or JSONDecodeError. A failed or non-JSON download never overwrites an existing cache. The CLI prints it and exits 3; the web form shows it in place of a 500. - /replay refuses uploads over 5 MB before parsing, checks the look-back range, and shows every error as escaped text. /act bounds field sizes. - Policy.parse reports short rows and non-numeric row numbers as PolicyError rather than IndexError/ValueError. - The client-data JSON embedded in the report's " cannot end the tag. Tests cover each of these. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_012ZCSLekTvZQ7X3PsAUJuXk --- patchowner/cli.py | 25 +++++++-- patchowner/kev.py | 28 +++++++++-- patchowner/report.py | 117 +++++++++++++++++++++++++++++++------------ patchowner/ssvc.py | 103 ++++++++++++++++++++++++++----------- patchowner/web.py | 42 ++++++++++++---- tests/test_kev.py | 64 +++++++++++++++++++++++ tests/test_report.py | 56 +++++++++++++++++++++ tests/test_web.py | 63 +++++++++++++++++++++++ 8 files changed, 419 insertions(+), 79 deletions(-) create mode 100644 tests/test_kev.py create mode 100644 tests/test_report.py create mode 100644 tests/test_web.py diff --git a/patchowner/cli.py b/patchowner/cli.py index fb45654..3fcb951 100644 --- a/patchowner/cli.py +++ b/patchowner/cli.py @@ -1,4 +1,5 @@ """Command line: `patchowner replay inventory.csv` and `patchowner serve`.""" + from __future__ import annotations import argparse @@ -7,6 +8,7 @@ from .engine import run_replay from .inventory import InventoryError +from .kev import FeedError from .report import BANNER from .ssvc import Policy, PolicyError from .state import StateStore @@ -16,23 +18,38 @@ def _replay(args: argparse.Namespace) -> int: path = Path(args.inventory) try: policy = Policy.load(args.policy) if args.policy else None - r = run_replay(path.read_text(), inventory_name=path.name, days=args.days, fallback_email=args.fallback, - budget=args.budget, refresh=args.refresh, policy=policy, state=StateStore(args.state)) + r = run_replay( + path.read_text(), + inventory_name=path.name, + days=args.days, + fallback_email=args.fallback, + budget=args.budget, + refresh=args.refresh, + policy=policy, + state=StateStore(args.state), + ) except InventoryError as e: print(f"Inventory problem: {e}", file=sys.stderr) return 2 except PolicyError as e: print(f"Policy problem: {e}", file=sys.stderr) return 2 + except FeedError as e: + print(f"Catalog problem: {e}", file=sys.stderr) + return 3 s = r.summary print(f"KEV catalog {s.catalog_version}: {s.advisories_in_window} advisories in the last {s.days} days; policy: {s.policy_name}") - print(f"{s.relevant_advisories} touched your {s.assets} assets; {s.sent} notices, {s.suppressed} suppressed, {s.unowned} without an owner") + print( + f"{s.relevant_advisories} touched your {s.assets} assets; {s.sent} notices, {s.suppressed} suppressed, {s.unowned} without an owner" + ) for u, n in s.by_urgency.items(): print(f" {u:<12}{n}") print("state: " + ", ".join(f"{k} {v}" for k, v in s.by_state.items())) for d in r.decisions: if d.sent: - print(f"- [{d.urgency}] {d.match.advisory.cve_id} -> {d.match.asset.asset} -> {d.recipient_email} ({d.match.tier}; {d.vector or 'no path'})") + print( + f"- [{d.urgency}] {d.match.advisory.cve_id} -> {d.match.asset.asset} -> {d.recipient_email} ({d.match.tier}; {d.vector or 'no path'})" + ) for w in s.warnings: print(f"! {w}") for e, n in s.over_budget.items(): diff --git a/patchowner/kev.py b/patchowner/kev.py index 4af0565..d00dbd6 100644 --- a/patchowner/kev.py +++ b/patchowner/kev.py @@ -1,7 +1,9 @@ """CISA Known Exploited Vulnerabilities feed: download, cache, and filter.""" + from __future__ import annotations import json +import urllib.error import urllib.request from dataclasses import dataclass from datetime import date, timedelta @@ -11,6 +13,10 @@ DEFAULT_CACHE = Path(__file__).resolve().parent.parent / "data" / "kev.json" +class FeedError(RuntimeError): + """The KEV catalog could not be downloaded or read. The message is written for the person running the tool.""" + + @dataclass(frozen=True) class Advisory: cve_id: str @@ -58,9 +64,25 @@ def load_feed(cache: Path = DEFAULT_CACHE, refresh: bool = False, timeout: int = """Return (advisories, catalog_version). Downloads when the cache is missing or refresh is set.""" if refresh or not cache.exists(): cache.parent.mkdir(parents=True, exist_ok=True) - with urllib.request.urlopen(KEV_URL, timeout=timeout) as resp: # noqa: S310 - fixed https URL - cache.write_bytes(resp.read()) - raw = json.loads(cache.read_text()) + try: + with urllib.request.urlopen(KEV_URL, timeout=timeout) as resp: # noqa: S310 - fixed https URL + body = resp.read() + except (urllib.error.URLError, TimeoutError, OSError) as e: + raise FeedError( + f"Could not download the CISA KEV catalog from {KEV_URL} ({e}). " + f"Check the network connection, or place a previously downloaded copy at {cache}." + ) from e + try: + json.loads(body) + except ValueError as e: + raise FeedError(f"The download from {KEV_URL} was not valid JSON ({e}). The existing cache, if any, was left untouched.") from e + cache.write_bytes(body) + try: + raw = json.loads(cache.read_text()) + except ValueError as e: + raise FeedError(f"The cached catalog at {cache} is not valid JSON ({e}). Delete it or run with --refresh.") from e + if not isinstance(raw, dict) or "vulnerabilities" not in raw: + raise FeedError(f"The catalog at {cache} does not look like the CISA KEV feed (no 'vulnerabilities' key). Run with --refresh.") return parse_feed(raw), str(raw.get("catalogVersion", "unknown")) diff --git a/patchowner/report.py b/patchowner/report.py index cee10aa..3ed91a6 100644 --- a/patchowner/report.py +++ b/patchowner/report.py @@ -1,4 +1,5 @@ """Render the replay result as a single HTML page: notices tab and policy-tree tab.""" + from __future__ import annotations import json @@ -7,25 +8,31 @@ from jinja2 import Environment, FileSystemLoader, select_autoescape -from .health import Health from .decide import ACT_NOW, DELIVERY, OUTCOME_TO_URGENCY, PLAN_UPDATE, UPDATE_SOON, WATCH, WHEN_TEXT, Decision, Summary +from .health import Health from .ssvc import AUTOMATABLE, EXPOSURE, HUMAN_IMPACT, OUTCOME_WORDS, OUTCOMES, POINTS, Policy, plain_policy _env = Environment( loader=FileSystemLoader(Path(__file__).parent / "templates"), autoescape=select_autoescape(["html"]), ) -SITEWIDE = ("This site is for educational and testing purposes only. Recommendations shown here are not professional security advice. " - "For vulnerabilities on the CISA KEV catalog, the default action is immediate patching per CISA guidance. " - "Always verify against vendor advisories and consult your security team before deferring any update.") -BANNER = ("Demo for testing the prioritization logic only. Not security advice. Every vulnerability here is actively exploited, " - "and CISA's guidance is to patch immediately.") -SLOW = {"defer": "Defer", "scheduled": "Plan update"} # answers that slow-walk an exploited vulnerability; each carries a caution +SITEWIDE = ( + "This site is for educational and testing purposes only. Recommendations shown here are not professional security advice. " + "For vulnerabilities on the CISA KEV catalog, the default action is immediate patching per CISA guidance. " + "Always verify against vendor advisories and consult your security team before deferring any update." +) +BANNER = ( + "Demo for testing the prioritization logic only. Not security advice. Every vulnerability here is actively exploited, " + "and CISA's guidance is to patch immediately." +) +SLOW = {"defer": "Defer", "scheduled": "Plan update"} # answers that slow-walk an exploited vulnerability; each carries a caution def caution(word: str) -> str: - return (f"Simulated recommendation: {word}, for testing the prioritization logic only. This vulnerability is actively exploited " - f"(CISA KEV). \u201c{word}\u201d here does not reflect CISA or vendor guidance. In a real environment, apply the patch immediately.") + return ( + f"Simulated recommendation: {word}, for testing the prioritization logic only. This vulnerability is actively exploited " + f"(CISA KEV). \u201c{word}\u201d here does not reflect CISA or vendor guidance. In a real environment, apply the patch immediately." + ) KLASS = {ACT_NOW: "now", UPDATE_SOON: "soon", PLAN_UPDATE: "plan", WATCH: "watch", "Defer": "defer"} @@ -47,10 +54,16 @@ def build_tree(policy: Policy, decisions: list[Decision]) -> list[dict]: values = ("active", ex, auto, hi) outcome, row = policy.outcome_for(values) leaf_decisions = by_leaf.get(values, []) - auto_node["children"].append({ - "value": hi, "key": "|".join(values), "row": row, "outcome": outcome, - "count": len(leaf_decisions), "notices": leaf_decisions, - }) + auto_node["children"].append( + { + "value": hi, + "key": "|".join(values), + "row": row, + "outcome": outcome, + "count": len(leaf_decisions), + "notices": leaf_decisions, + } + ) auto_node["count"] += len(leaf_decisions) ex_node["children"].append(auto_node) ex_node["count"] += auto_node["count"] @@ -58,47 +71,89 @@ def build_tree(policy: Policy, decisions: list[Decision]) -> list[dict]: return tree +def _script_safe_json(data: dict) -> str: + """JSON that can sit inside a ' in either must not be able to end the tag; the characters are written as JSON escapes instead.""" + return json.dumps(data).replace("<", "\\u003c").replace(">", "\\u003e").replace("&", "\\u0026") + + def _client_data(policy: Policy, decisions: list[Decision], share: str) -> str: """What the page needs to relabel a leaf and recompute counts without a server.""" notices = [ - {"id": i, "key": d.key, "leaf": d.leaf_key, "altLeaf": d.alt_leaf_key, "question": bool(d.questions), "sent": d.sent, - "cve": d.match.advisory.cve_id, "asset": d.match.asset.asset, "to": d.recipient_email, - "state": {"action": d.state.action, "by": d.state.by, "note": d.state.note, "sentence": d.state.sentence} if d.state else None} + { + "id": i, + "key": d.key, + "leaf": d.leaf_key, + "altLeaf": d.alt_leaf_key, + "question": bool(d.questions), + "sent": d.sent, + "cve": d.match.advisory.cve_id, + "asset": d.match.asset.asset, + "to": d.recipient_email, + "state": {"action": d.state.action, "by": d.state.by, "note": d.state.note, "sentence": d.state.sentence} if d.state else None, + } for i, d in enumerate(decisions) ] rows = [{"row": row, "values": list(values), "outcome": outcome} for values, (outcome, row) in policy.rows.items()] - return json.dumps({ - "notices": notices, "rows": rows, "header": policy.header, - "share": share, "slow": SLOW, "cautions": {o: caution(w) for o, w in SLOW.items()}, - "words": OUTCOME_TO_URGENCY, "klass": OUTCOME_KLASS, "outcomes": list(OUTCOMES), - "delivery": {o: {"ack": dl.acknowledge_within, "plan": dl.plan_within, "esc": dl.escalate_after, "oncall": dl.notify_oncall} for o, dl in DELIVERY.items()}, - "points": [{"name": pt.name, "question": pt.question, "values": list(pt.values), "plain": pt.plain, "clause": pt.clause} for pt in POINTS], - }) + return _script_safe_json( + { + "notices": notices, + "rows": rows, + "header": policy.header, + "share": share, + "slow": SLOW, + "cautions": {o: caution(w) for o, w in SLOW.items()}, + "words": OUTCOME_TO_URGENCY, + "klass": OUTCOME_KLASS, + "outcomes": list(OUTCOMES), + "delivery": { + o: {"ack": dl.acknowledge_within, "plan": dl.plan_within, "esc": dl.escalate_after, "oncall": dl.notify_oncall} + for o, dl in DELIVERY.items() + }, + "points": [ + {"name": pt.name, "question": pt.question, "values": list(pt.values), "plain": pt.plain, "clause": pt.clause} + for pt in POINTS + ], + } + ) def share_text(s: Summary, health: Health, inventory_name: str) -> str: """The plain-text summary the Share button copies: what a person would paste into a chat.""" urgent = s.by_urgency.get(ACT_NOW, 0) - return (f"PatchOwner replay: KEV catalog {s.catalog_version}, last {s.days} days, against {inventory_name} ({s.assets} assets). " - f"{s.advisories_in_window} advisories published, {s.relevant_advisories} touched something we own, " - f"{s.sent} notices sent ({urgent} Act now), {s.suppressed} suppressed with a reason. " - f"Health: {health.word}" + (f", {health.issues} thing{'s' if health.issues != 1 else ''} to fix." if health.issues else ".") - + " Demo only, not security advice.") + return ( + f"PatchOwner replay: KEV catalog {s.catalog_version}, last {s.days} days, against {inventory_name} ({s.assets} assets). " + f"{s.advisories_in_window} advisories published, {s.relevant_advisories} touched something we own, " + f"{s.sent} notices sent ({urgent} Act now), {s.suppressed} suppressed with a reason. " + f"Health: {health.word}" + + (f", {health.issues} thing{'s' if health.issues != 1 else ''} to fix." if health.issues else ".") + + " Demo only, not security advice." + ) def render_html(summary: Summary, decisions: list[Decision], policy: Policy, health: Health, *, inventory_name: str) -> str: tpl = _env.get_template("report.html") return tpl.render( - s=summary, inventory_name=inventory_name, policy=policy, health=health, + s=summary, + inventory_name=inventory_name, + policy=policy, + health=health, decisions=decisions, sent=[d for d in decisions if d.sent], suppressed=[d for d in decisions if not d.sent], - when_text=WHEN_TEXT, klass=KLASS, outcome_klass=OUTCOME_KLASS, outcomes=OUTCOMES, points=POINTS, + when_text=WHEN_TEXT, + klass=KLASS, + outcome_klass=OUTCOME_KLASS, + outcomes=OUTCOMES, + points=POINTS, urgencies=[ACT_NOW, UPDATE_SOON, PLAN_UPDATE, WATCH], tree=build_tree(policy, decisions), plain_lines=plain_policy(policy), delivery=DELIVERY, outcome_words=OUTCOME_WORDS, - sitewide=SITEWIDE, banner=BANNER, slow=SLOW, caution=caution, + sitewide=SITEWIDE, + banner=BANNER, + slow=SLOW, + caution=caution, client_data=_client_data(policy, decisions, share_text(summary, health, inventory_name)), ) diff --git a/patchowner/ssvc.py b/patchowner/ssvc.py index 919f956..9abc2dd 100644 --- a/patchowner/ssvc.py +++ b/patchowner/ssvc.py @@ -3,6 +3,7 @@ Decision points and their values are the SEI/CERT vocabulary and are not customizable. The outcome label on each row of a policy is the organization's risk appetite and is. """ + from __future__ import annotations import csv @@ -20,37 +21,58 @@ @dataclass(frozen=True) class DecisionPoint: - name: str # SSVC name, fixed vocabulary, shown to auditors - key: str # abbreviated-vector key per SSVC v2 "Communication Formats" - values: tuple[str, ...] # SSVC values, fixed vocabulary + name: str # SSVC name, fixed vocabulary, shown to auditors + key: str # abbreviated-vector key per SSVC v2 "Communication Formats" + values: tuple[str, ...] # SSVC values, fixed vocabulary abbrev: dict[str, str] - question: str # the same question in plain words, shown to people - plain: dict[str, str] # SSVC value -> plain words - clause: dict[str, str] # SSVC value -> "because ..." clause for one-sentence explanations + question: str # the same question in plain words, shown to people + plain: dict[str, str] # SSVC value -> plain words + clause: dict[str, str] # SSVC value -> "because ..." clause for one-sentence explanations def word(self, value: str) -> str: return self.plain[value] EXPLOITATION = DecisionPoint( - "Exploitation", "E", ("none", "public poc", "active"), {"none": "N", "public poc": "P", "active": "A"}, - "Is it being attacked?", {"none": "no", "public poc": "a proof exists", "active": "yes, right now"}, + "Exploitation", + "E", + ("none", "public poc", "active"), + {"none": "N", "public poc": "P", "active": "A"}, + "Is it being attacked?", + {"none": "no", "public poc": "a proof exists", "active": "yes, right now"}, {"none": "nobody is attacking it", "public poc": "a public proof of attack exists", "active": "attackers are using it right now"}, ) EXPOSURE = DecisionPoint( - "System Exposure", "Se", ("small", "controlled", "open"), {"small": "S", "controlled": "C", "open": "O"}, - "Can attackers reach it?", {"small": "isolated", "controlled": "from inside only", "open": "from the internet"}, + "System Exposure", + "Se", + ("small", "controlled", "open"), + {"small": "S", "controlled": "C", "open": "O"}, + "Can attackers reach it?", + {"small": "isolated", "controlled": "from inside only", "open": "from the internet"}, {"small": "it's isolated", "controlled": "it's reachable only from inside", "open": "it's reachable from the internet"}, ) AUTOMATABLE = DecisionPoint( - "Automatable", "A", ("no", "yes"), {"no": "N", "yes": "Y"}, - "Can the attack run by itself?", {"no": "no, it needs a person", "yes": "yes"}, + "Automatable", + "A", + ("no", "yes"), + {"no": "N", "yes": "Y"}, + "Can the attack run by itself?", + {"no": "no, it needs a person", "yes": "yes"}, {"no": "the attack needs a person's help", "yes": "the attack can run by itself"}, ) HUMAN_IMPACT = DecisionPoint( - "Human Impact", "H", ("low", "medium", "high", "very high"), {"low": "L", "medium": "M", "high": "H", "very high": "Vh"}, - "How much would it hurt?", {"low": "a little", "medium": "some", "high": "a lot", "very high": "business-stopping"}, - {"low": "it would hurt a little", "medium": "it would hurt some", "high": "it would hurt a lot", "very high": "it could stop the business"}, + "Human Impact", + "H", + ("low", "medium", "high", "very high"), + {"low": "L", "medium": "M", "high": "H", "very high": "Vh"}, + "How much would it hurt?", + {"low": "a little", "medium": "some", "high": "a lot", "very high": "business-stopping"}, + { + "low": "it would hurt a little", + "medium": "it would hurt some", + "high": "it would hurt a lot", + "very high": "it could stop the business", + }, ) POINTS: tuple[DecisionPoint, ...] = (EXPLOITATION, EXPOSURE, AUTOMATABLE, HUMAN_IMPACT) OUTCOMES: tuple[str, ...] = ("defer", "scheduled", "out-of-cycle", "immediate") @@ -58,9 +80,21 @@ def word(self, value: str) -> str: CRITICALITY_TO_IMPACT = {"low": "low", "medium": "medium", "high": "high", "critical": "very high"} NOT_AUTOMATABLE_HINTS = ( # most specific first: the first hit is quoted in the question to a person - "authenticated attacker", "authenticated user", "user interaction", "physical access", "local attacker", - "local user", "valid credentials", "crafted file", "malicious file", "an authenticated", "locally", - "convince", "tricking", "opening a", "open a", + "authenticated attacker", + "authenticated user", + "user interaction", + "physical access", + "local attacker", + "local user", + "valid credentials", + "crafted file", + "malicious file", + "an authenticated", + "locally", + "convince", + "tricking", + "opening a", + "open a", ) @@ -85,7 +119,7 @@ def values(self) -> tuple[str, ...]: def leaf_key(self) -> str: return "|".join(self.values) - def with_value(self, point: DecisionPoint, value: str) -> "Assessment": + def with_value(self, point: DecisionPoint, value: str) -> Assessment: return Assessment(tuple(s if s.point is not point else Step(point, value, "confirmed by a person", True) for s in self.steps)) @property @@ -120,12 +154,12 @@ class Policy: header: list[str] @classmethod - def load(cls, path: Path | str, name: str | None = None) -> "Policy": + def load(cls, path: Path | str, name: str | None = None) -> Policy: path = Path(path) return cls.parse(path.read_text(), name or path.stem) @classmethod - def parse(cls, text: str, name: str) -> "Policy": + def parse(cls, text: str, name: str) -> Policy: reader = csv.reader(io.StringIO(text)) header = next(reader, None) if not header or len(header) != 2 + len(POINTS): @@ -134,10 +168,15 @@ def parse(cls, text: str, name: str) -> "Policy": for line_no, line in enumerate(reader, start=2): if not line or not any(c.strip() for c in line): continue - row_num = int(line[0]) - values = tuple(c.strip().lower() for c in line[1:1 + len(POINTS)]) + if len(line) != 2 + len(POINTS): + raise PolicyError(f"Line {line_no}: expected {2 + len(POINTS)} columns, found {len(line)}.") + try: + row_num = int(line[0]) + except ValueError: + raise PolicyError(f"Line {line_no}: the first column must be the row number, not '{line[0]}'.") from None + values = tuple(c.strip().lower() for c in line[1 : 1 + len(POINTS)]) outcome = line[1 + len(POINTS)].strip().lower() - for p, v in zip(POINTS, values): + for p, v in zip(POINTS, values, strict=True): if v not in p.values: raise PolicyError(f"Line {line_no}: '{v}' is not a valid {p.name} value ({', '.join(p.values)}).") if outcome not in OUTCOMES: @@ -151,7 +190,7 @@ def parse(cls, text: str, name: str) -> "Policy": return cls(name=name, rows=rows, header=header) @classmethod - def default(cls) -> "Policy": + def default(cls) -> Policy: return cls.load(DEFAULT_POLICY, "SEI deployer tree (default)") def evaluate(self, a: Assessment) -> Outcome: @@ -192,11 +231,15 @@ def assess(adv: Advisory, asset: Asset) -> Assessment: steps.append(Step(EXPOSURE, "controlled", "not internet-facing; assumed reachable from the internal network", False)) hint = automatable_hint(adv.description) - steps.append(Step( - AUTOMATABLE, "yes", - "assumed the attack can run by itself, the worst case, until a person says otherwise", False, - question=f"CISA's text mentions “{hint}”. Does this attack need a person's help?" if hint else None, - )) + steps.append( + Step( + AUTOMATABLE, + "yes", + "assumed the attack can run by itself, the worst case, until a person says otherwise", + False, + question=f"CISA's text mentions “{hint}”. Does this attack need a person's help?" if hint else None, + ) + ) if asset.human_impact: steps.append(Step(HUMAN_IMPACT, asset.human_impact, "human impact set explicitly in the inventory", True)) diff --git a/patchowner/web.py b/patchowner/web.py index 299163f..dd27b59 100644 --- a/patchowner/web.py +++ b/patchowner/web.py @@ -1,20 +1,27 @@ """Upload a CSV, get the replay page. The whole web demo.""" + from __future__ import annotations +import html from pathlib import Path from fastapi import FastAPI, File, Form, UploadFile from fastapi.responses import HTMLResponse, JSONResponse -from pydantic import BaseModel +from pydantic import BaseModel, Field from .engine import run_replay -from .report import BANNER, SITEWIDE from .inventory import OPTIONAL, REQUIRED, InventoryError +from .kev import FeedError +from .report import BANNER, SITEWIDE from .state import ACTIONS, StateStore app = FastAPI(title="PatchOwner demo") _store = StateStore("out/state.json") +MAX_UPLOAD_BYTES = 5 * 1024 * 1024 # an inventory CSV is a few KB; anything bigger is a mistake, not a list of assets +MIN_DAYS, MAX_DAYS = 1, 3650 +DEFAULT_FALLBACK = "security@example.com" + def configure(state_path: str | Path) -> None: global _store @@ -22,10 +29,10 @@ def configure(state_path: str | Path) -> None: class Act(BaseModel): - key: str - action: str - by: str = "" - note: str = "" + key: str = Field(min_length=1, max_length=500) + action: str = Field(max_length=50) + by: str = Field(default="", max_length=200) + note: str = Field(default="", max_length=2000) @app.post("/act") @@ -33,7 +40,10 @@ def act(a: Act) -> JSONResponse: if a.action not in ACTIONS: return JSONResponse({"ok": False, "error": f"unknown action; use one of {', '.join(ACTIONS)}"}, status_code=400) rec = _store.record(a.key, a.action, a.by, a.note) - return JSONResponse({"ok": True, "state": {"action": rec.action, "by": rec.by, "note": rec.note, "at": rec.at, "sentence": rec.sentence}}) + return JSONResponse( + {"ok": True, "state": {"action": rec.action, "by": rec.by, "note": rec.note, "at": rec.at, "sentence": rec.sentence}} + ) + FORM = """ PatchOwner @@ -61,8 +71,9 @@ def act(a: Act) -> JSONResponse: def _form(error: str = "") -> str: # str.format is not used here: the CSS braces would collide with it. return ( - FORM.replace("@@BANNER@@", BANNER).replace("@@SITEWIDE@@", SITEWIDE) - .replace("@@ERROR@@", f'

{error}
' if error else "") + FORM.replace("@@BANNER@@", BANNER) + .replace("@@SITEWIDE@@", SITEWIDE) + .replace("@@ERROR@@", f'
{html.escape(error)}
' if error else "") .replace("@@REQUIRED@@", ", ".join(f"{c}" for c in REQUIRED)) .replace("@@OPTIONAL@@", ", ".join(f"{c}" for c in OPTIONAL)) ) @@ -74,8 +85,15 @@ def index() -> str: @app.post("/replay", response_class=HTMLResponse) -async def replay(inventory: UploadFile = File(...), days: int = Form(90), fallback: str = Form("security@example.com")) -> str: - raw = await inventory.read() +async def replay(inventory: UploadFile = File(...), days: int = Form(90), fallback: str = Form(DEFAULT_FALLBACK)) -> str: + if not MIN_DAYS <= days <= MAX_DAYS: + return _form(f"Look back must be between {MIN_DAYS} and {MAX_DAYS} days.") + fallback = fallback.strip().lower() or DEFAULT_FALLBACK + raw = await inventory.read(MAX_UPLOAD_BYTES + 1) + if len(raw) > MAX_UPLOAD_BYTES: + return _form( + f"That file is larger than {MAX_UPLOAD_BYTES // (1024 * 1024)} MB. An inventory CSV should be far smaller; check you picked the right file." + ) try: text = raw.decode("utf-8-sig") except UnicodeDecodeError: @@ -84,4 +102,6 @@ async def replay(inventory: UploadFile = File(...), days: int = Form(90), fallba r = run_replay(text, inventory_name=inventory.filename or "inventory.csv", days=days, fallback_email=fallback, state=_store) except InventoryError as e: return _form(str(e)) + except FeedError as e: + return _form(str(e)) return r.html diff --git a/tests/test_kev.py b/tests/test_kev.py new file mode 100644 index 0000000..34b0ccb --- /dev/null +++ b/tests/test_kev.py @@ -0,0 +1,64 @@ +"""The KEV feed: a bad download or a bad cache is a FeedError with a sentence, not a traceback.""" + +import json +import urllib.error +import urllib.request +from datetime import date + +import pytest + +from patchowner import kev + +FEED = { + "catalogVersion": "2026.09.29", + "vulnerabilities": [ + { + "cveID": "CVE-2026-1", + "vendorProject": "Fortinet", + "product": "FortiOS", + "vulnerabilityName": "n", + "shortDescription": "d", + "requiredAction": "Apply updates.", + "dateAdded": "2026-09-01", + "dueDate": "2026-09-15", + "knownRansomwareCampaignUse": "Known", + } + ], +} + + +def test_cached_feed_is_parsed(tmp_path): + cache = tmp_path / "kev.json" + cache.write_text(json.dumps(FEED)) + advisories, version = kev.load_feed(cache) + assert version == "2026.09.29" and advisories[0].cve_id == "CVE-2026-1" + assert advisories[0].ransomware_known and advisories[0].due_date == date(2026, 9, 15) + + +def test_download_failure_is_a_feed_error_and_leaves_no_cache(tmp_path, monkeypatch): + def fail(*_a, **_k): + raise urllib.error.URLError("no route to host") + + monkeypatch.setattr(urllib.request, "urlopen", fail) + cache = tmp_path / "kev.json" + with pytest.raises(kev.FeedError, match="Could not download"): + kev.load_feed(cache) + assert not cache.exists() + + +def test_corrupt_cache_is_a_feed_error(tmp_path): + cache = tmp_path / "kev.json" + cache.write_text("{not json") + with pytest.raises(kev.FeedError, match="not valid JSON"): + kev.load_feed(cache) + cache.write_text(json.dumps({"something": "else"})) + with pytest.raises(kev.FeedError, match="does not look like the CISA KEV feed"): + kev.load_feed(cache) + + +def test_window_keeps_recent_entries_newest_first(): + old = kev.parse_feed(FEED)[0] + new = kev.Advisory("CVE-2026-2", "V", "P", "", "", "", date(2026, 9, 20), None, False) + got = kev.in_window([old, new], days=30, today=date(2026, 9, 25)) + assert [a.cve_id for a in got] == ["CVE-2026-2", "CVE-2026-1"] + assert kev.in_window([old, new], days=3, today=date(2026, 9, 25)) == [] diff --git a/tests/test_report.py b/tests/test_report.py new file mode 100644 index 0000000..5bdd55a --- /dev/null +++ b/tests/test_report.py @@ -0,0 +1,56 @@ +"""The report is one HTML file that takes text from outside (inventory, CISA) and must not let it run.""" + +from datetime import date + +from patchowner.decide import decide, summarize +from patchowner.health import assess_health +from patchowner.inventory import Asset +from patchowner.kev import Advisory +from patchowner.matching import Match +from patchowner.report import _client_data, render_html, share_text +from patchowner.ssvc import Policy + +TODAY = date(2026, 9, 10) +HOSTILE = "" + + +def hostile_decisions(): + adv = Advisory( + cve_id="CVE-2026-1", + vendor="Fortinet", + product="FortiOS", + name="", + description=HOSTILE, + required_action="Apply update.", + date_added=date(2026, 9, 1), + due_date=None, + ransomware_known=False, + ) + asset = Asset( + asset=HOSTILE, + vendor="Fortinet", + product="FortiOS", + internet_exposed=True, + criticality="high", + owner_email="dana@x", + owner_name="Dana", + ) + return decide([Match(adv, asset, "exact", 100, "r")], fallback_email="sec@x", today=TODAY) + + +def test_client_data_cannot_close_the_script_tag(): + out = _client_data(Policy.default(), hostile_decisions(), share="s") + assert "" not in out + assert "\\u003c/script\\u003e" in out + + +def test_rendered_report_escapes_hostile_text_everywhere(): + ds = hostile_decisions() + s = summarize(ds, days=90, catalog_version="2026.09.10", policy_name="p", advisories_in_window=1, assets=1, budget=10, warnings=[]) + h = assess_health([d.match.asset for d in ds], ds, s, today=TODAY) + html = render_html(s, ds, Policy.default(), h, inventory_name="inventory.csv") + # Exactly the page's own scripts: the theme bootstrap, the data block, and the behaviour script. + assert html.count("") == 3 + assert "alert(1)" in html # the text is shown, as text + assert "