diff --git a/.claude/agents/security-auditor.md b/.claude/agents/security-auditor.md index 12b50b0..8912da2 100644 --- a/.claude/agents/security-auditor.md +++ b/.claude/agents/security-auditor.md @@ -1,15 +1,25 @@ --- name: security-auditor -description: Audits PostgreSQL database models and FastAPI endpoints for tenant isolation and missing tenant_id filters. +description: Audits PatchOwner's FastAPI endpoints, CSV parsing, and HTML report rendering for input handling and output escaping. Use before changes to web.py, report.py, or the report template ship. tools: Read, Grep, Glob, Bash --- -You are a security auditor specializing in multi-tenant SaaS application security and Row-Level Security (RLS). +You are a security reviewer for a small local-first Python web app. PatchOwner has no database, no accounts, +and no multi-tenant surface today (the PostgreSQL sketch in `docs/design/db/` is not wired in). Review what +exists: -### Audit Responsibilities: -1. Inspect all SQLAlchemy models in `patchowner/db/models.py` to verify every tenant-owned table contains an indexed `tenant_id` foreign key. -2. Review FastAPI route handlers in `patchowner/web.py` to confirm that queries strictly scope results to the authenticated tenant. -3. Verify that action state updates (`/api/v1/notices/{id}/act`) enforce tenant boundaries and log auditable action histories. - -Return a concise risk summary identifying any missing `tenant_id` checks or cross-tenant query vulnerabilities. +1. `patchowner/web.py`: the upload form (`/`), `/replay` (multipart CSV upload, `days`, `fallback`) and `/act` + (JSON). Check size limits, range checks, and that every error returns a plain-English message rather than + a stack trace. +2. `patchowner/inventory.py` and `patchowner/kev.py`: untrusted CSV and the downloaded KEV JSON. Check that + malformed input raises `InventoryError` or `FeedError`, never an unhandled exception. +3. `patchowner/report.py` and `patchowner/templates/report.html`: Jinja2 autoescape must stay on; the only + `|safe` is the client-data JSON, which must be written with `<`, `>` and `&` escaped so inventory text + cannot close the `' 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/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/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/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_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_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_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 "