From d91ac851cfbb3525033f058e6df68e5d6964fe09 Mon Sep 17 00:00:00 2001 From: Alex Schapiro Date: Fri, 7 Aug 2026 00:51:26 +0000 Subject: [PATCH] coverage: make the negative space of a scan a first-class artifact Promote the coverage ledger from runtime state to a deliverable: coverage.json beside vulnerabilities.json, a Coverage section rendered into the report from the ledger rather than transcribed by an agent, and SARIF pass / notApplicable / open results so a consumer can tell 'tested and clean' from 'never tested'. Ground it in what the runtime observed rather than only what agents claimed: a risk class an agent carried a skill for and never accounted for is published as a gap (and surfaced back to the root agent from finish_scan while it can still act), and a run cut short is stamped incomplete on both the artifact and the SARIF invocation. Also: make the ledger's duplicate check and insertion one critical section and persist under the lock; key a checkout and the URL it was cloned from onto one threat-model identity; render the calibration metadata (counterevidence, confidence, severity change conditions, fix verification) that was being stored and then dropped. --- strix/report/coverage.py | 441 ++++++++++++++++++++++++++ strix/report/sarif.py | 158 +++++++++ strix/report/state.py | 39 ++- strix/report/writer.py | 57 +++- strix/tools/coverage/tools.py | 89 ++++-- strix/tools/finish/tool.py | 42 ++- strix/tools/threat_model/tools.py | 38 ++- tests/test_coverage_tool.py | 45 +++ tests/test_finish_coverage_gate.py | 65 ++++ tests/test_report_coverage.py | 203 ++++++++++++ tests/test_report_writer.py | 35 ++ tests/test_sarif.py | 129 ++++++++ tests/test_state_coverage_artifact.py | 77 +++++ tests/test_threat_model_tool.py | 34 ++ 14 files changed, 1390 insertions(+), 62 deletions(-) create mode 100644 strix/report/coverage.py create mode 100644 tests/test_finish_coverage_gate.py create mode 100644 tests/test_report_coverage.py create mode 100644 tests/test_state_coverage_artifact.py diff --git a/strix/report/coverage.py b/strix/report/coverage.py new file mode 100644 index 00000000..5ffc7e71 --- /dev/null +++ b/strix/report/coverage.py @@ -0,0 +1,441 @@ +"""``coverage.json`` — the negative space of a scan, with provenance. + +A findings list answers "what is wrong". It cannot answer "what did you +check", and in a compliance context that second question is the one that +decides whether a clean result means anything: an auditor reading zero SQL +injection findings cannot tell "tested fourteen endpoints, all parameterized" +apart from "never looked". + +This module assembles the artifact that answers it. Two kinds of statement go +in, and they are kept apart on purpose: + +- ``agent_reported`` — the coverage ledger (:mod:`strix.tools.coverage.tools`). + Rich and specific, but it is an agent's account of its own work. +- ``machine_observed`` — facts the runtime recorded regardless of what any + agent claimed: which agents ran and how they terminated, which skills they + carried, how many findings were filed, whether the run finished or was cut + short. + +A coverage claim is an attestation, so conflating the two would be the worst +possible failure: a hallucinated "tested and clean" is strictly less honest +than no coverage section at all. Every entry therefore carries its ``source``, +and machine-observed facts contradict rather than confirm — an agent that +carried the ``sql_injection`` skill and recorded nothing about SQL injection +shows up under ``gaps``, and a run that hit its budget ceiling is stamped +``complete: false`` no matter how tidy the ledger looks. +""" + +from __future__ import annotations + +import json +import logging +from datetime import UTC, datetime +from typing import TYPE_CHECKING, Any + +from strix.report.writer import atomic_write_text +from strix.skills import get_available_skills + + +if TYPE_CHECKING: + from pathlib import Path + + +logger = logging.getLogger(__name__) + +COVERAGE_FILENAME = "coverage.json" +COVERAGE_SCHEMA_VERSION = 1 + +#: Ledger outcomes rendered for a reader who has never seen our enum. +OUTCOME_LABELS: dict[str, str] = { + "reported": "Finding reported", + "no_issue_found": "No issue identified", + "ruled_out": "Ruled out", + "not_applicable": "Not applicable", + "needs_follow_up": "Requires further review", +} + +#: Statuses that mean the agent stopped early rather than finishing its task. +_INCOMPLETE_AGENT_STATUSES = frozenset({"crashed", "stopped", "running", "waiting"}) + +#: Run statuses that mean the scan itself did not run to completion. +_INCOMPLETE_RUN_STATUSES = frozenset({"failed", "interrupted", "stopped", "running"}) + +#: Only this skill category names a vulnerability class. ``tooling`` and +#: ``reconnaissance`` skills describe how an agent works, not what it hunts, +#: so holding one implies no coverage obligation. +_RISK_SKILL_CATEGORY = "vulnerabilities" + +#: Alternate phrasings for skills whose common name and whose canonical risk +#: wording share no words. Matching a skill to a ledger row is textual, so +#: without these an agent that recorded "object-level authorization" would be +#: reported as never having looked at ``idor``. Only genuine synonyms belong +#: here — a wrong alias suppresses a real gap, which is the costlier error. +_SKILL_ALIASES: dict[str, tuple[str, ...]] = { + "idor": ("object level authorization", "bola", "broken object level"), + "xss": ("cross site scripting",), + "csrf": ("cross site request forgery",), + "ssrf": ("server side request forgery",), + "rce": ("remote code execution", "command injection"), + "sqli": ("sql injection",), + "xxe": ("xml external entity",), + "lfi": ("local file inclusion", "path traversal"), + "ssti": ("server side template injection",), +} + + +def read_agent_graph(state_dir: Path) -> dict[str, Any]: + """Load the coordinator's snapshot, or ``{}`` when it isn't readable. + + The snapshot is the runtime's own record of the agent tree, written on + every graph mutation. Reading it here (rather than holding a coordinator + reference) keeps artifact assembly usable from a finished or resumed run, + where the live coordinator is gone but the file is still on disk. + """ + path = state_dir / "agents.json" + if not path.is_file(): + return {} + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + logger.warning("agent graph snapshot at %s is unreadable", path, exc_info=True) + return {} + return data if isinstance(data, dict) else {} + + +def _normalized(text: str) -> str: + """Lowercase *text* with punctuation flattened to spaces, for matching.""" + return "".join(char if char.isalnum() else " " for char in text.lower()) + + +def _skill_leaf(skill: str) -> str: + return skill.rsplit("/", maxsplit=1)[-1].strip().lower() + + +def _risk_skill_names() -> frozenset[str]: + """Bare names of every skill that denotes a vulnerability class.""" + try: + return frozenset(get_available_skills().get(_RISK_SKILL_CATEGORY, ())) + except OSError: + logger.warning("could not enumerate skills for coverage gaps", exc_info=True) + return frozenset() + + +def agents_from_graph(graph: dict[str, Any]) -> list[dict[str, Any]]: + """Flatten the coordinator snapshot into one record per agent.""" + statuses = graph.get("statuses") + if not isinstance(statuses, dict): + return [] + raw_names = graph.get("names") + names: dict[str, Any] = raw_names if isinstance(raw_names, dict) else {} + raw_metadata = graph.get("metadata") + metadata: dict[str, Any] = raw_metadata if isinstance(raw_metadata, dict) else {} + + agents: list[dict[str, Any]] = [] + for agent_id, status in statuses.items(): + raw_meta = metadata.get(agent_id) + meta: dict[str, Any] = raw_meta if isinstance(raw_meta, dict) else {} + raw_skills = meta.get("skills") + skills: list[Any] = raw_skills if isinstance(raw_skills, list) else [] + agents.append( + { + "agent_id": agent_id, + "agent_name": names.get(agent_id) or agent_id, + "status": str(status), + "skills": [str(skill) for skill in skills], + "task": str(meta.get("task") or ""), + } + ) + agents.sort(key=lambda agent: str(agent["agent_name"])) + return agents + + +def _skill_phrasings(skill: str) -> list[list[str]]: + """Word lists that would each count as a ledger row naming *skill*.""" + phrasings = [skill, *_SKILL_ALIASES.get(skill, ())] + return [terms for phrase in phrasings if (terms := _normalized(phrase).split())] + + +def _entry_is_about(entry: dict[str, Any], phrasings: list[list[str]]) -> bool: + """True when a ledger row plausibly concerns any phrasing of a risk class.""" + haystack = _normalized(f"{entry.get('risk_area', '')} {entry.get('surface', '')}") + return any(all(term in haystack for term in terms) for terms in phrasings) + + +def skill_coverage_gaps( + entries: list[dict[str, Any]], agents: list[dict[str, Any]] +) -> list[dict[str, Any]]: + """Vulnerability classes an agent was equipped for but never recorded. + + A skill assigned to an agent is a declaration of intent that the runtime + observed independently of anything the agent later said. When no ledger + row mentions that class, the class is unaccounted for — which is a very + different report line from "tested, nothing found". + """ + risk_skills = _risk_skill_names() + if not risk_skills: + return [] + + carriers: dict[str, list[str]] = {} + for agent in agents: + for skill in agent["skills"]: + leaf = _skill_leaf(skill) + if leaf in risk_skills: + carriers.setdefault(leaf, []).append(str(agent["agent_name"])) + + gaps: list[dict[str, Any]] = [] + for skill, agent_names in sorted(carriers.items()): + phrasings = _skill_phrasings(skill) + if any(_entry_is_about(entry, phrasings) for entry in entries): + continue + gaps.append( + { + "kind": "unrecorded_risk_class", + "risk_area": skill.replace("_", " "), + "detail": ( + f"Agent(s) {', '.join(sorted(set(agent_names)))} were assigned the " + f"'{skill}' skill, but no coverage entry records this class being " + "assessed. Treat it as unexamined, not as clean." + ), + } + ) + return gaps + + +def _silent_agent_gaps( + entries: list[dict[str, Any]], agents: list[dict[str, Any]] +) -> list[dict[str, Any]]: + """Agents that ran and recorded nothing at all.""" + recorded_ids = {str(entry.get("agent_id")) for entry in entries if entry.get("agent_id")} + gaps: list[dict[str, Any]] = [] + for agent in agents: + if agent["agent_id"] in recorded_ids: + continue + gaps.append( + { + "kind": "agent_recorded_no_coverage", + "agent_name": agent["agent_name"], + "detail": ( + f"{agent['agent_name']} ran (status: {agent['status']}) without " + "recording any coverage. Whatever it examined is absent from this " + "record." + ), + } + ) + return gaps + + +def _unresolved_gaps(entries: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Ledger rows the agents themselves left open.""" + return [ + { + "kind": "needs_follow_up", + "surface": entry.get("surface", ""), + "risk_area": entry.get("risk_area", ""), + "detail": str(entry.get("evidence") or "Left open without a stated reason."), + } + for entry in entries + if entry.get("outcome") == "needs_follow_up" + ] + + +def _completeness( + run_record: dict[str, Any], + agents: list[dict[str, Any]], + exit_reason: str | None, +) -> dict[str, Any]: + """Whether this record can be read as a complete account of the scan. + + Any of these makes it partial, and the caveats say which: the run did not + reach ``completed``, an agent was still live or died when the scan ended, + or the run stopped for a reason other than the root agent deciding it was + done (budget ceilings are the common case). + """ + status = str(run_record.get("status") or "unknown") + caveats: list[str] = [] + + if status in _INCOMPLETE_RUN_STATUSES: + caveats.append( + f"The scan ended with status '{status}' rather than completing, so coverage " + "reflects only the work finished before it stopped." + ) + unfinished = [agent for agent in agents if agent["status"] in _INCOMPLETE_AGENT_STATUSES] + if unfinished: + names = ", ".join(sorted(str(agent["agent_name"]) for agent in unfinished)) + caveats.append( + f"{len(unfinished)} agent(s) did not finish cleanly ({names}); any surface they " + "held is under-covered." + ) + if exit_reason and exit_reason not in {"finished_by_tool", "completed"}: + caveats.append( + f"The run terminated via '{exit_reason}' rather than the root agent finishing, " + "so remaining scope was not reached." + ) + + return { + "complete": not caveats, + "scan_status": status, + "exit_reason": exit_reason, + "caveats": caveats, + } + + +def _outcome_counts(entries: list[dict[str, Any]]) -> dict[str, int]: + counts: dict[str, int] = {} + for entry in entries: + outcome = str(entry.get("outcome", "")) + counts[outcome] = counts.get(outcome, 0) + 1 + return {label: counts[label] for label in OUTCOME_LABELS if label in counts} + + +def build_coverage_document( + *, + run_record: dict[str, Any], + entries: list[dict[str, Any]], + agent_graph: dict[str, Any], + vulnerability_reports: list[dict[str, Any]], + exit_reason: str | None = None, +) -> dict[str, Any]: + """Assemble the ``coverage.json`` document.""" + agents = agents_from_graph(agent_graph) + skills_exercised = sorted( + {_skill_leaf(skill) for agent in agents for skill in agent["skills"] if skill} + ) + + ledger = [ + { + "surface": entry.get("surface", ""), + "risk_area": entry.get("risk_area", ""), + "outcome": entry.get("outcome", ""), + "outcome_label": OUTCOME_LABELS.get(str(entry.get("outcome", "")), ""), + "evidence": entry.get("evidence", ""), + "recorded_by": entry.get("agent_name", ""), + "recorded_at": entry.get("created_at", ""), + "updated_at": entry.get("updated_at", ""), + "previous_outcomes": [ + str(previous.get("outcome", "")) + for previous in entry.get("history", []) + if isinstance(previous, dict) + ], + "source": "agent_reported", + } + for entry in entries + ] + + gaps = [ + *_unresolved_gaps(entries), + *skill_coverage_gaps(entries, agents), + *_silent_agent_gaps(entries, agents), + ] + + return { + "schema_version": COVERAGE_SCHEMA_VERSION, + "generated_at": datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S UTC"), + "run_id": run_record.get("run_id"), + "run_name": run_record.get("run_name"), + "scope": { + "targets": run_record.get("targets_info") or [], + "scan_mode": run_record.get("scan_mode"), + "scope_mode": run_record.get("scope_mode"), + "diff_scope": run_record.get("diff_scope"), + "instruction": run_record.get("instruction") or "", + }, + "summary": { + "surfaces_reviewed": len(ledger), + "outcomes": _outcome_counts(entries), + "findings_filed": len(vulnerability_reports), + "gaps": len(gaps), + }, + "machine_observed": { + "agents": agents, + "skills_exercised": skills_exercised, + "findings_filed": len(vulnerability_reports), + "source": "runtime", + }, + "completeness": _completeness(run_record, agents, exit_reason), + "entries": ledger, + "gaps": gaps, + } + + +def write_coverage(run_dir: Path, document: dict[str, Any]) -> Path: + """Write ``coverage.json`` into the run directory and return its path.""" + path = run_dir / COVERAGE_FILENAME + atomic_write_text(path, json.dumps(document, ensure_ascii=False, indent=2, default=str)) + logger.info( + "Saved coverage record to: %s (%d surface(s), %d gap(s))", + path, + len(document.get("entries", [])), + len(document.get("gaps", [])), + ) + return path + + +def render_coverage_markdown(document: dict[str, Any]) -> str: + """Render the coverage appendix appended to the executive report. + + Built from the document rather than written by an agent, so the table in + the client-facing report cannot drift from the ledger it summarizes. + """ + summary = document.get("summary", {}) + completeness = document.get("completeness", {}) + entries = document.get("entries", []) + gaps = document.get("gaps", []) + + lines = [ + "# Coverage", + "", + "Surfaces assessed during this engagement and how each was closed. Entries are " + "reported by the testing agents; the completeness notes below are recorded by " + "the runtime.", + "", + f"**Surfaces reviewed:** {summary.get('surfaces_reviewed', 0)} ", + f"**Findings filed:** {summary.get('findings_filed', 0)} ", + f"**Open items and gaps:** {summary.get('gaps', 0)}", + "", + ] + + if entries: + lines += [ + "| Surface | Risk Area | Outcome | Evidence |", + "| --- | --- | --- | --- |", + ] + lines += [ + "| {} | {} | {} | {} |".format( + _cell(entry.get("surface")), + _cell(entry.get("risk_area")), + _cell(entry.get("outcome_label") or entry.get("outcome")), + _cell(entry.get("evidence")), + ) + for entry in entries + ] + lines.append("") + else: + lines += [ + "No surfaces were recorded for this scan. The absence of findings below " + "cannot be read as evidence that any particular area was tested.", + "", + ] + + if gaps: + lines += ["## Not Covered", ""] + lines += [f"- {_inline(gap.get('detail'))}" for gap in gaps] + lines.append("") + + caveats = completeness.get("caveats") or [] + if caveats: + lines += ["## Completeness", ""] + lines += [f"- {_inline(caveat)}" for caveat in caveats] + lines.append("") + + return "\n".join(lines) + + +def _cell(value: Any) -> str: + """Flatten *value* for a markdown table cell.""" + text = _inline(value) + return text.replace("|", "\\|") if text else "—" + + +def _inline(value: Any) -> str: + """Collapse *value* to a single line of plain text.""" + return " ".join(str(value or "").split()) diff --git a/strix/report/sarif.py b/strix/report/sarif.py index fc6e05db..24846146 100644 --- a/strix/report/sarif.py +++ b/strix/report/sarif.py @@ -40,6 +40,15 @@ Design notes: * Findings without safe locations still appear in the SARIF output, anchored to SECURITY.md and flagged via ``properties.synthetic_location`` rather than being dropped silently. + * Coverage — what was examined and cleared — rides in the same document + as non-failing results (``kind`` of ``pass`` / ``notApplicable`` / + ``open``, which SARIF defines precisely for "the rule ran and did not + fire"). Consumers that only want alerts filter on ``kind == "fail"`` + and are unaffected; consumers that need to distinguish "tested and + clean" from "never tested" now can. + * Whether the run itself completed is recorded on ``run.invocations``: + ``executionSuccessful`` plus a ``toolExecutionNotifications`` entry per + completeness caveat. A truncated run must not read as a clean one. """ from __future__ import annotations @@ -199,6 +208,7 @@ def build_sarif_report( *, tool_version: str | None = None, repository_context: dict[str, Any] | None = None, + coverage: dict[str, Any] | None = None, ) -> dict[str, Any]: """Return a SARIF 2.1.0 document for findings. @@ -209,6 +219,12 @@ def build_sarif_report( can bind alerts to the scanned commit; it is omitted for URL / IP (DAST) targets that have no repository. + ``coverage`` (optional) is the document from + :func:`strix.report.coverage.build_coverage_document`. Its cleared + surfaces become non-failing results and its completeness caveats become + invocation notifications, so a consumer can tell an untested area from a + tested-and-clean one without reading a second artifact. + Findings without safe source locations are anchored synthetically to SECURITY.md and flagged via ``properties.synthetic_location``. They're still emitted as proper SARIF results so they (a) flow @@ -247,6 +263,9 @@ def build_sarif_report( ) ) + if coverage: + _append_coverage(coverage, rules_by_id, rule_index_by_id, results) + driver: dict[str, Any] = { "name": TOOL_NAME, "informationUri": TOOL_INFORMATION_URI, @@ -260,6 +279,9 @@ def build_sarif_report( "results": results, } + if coverage: + run["invocations"] = [_coverage_invocation(coverage)] + run_properties: dict[str, Any] = {} if synthetic_location_count: # Surface the count for observability without duplicating the @@ -292,6 +314,7 @@ def write_sarif_report( *, tool_version: str | None = None, repository_context: dict[str, Any] | None = None, + coverage: dict[str, Any] | None = None, ) -> None: """Write a SARIF report to disk, creating parent directories first. @@ -304,6 +327,7 @@ def write_sarif_report( vulnerability_reports, tool_version=tool_version, repository_context=repository_context, + coverage=coverage, ) tmp_path = output_path.with_name(f"{output_path.name}.{os.getpid()}.tmp") try: @@ -321,6 +345,7 @@ def write_sarif( *, tool_version: str | None = None, repository_context: dict[str, Any] | None = None, + coverage: dict[str, Any] | None = None, filename: str = "findings.sarif", ) -> Path: """Write ``findings.sarif`` alongside existing outputs in ``run_dir``. @@ -335,6 +360,7 @@ def write_sarif( reports, tool_version=tool_version, repository_context=repository_context, + coverage=coverage, ) logger.info( "Wrote SARIF 2.1.0 report: %s (%d results)", @@ -526,6 +552,14 @@ def _result_properties( "impact", "technical_analysis", "remediation_steps", + # Calibration metadata. A downstream triager deciding whether to act + # on an alert needs the case against it and how firm the call is, not + # just the case for it. + "counterevidence", + "confidence", + "confidence_rationale", + "severity_change_conditions", + "fix_verification", ): value = report.get(key) if value not in (None, ""): @@ -613,6 +647,130 @@ def _build_fixes(report: dict[str, Any]) -> list[dict[str, Any]] | None: return [fix] +# --------------------------------------------------------------------------- +# Coverage +# --------------------------------------------------------------------------- + +_COVERAGE_RULE_PREFIX = "strix-coverage" + +# SARIF's ``result.kind`` already models every closure state the coverage +# ledger tracks, so no bespoke vocabulary is needed: ``pass`` is "checked and +# it holds", ``notApplicable`` is "the rule does not apply to this artifact", +# ``open`` is "a reviewer still has to decide". ``reported`` is absent on +# purpose — those surfaces are already in ``results`` as real ``fail`` +# findings, and emitting them twice would double-count them. +_OUTCOME_TO_KIND = { + "no_issue_found": "pass", + "ruled_out": "pass", + "not_applicable": "notApplicable", + "needs_follow_up": "open", +} + + +def _coverage_rule_id(risk_area: str) -> str: + slug = _slugify(risk_area) or "unspecified" + return f"{_COVERAGE_RULE_PREFIX}/{slug}" + + +def _build_coverage_rule(rule_id: str, risk_area: str) -> dict[str, Any]: + """A rule descriptor for a risk area that was assessed but did not fire.""" + description = f"Coverage of {risk_area} across the assessed attack surface." + return { + "id": rule_id, + "name": _rule_name(rule_id, risk_area), + "shortDescription": {"text": f"Coverage: {risk_area}"}, + "fullDescription": {"text": description}, + # Coverage results never raise an alert; the level lives on the + # result as ``none`` and the rule's default has to agree. + "defaultConfiguration": {"level": "none"}, + "help": {"text": description, "markdown": description}, + "properties": {"tags": ["coverage"]}, + } + + +def _build_coverage_result( + rule_id: str, + rule_index: int, + kind: str, + entry: dict[str, Any], +) -> dict[str, Any]: + """One non-failing result recording that a surface was assessed.""" + surface = _string_value(entry.get("surface")) or "unspecified surface" + risk_area = _string_value(entry.get("risk_area")) or "unspecified risk" + evidence = _string_value(entry.get("evidence")) + label = _string_value(entry.get("outcome_label")) or str(entry.get("outcome", "")) + + message = f"{risk_area} — {label}: {surface}" + if evidence: + message = f"{message}\n\n{evidence}" + + result: dict[str, Any] = { + "ruleId": rule_id, + "ruleIndex": rule_index, + "kind": kind, + # SARIF requires ``level: none`` for any result whose kind is not + # ``fail``; anything else makes the document invalid. + "level": "none", + "message": {"text": message}, + "locations": [{"logicalLocations": [{"fullyQualifiedName": surface}]}], + "properties": { + "strix": { + "coverage_outcome": entry.get("outcome", ""), + "risk_area": risk_area, + "surface": surface, + "recorded_by": entry.get("recorded_by", ""), + "source": entry.get("source", "agent_reported"), + } + }, + } + return result + + +def _append_coverage( + coverage: dict[str, Any], + rules_by_id: dict[str, dict[str, Any]], + rule_index_by_id: dict[str, int], + results: list[dict[str, Any]], +) -> None: + """Add the coverage ledger's cleared surfaces to an in-progress run.""" + entries = coverage.get("entries") + if not isinstance(entries, list): + return + for entry in entries: + if not isinstance(entry, dict): + continue + kind = _OUTCOME_TO_KIND.get(str(entry.get("outcome", ""))) + if kind is None: + continue + rule_id = _coverage_rule_id(str(entry.get("risk_area", ""))) + if rule_id not in rules_by_id: + rule_index_by_id[rule_id] = len(rules_by_id) + rules_by_id[rule_id] = _build_coverage_rule( + rule_id, _string_value(entry.get("risk_area")) or "unspecified risk" + ) + results.append(_build_coverage_result(rule_id, rule_index_by_id[rule_id], kind, entry)) + + +def _coverage_invocation(coverage: dict[str, Any]) -> dict[str, Any]: + """Record whether the run was complete enough for its results to be final. + + ``executionSuccessful: false`` is the standard signal that a consumer must + not read "no results" as "nothing to find" — a scan cut short by its + budget produces a document that otherwise looks identical to a clean one. + """ + completeness = coverage.get("completeness") + completeness = completeness if isinstance(completeness, dict) else {} + caveats = completeness.get("caveats") + caveats = caveats if isinstance(caveats, list) else [] + + invocation: dict[str, Any] = {"executionSuccessful": bool(completeness.get("complete", True))} + if caveats: + invocation["toolExecutionNotifications"] = [ + {"level": "warning", "message": {"text": str(caveat)}} for caveat in caveats + ] + return invocation + + # --------------------------------------------------------------------------- # Location handling # --------------------------------------------------------------------------- diff --git a/strix/report/state.py b/strix/report/state.py index 4601ede6..df37b520 100644 --- a/strix/report/state.py +++ b/strix/report/state.py @@ -13,7 +13,8 @@ from agents.usage import Usage from strix.config import codex from strix.config.loader import load_settings -from strix.core.paths import run_dir_for +from strix.core.paths import run_dir_for, runtime_state_dir +from strix.report.coverage import render_coverage_markdown, write_coverage from strix.report.sarif import write_sarif from strix.report.usage import LLMUsageLedger from strix.report.writer import ( @@ -432,14 +433,47 @@ class ReportState: {str(scan_results.get("recommendations", "")).strip()} """ + def _coverage_document(self) -> dict[str, Any] | None: + """Assemble the coverage record, or None when it can't be built. + + Coverage is a secondary artifact: a failure here must not cost the + caller its findings, so this swallows and logs rather than raising + into :meth:`_save_artifacts`. + """ + try: + from strix.report.coverage import build_coverage_document, read_agent_graph + from strix.tools.coverage.tools import get_coverage_entries + + return build_coverage_document( + run_record=self.run_record, + entries=get_coverage_entries(), + agent_graph=read_agent_graph(runtime_state_dir(self.get_run_dir())), + vulnerability_reports=self.vulnerability_reports, + exit_reason=self.scan_ended_exit_reason, + ) + except Exception: + logger.exception("coverage document build failed (non-fatal)") + return None + def _save_artifacts(self) -> None: """Write scan artifacts under ``run_dir``.""" run_dir = self.get_run_dir() try: run_dir.mkdir(parents=True, exist_ok=True) + coverage = self._coverage_document() + if coverage is not None: + try: + write_coverage(run_dir, coverage) + except OSError: + logger.exception("coverage.json write failed (non-fatal)") + if self.final_scan_result: - write_executive_report(run_dir, self.final_scan_result) + write_executive_report( + run_dir, + self.final_scan_result, + render_coverage_markdown(coverage) if coverage else None, + ) if self.vulnerability_reports: write_vulnerabilities(run_dir, self.vulnerability_reports, self._saved_vuln_ids) @@ -456,6 +490,7 @@ class ReportState: self.vulnerability_reports, tool_version=_strix_version(), repository_context=self._sarif_repository_context(), + coverage=coverage, ) except Exception: logger.exception("SARIF emit failed (non-fatal; CSV/MD unaffected)") diff --git a/strix/report/writer.py b/strix/report/writer.py index ec592f14..3447bea6 100644 --- a/strix/report/writer.py +++ b/strix/report/writer.py @@ -107,18 +107,32 @@ def read_run_record(run_dir: Path) -> dict[str, Any]: def write_run_record(run_dir: Path, run_record: dict[str, Any]) -> None: - _atomic_write_text( + atomic_write_text( run_record_path(run_dir), json.dumps(run_record, ensure_ascii=False, indent=2, default=str), ) -def write_executive_report(run_dir: Path, final_scan_result: str) -> None: +def write_executive_report( + run_dir: Path, + final_scan_result: str, + coverage_markdown: str | None = None, +) -> None: + """Write the client-facing report, optionally with a coverage appendix. + + ``coverage_markdown`` is rendered from the coverage ledger rather than + authored by an agent, so what the report claims was examined stays tied to + what was actually recorded. + """ path = run_dir / "penetration_test_report.md" - with path.open("w", encoding="utf-8") as f: - f.write("# Security Penetration Test Report\n\n") - f.write(f"**Generated:** {datetime.now(UTC).strftime('%Y-%m-%d %H:%M:%S UTC')}\n\n") - f.write(f"{final_scan_result}\n") + sections = [ + "# Security Penetration Test Report\n", + f"**Generated:** {datetime.now(UTC).strftime('%Y-%m-%d %H:%M:%S UTC')}\n", + f"{final_scan_result}\n", + ] + if coverage_markdown: + sections.append(f"{coverage_markdown.rstrip()}\n") + atomic_write_text(path, "\n".join(sections)) logger.info("Saved final penetration test report to: %s", path) @@ -133,7 +147,7 @@ def write_vulnerabilities( new_reports = [r for r in vulnerability_reports if r["id"] not in saved_vuln_ids] for report in new_reports: - _atomic_write_text( + atomic_write_text( vuln_dir / f"{report['id']}.md", render_vulnerability_md(report), ) @@ -158,9 +172,9 @@ def write_vulnerabilities( "file": f"vulnerabilities/{report['id']}.md", }, ) - _atomic_write_text(csv_path, csv_buf.getvalue()) + atomic_write_text(csv_path, csv_buf.getvalue()) - _atomic_write_text( + atomic_write_text( run_dir / "vulnerabilities.json", json.dumps(vulnerability_reports, ensure_ascii=False, indent=2, default=str), ) @@ -175,7 +189,8 @@ def write_vulnerabilities( return len(new_reports) -def _atomic_write_text(path: Path, payload: str) -> None: +def atomic_write_text(path: Path, payload: str) -> None: + """Write *payload* to *path* via a sibling temp file and an atomic rename.""" path.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile( mode="w", @@ -215,6 +230,8 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL cvss = report.get("cvss") if cvss is not None: metadata.append(("CVSS", cvss)) + if report.get("confidence"): + metadata.append(("Confidence", str(report["confidence"]).title())) if report.get("fix_effort"): metadata.append(("Fix Effort", str(report["fix_effort"]).title())) for label, value in metadata: @@ -236,6 +253,21 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL lines.append(str(report["impact"])) lines.append("") + if report.get("counterevidence"): + lines.append("## Counterevidence\n") + lines.append(str(report["counterevidence"])) + lines.append("") + + if report.get("confidence_rationale"): + lines.append("## Confidence Rationale\n") + lines.append(str(report["confidence_rationale"])) + lines.append("") + + if report.get("severity_change_conditions"): + lines.append("## What Would Change This Severity\n") + lines.append(str(report["severity_change_conditions"])) + lines.append("") + if report.get("technical_analysis"): lines.append("## Technical Analysis\n") lines.append(str(report["technical_analysis"])) @@ -289,6 +321,11 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL lines.append(str(report["remediation_steps"])) lines.append("") + if report.get("fix_verification"): + lines.append("## Fix Verification\n") + lines.append(str(report["fix_verification"])) + lines.append("") + if report.get("assumptions"): lines.append("## Assumptions\n") lines.append(str(report["assumptions"])) diff --git a/strix/tools/coverage/tools.py b/strix/tools/coverage/tools.py index b38b4fea..94c31fb9 100644 --- a/strix/tools/coverage/tools.py +++ b/strix/tools/coverage/tools.py @@ -4,6 +4,13 @@ Findings answer "what did we find". Coverage answers "what did we look at, and how did each one close" — the negative space a client report needs in order to be trustworthy. Every agent records the surfaces it reviewed; the root agent reconciles them at the end of the scan. + +Entries here are **agent-reported**: an agent's own account of what it +assessed. ``strix.report.coverage`` pairs them with machine-observed facts +(which agents ran, which skills they carried, how the run terminated) and +labels the provenance of each, so a reader can tell a self-report from an +observation. The runtime mirror under ``{state_dir}`` exists for resume; the +client-facing artifact is ``{run_dir}/coverage.json``. """ from __future__ import annotations @@ -96,13 +103,19 @@ def hydrate_coverage_from_disk(state_dir: Path) -> None: ) -def _persist() -> None: +def _persist_locked() -> None: + """Mirror the ledger to disk. Callers must already hold ``_coverage_lock``. + + Serialization and the rename happen in one critical section. Releasing + the lock in between would let a writer holding an older serialization win + the rename and silently roll back a concurrent agent's entry, so the + ledger would hydrate short on resume. + """ path = _coverage_path if path is None: return try: - with _coverage_lock: - payload = json.dumps(_coverage_storage, ensure_ascii=False, default=str) + payload = json.dumps(_coverage_storage, ensure_ascii=False, default=str) path.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile( mode="w", @@ -155,17 +168,23 @@ def _validate( return normalized, errors -def _duplicate_of(surface: str, risk_area: str) -> tuple[str, dict[str, Any]] | None: - """Find an existing row for this exact surface and risk area.""" +def _duplicate_of_locked(surface: str, risk_area: str) -> tuple[str, dict[str, Any]] | None: + """Find an existing row for this exact surface and risk area. + + Callers must already hold ``_coverage_lock``. The uniqueness check and the + insertion that depends on it have to be one critical section: otherwise + two agents recording the same surface concurrently both see "no + duplicate", and the ledger ends up with exactly the parallel rows this + rejection exists to prevent. + """ key = (surface.strip().lower(), risk_area.strip().lower()) - with _coverage_lock: - for entry_id, entry in _coverage_storage.items(): - existing = ( - str(entry.get("surface", "")).strip().lower(), - str(entry.get("risk_area", "")).strip().lower(), - ) - if existing == key: - return entry_id, dict(entry) + for entry_id, entry in _coverage_storage.items(): + existing = ( + str(entry.get("surface", "")).strip().lower(), + str(entry.get("risk_area", "")).strip().lower(), + ) + if existing == key: + return entry_id, dict(entry) return None @@ -184,26 +203,6 @@ def _record_impl( if errors: return {"success": False, "error": "Validation failed", "errors": errors} - duplicate = _duplicate_of(surface, risk_area) - if duplicate is not None: - existing_id, existing = duplicate - owner = existing.get("agent_name") or "another agent" - return { - "success": False, - "error": ( - f"'{surface.strip()}' ({risk_area.strip()}) already has coverage entry " - f"{existing_id}, recorded by {owner} as " - f"'{existing.get('outcome', '')}'. Two rows for one surface leave the " - "report showing a stale conclusion beside its replacement. If your " - "review reached a different conclusion, move that entry with " - f"update_coverage(entry_id='{existing_id}', ...) and say in evidence " - "what changed. If you reviewed something genuinely different, name the " - "surface or risk area more precisely and record it again." - ), - "existing_entry_id": existing_id, - "existing_outcome": existing.get("outcome", ""), - } - entry: dict[str, Any] = { "surface": surface.strip(), "risk_area": risk_area.strip(), @@ -218,11 +217,31 @@ def _record_impl( entry["agent_name"] = agent_name with _coverage_lock: + duplicate = _duplicate_of_locked(surface, risk_area) + if duplicate is not None: + existing_id, existing = duplicate + owner = existing.get("agent_name") or "another agent" + return { + "success": False, + "error": ( + f"'{surface.strip()}' ({risk_area.strip()}) already has coverage entry " + f"{existing_id}, recorded by {owner} as " + f"'{existing.get('outcome', '')}'. Two rows for one surface leave the " + "report showing a stale conclusion beside its replacement. If your " + "review reached a different conclusion, move that entry with " + f"update_coverage(entry_id='{existing_id}', ...) and say in evidence " + "what changed. If you reviewed something genuinely different, name the " + "surface or risk area more precisely and record it again." + ), + "existing_entry_id": existing_id, + "existing_outcome": existing.get("outcome", ""), + } + entry_id = _generate_entry_id() if entry_id is None: return {"success": False, "error": "Could not allocate a coverage entry id"} _coverage_storage[entry_id] = entry - _persist() + _persist_locked() logger.info( "Coverage recorded: id=%s outcome=%s surface=%s", entry_id, @@ -284,7 +303,7 @@ def _update_impl( existing["agent_id"] = agent_id if agent_name: existing["agent_name"] = agent_name - _persist() + _persist_locked() logger.info( "Coverage updated: id=%s %s -> %s surface=%s", key, diff --git a/strix/tools/finish/tool.py b/strix/tools/finish/tool.py index 38e6e56d..8c9ba4c0 100644 --- a/strix/tools/finish/tool.py +++ b/strix/tools/finish/tool.py @@ -22,6 +22,7 @@ def _do_finish( methodology: str, technical_analysis: str, recommendations: str, + agent_graph: dict[str, Any], ) -> dict[str, Any]: if parent_id is not None: return { @@ -63,7 +64,7 @@ def _do_finish( recommendations=recommendations.strip(), ) vuln_count = len(report_state.vulnerability_reports) - coverage_summary = _coverage_summary() + coverage_summary = _coverage_summary(agent_graph) except (ImportError, AttributeError) as e: logger.exception("finish_scan persistence failed") return {"success": False, "error": f"Failed to complete scan: {e!s}"} @@ -82,8 +83,17 @@ def _do_finish( return result -def _coverage_summary() -> dict[str, Any]: - """Coverage counts plus a warning when surfaces were left unresolved.""" +def _coverage_summary(agent_graph: dict[str, Any]) -> dict[str, Any]: + """Coverage counts, unresolved surfaces, and gaps the runtime can see. + + The gap list is derived from the agent graph rather than from the ledger, + so it catches the failure the ledger cannot: a risk class an agent was + equipped for and never accounted for. Surfacing it here — in the response + to the call that ends the scan — is the last point at which the root agent + can still dispatch work or record the class as unresolved instead of + letting the report imply it was clean. + """ + from strix.report.coverage import agents_from_graph, skill_coverage_gaps from strix.tools.coverage.tools import get_coverage_entries, outcome_counts entries = get_coverage_entries() @@ -113,6 +123,15 @@ def _coverage_summary() -> dict[str, Any]: {"surface": e.get("surface", ""), "risk_area": e.get("risk_area", "")} for e in unresolved ] + + gaps = skill_coverage_gaps(entries, agents_from_graph(agent_graph)) + if gaps: + summary["coverage_gaps"] = [gap["detail"] for gap in gaps] + summary["coverage_gap_warning"] = ( + f"{len(gaps)} risk class(es) assigned to agents have no coverage entry and " + "will be published as unexamined. Record them (or a needs_follow_up row) " + "before the report goes out." + ) return summary @@ -210,15 +229,13 @@ async def finish_scan( - ``methodology`` — frameworks followed (OWASP WSTG, PTES, OSSTMM, NIST), engagement type (black/gray/white box), scope and constraints, categories of testing performed. **No** - internal execution detail. End this section with a - **Reviewed Surfaces** markdown table built from - ``list_coverage`` — columns ``Surface`` | ``Risk Area`` | - ``Outcome`` | ``Notes`` — so the reader can see what was - examined and cleared, not only what was found. Render - outcomes in client-facing language (``Finding reported``, - ``No issue identified``, ``Not applicable``, ``Requires - further review``). If any surface requires further review, - call that out explicitly beneath the table. + internal execution detail. Do **not** hand-write a table of + reviewed surfaces here: the report gains a ``Coverage`` + section rendered directly from the coverage ledger, and a + transcribed copy would only drift from it. Describe the + approach and the classes of testing, and reconcile the ledger + via ``list_coverage`` (step 5) so that rendered section is + complete and honest. - ``technical_analysis`` — consolidated findings overview with severity model and systemic root causes. Reference individual vuln reports for repro steps; don't duplicate raw evidence. @@ -333,6 +350,7 @@ async def finish_scan( methodology=methodology, technical_analysis=technical_analysis, recommendations=recommendations, + agent_graph=await coordinator.snapshot() if coordinator is not None else {}, ) if ( result.get("success") diff --git a/strix/tools/threat_model/tools.py b/strix/tools/threat_model/tools.py index d5c7a7b1..ec5975c5 100644 --- a/strix/tools/threat_model/tools.py +++ b/strix/tools/threat_model/tools.py @@ -104,21 +104,53 @@ def _normalize_remote_target(target: str) -> str: return f"{authority}{path}" +def _normalize_git_remote(remote: str) -> str: + """Collapse a git remote URL onto the same key its clone URL would produce. + + A remote reaches us in whichever spelling the clone used — + ``git@github.com:org/repo.git``, ``https://github.com/org/repo``, + ``ssh://git@github.com/org/repo.git`` — and each is the same repository. + Rewriting scp-style syntax into a URL and dropping the ``.git`` suffix and + any embedded credentials lets :func:`_normalize_remote_target` produce one + identity for all of them, and crucially the *same* identity a caller gets + when it names the repository by its remote URL rather than by a checkout + path. Without that, the model saved by an agent working in the checkout is + invisible to an agent that asks for the repository by URL, and the two + derive conflicting models of one target. + """ + candidate = remote.strip() + scp_style = re.match(r"^(?:[^@/]+@)?(?P[^:/]+):(?P.+)$", candidate) + if scp_style and "://" not in candidate: + candidate = f"https://{scp_style['host']}/{scp_style['path'].lstrip('/')}" + elif "://" in candidate: + # The transport a clone happened to use says nothing about which + # repository this is, and each scheme carries a different default + # port into the authority. Collapsing them all onto https keeps one + # repository on one key however it was cloned. + candidate = f"https://{candidate.split('://', 1)[1]}" + normalized = _normalize_remote_target(candidate) + return normalized.removesuffix(".git") + + def _target_identity(target: str) -> tuple[str, str]: """Return the (stable identity, revision) pair a cached model is keyed on. A checkout is keyed on its remote (so the same repository cloned to two paths shares one model, and a subdirectory resolves to the whole tree) and pinned to ``HEAD``. Everything else — a host, a URL, an API base, a named - scope — is keyed on its normalized form and carries no revision. + scope — is keyed on its normalized form and carries no revision. Both + routes run through the same normalization, so a checkout and the URL it + was cloned from land on one key. """ directory = _local_directory(target) if directory is None: - return _normalize_remote_target(target), _UNVERSIONED + return _normalize_remote_target(target).removesuffix(".git"), _UNVERSIONED remote = _git(directory, ["config", "--get", "remote.origin.url"]) revision = _git(directory, ["rev-parse", "HEAD"]) or _UNVERSIONED + if remote: + return _normalize_git_remote(remote), revision toplevel = _git(directory, ["rev-parse", "--show-toplevel"]) - return remote or toplevel or str(directory), revision + return toplevel or str(directory), revision def _cache_path(identity: str) -> Path: diff --git a/tests/test_coverage_tool.py b/tests/test_coverage_tool.py index 6191c2a1..3061dd89 100644 --- a/tests/test_coverage_tool.py +++ b/tests/test_coverage_tool.py @@ -2,6 +2,9 @@ from __future__ import annotations +import json +import threading +from concurrent.futures import ThreadPoolExecutor from typing import TYPE_CHECKING, Any import pytest @@ -237,3 +240,45 @@ def test_a_different_risk_area_on_one_surface_is_still_its_own_entry() -> None: assert second["success"] is True assert len(get_coverage_entries()) == 2 + + +def test_concurrent_records_of_one_surface_yield_a_single_row() -> None: + """Duplicate detection and insertion must be one critical section. + + Two agents recording the same surface at the same moment would otherwise + both pass the "no duplicate" check, and the report would show a stale + conclusion beside its replacement — the exact outcome the rejection exists + to prevent. + """ + barrier = threading.Barrier(8) + + def attempt(index: int) -> dict[str, Any]: + barrier.wait() + return _record(agent_id=f"agent-{index}", agent_name=f"tester-{index}") + + with ThreadPoolExecutor(max_workers=8) as pool: + results = list(pool.map(attempt, range(8))) + + assert sum(1 for result in results if result["success"]) == 1 + assert len(get_coverage_entries()) == 1 + + +def test_concurrent_records_all_survive_persistence(coverage_store: Path) -> None: + """A writer holding an older snapshot must not win the rename. + + If it did, the mirror would come back short on resume and coverage + recorded before a crash would silently disappear from the report. + """ + barrier = threading.Barrier(8) + + def attempt(index: int) -> dict[str, Any]: + barrier.wait() + return _record(surface=f"GET /api/resource/{index}", agent_id=f"agent-{index}") + + with ThreadPoolExecutor(max_workers=8) as pool: + list(pool.map(attempt, range(8))) + + persisted = json.loads((coverage_store / "coverage.json").read_text(encoding="utf-8")) + assert len(persisted) == 8 + hydrate_coverage_from_disk(coverage_store) + assert len(get_coverage_entries()) == 8 diff --git a/tests/test_finish_coverage_gate.py b/tests/test_finish_coverage_gate.py new file mode 100644 index 00000000..52a19a7d --- /dev/null +++ b/tests/test_finish_coverage_gate.py @@ -0,0 +1,65 @@ +"""finish_scan confronts the root agent with the coverage the runtime can see.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +from strix.tools.coverage.tools import _record_impl, hydrate_coverage_from_disk +from strix.tools.finish.tool import _coverage_summary + + +if TYPE_CHECKING: + from pathlib import Path + + +_GRAPH = { + "statuses": {"agent-1": "completed"}, + "names": {"agent-1": "injection-tester"}, + "metadata": {"agent-1": {"skills": ["sql_injection", "xss"]}}, +} + + +@pytest.fixture(autouse=True) +def _empty_ledger(tmp_path: Path) -> None: + hydrate_coverage_from_disk(tmp_path) + + +def _record(risk_area: str) -> None: + _record_impl( + surface="POST /api/orders/{id}", + risk_area=risk_area, + outcome="no_issue_found", + evidence="Parameters fuzzed; no anomalies.", + agent_id="agent-1", + agent_name="injection-tester", + ) + + +def test_unrecorded_risk_class_is_reported_back_to_the_root_agent() -> None: + _record("SQL injection") + + summary = _coverage_summary(_GRAPH) + + assert summary["coverage_recorded"] == 1 + assert len(summary["coverage_gaps"]) == 1 + assert "xss" in summary["coverage_gaps"][0] + assert "unexamined" in summary["coverage_gap_warning"] + + +def test_fully_accounted_coverage_raises_no_gap_warning() -> None: + _record("SQL injection") + _record("cross-site scripting") + + summary = _coverage_summary(_GRAPH) + + assert "coverage_gaps" not in summary + assert "coverage_gap_warning" not in summary + + +def test_an_empty_ledger_still_warns_first() -> None: + summary = _coverage_summary(_GRAPH) + + assert summary["coverage_recorded"] == 0 + assert "No coverage was recorded" in summary["coverage_warning"] diff --git a/tests/test_report_coverage.py b/tests/test_report_coverage.py new file mode 100644 index 00000000..918daf70 --- /dev/null +++ b/tests/test_report_coverage.py @@ -0,0 +1,203 @@ +"""Tests for the coverage artifact assembled in strix.report.coverage.""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING, Any + +from strix.report.coverage import ( + build_coverage_document, + read_agent_graph, + render_coverage_markdown, + write_coverage, +) + + +if TYPE_CHECKING: + from pathlib import Path + + +def _entry(**overrides: Any) -> dict[str, Any]: + base: dict[str, Any] = { + "surface": "POST /api/orders/{id}", + "risk_area": "object-level authorization", + "outcome": "no_issue_found", + "evidence": "Two tenants tested; both received 403.", + "agent_id": "agent-1", + "agent_name": "authz-tester", + "created_at": "2026-07-02 10:00:00 UTC", + } + base.update(overrides) + return base + + +def _graph(**overrides: Any) -> dict[str, Any]: + base: dict[str, Any] = { + "statuses": {"agent-1": "completed"}, + "names": {"agent-1": "authz-tester"}, + "metadata": {"agent-1": {"skills": ["idor"], "task": "authz review"}}, + } + base.update(overrides) + return base + + +def _document(**overrides: Any) -> dict[str, Any]: + kwargs: dict[str, Any] = { + "run_record": {"run_id": "r1", "run_name": "run-1", "status": "completed"}, + "entries": [_entry()], + "agent_graph": _graph(), + "vulnerability_reports": [], + } + kwargs.update(overrides) + return build_coverage_document(**kwargs) + + +def test_document_reports_surfaces_and_outcomes() -> None: + doc = _document() + + assert doc["summary"]["surfaces_reviewed"] == 1 + assert doc["summary"]["outcomes"] == {"no_issue_found": 1} + assert doc["entries"][0]["outcome_label"] == "No issue identified" + assert doc["entries"][0]["recorded_by"] == "authz-tester" + + +def test_ledger_entries_are_labelled_as_agent_reported() -> None: + """A reader has to be able to tell a self-report from an observation.""" + doc = _document() + + assert doc["entries"][0]["source"] == "agent_reported" + assert doc["machine_observed"]["source"] == "runtime" + assert doc["machine_observed"]["skills_exercised"] == ["idor"] + + +def test_assigned_risk_skill_without_coverage_becomes_a_gap() -> None: + """An agent carrying the sql_injection skill that records nothing about it + leaves the class unexamined, not clean.""" + doc = _document( + agent_graph=_graph( + metadata={"agent-1": {"skills": ["idor", "sql_injection"], "task": "review"}} + ) + ) + + gaps = [gap for gap in doc["gaps"] if gap["kind"] == "unrecorded_risk_class"] + assert [gap["risk_area"] for gap in gaps] == ["sql injection"] + + +def test_recorded_risk_class_is_not_reported_as_a_gap() -> None: + doc = _document( + entries=[_entry(risk_area="SQL injection", surface="GET /search?q=")], + agent_graph=_graph(metadata={"agent-1": {"skills": ["sql_injection"]}}), + ) + + assert not [gap for gap in doc["gaps"] if gap["kind"] == "unrecorded_risk_class"] + + +def test_synonym_phrasing_counts_as_recorded_coverage() -> None: + """The ledger says "object-level authorization"; the skill is called idor.""" + doc = _document(agent_graph=_graph(metadata={"agent-1": {"skills": ["idor"]}})) + + assert not [gap for gap in doc["gaps"] if gap["kind"] == "unrecorded_risk_class"] + + +def test_non_risk_skills_carry_no_coverage_obligation() -> None: + """Tooling skills describe how an agent works, not what it hunts.""" + doc = _document(agent_graph=_graph(metadata={"agent-1": {"skills": ["idor", "caido"]}})) + + assert not [gap for gap in doc["gaps"] if gap.get("risk_area") == "caido"] + + +def test_agent_that_recorded_nothing_is_a_gap() -> None: + doc = _document( + agent_graph=_graph( + statuses={"agent-1": "completed", "agent-2": "completed"}, + names={"agent-1": "authz-tester", "agent-2": "recon"}, + metadata={}, + ) + ) + + silent = [gap for gap in doc["gaps"] if gap["kind"] == "agent_recorded_no_coverage"] + assert [gap["agent_name"] for gap in silent] == ["recon"] + + +def test_needs_follow_up_is_carried_as_an_open_gap() -> None: + doc = _document( + entries=[_entry(outcome="needs_follow_up", evidence="Auth wall blocked testing.")] + ) + + assert doc["gaps"][0]["kind"] == "needs_follow_up" + assert doc["gaps"][0]["detail"] == "Auth wall blocked testing." + + +def test_completed_run_with_finished_agents_is_complete() -> None: + doc = _document(exit_reason="finished_by_tool") + + assert doc["completeness"]["complete"] is True + assert doc["completeness"]["caveats"] == [] + + +def test_budget_exhausted_run_is_not_a_complete_record() -> None: + """A truncated scan must not read like a clean one.""" + doc = _document(exit_reason="budget_exhausted") + + assert doc["completeness"]["complete"] is False + assert "budget_exhausted" in doc["completeness"]["caveats"][0] + + +def test_unfinished_agent_makes_the_record_partial() -> None: + doc = _document( + agent_graph=_graph(statuses={"agent-1": "crashed"}), + exit_reason="finished_by_tool", + ) + + assert doc["completeness"]["complete"] is False + assert "authz-tester" in doc["completeness"]["caveats"][0] + + +def test_failed_run_status_makes_the_record_partial() -> None: + doc = _document( + run_record={"run_id": "r1", "status": "failed"}, + exit_reason="finished_by_tool", + ) + + assert doc["completeness"]["complete"] is False + + +def test_write_coverage_emits_a_top_level_artifact(tmp_path: Path) -> None: + path = write_coverage(tmp_path, _document()) + + assert path == tmp_path / "coverage.json" + assert json.loads(path.read_text(encoding="utf-8"))["schema_version"] == 1 + + +def test_markdown_renders_a_surface_table() -> None: + markdown = render_coverage_markdown(_document()) + + assert "# Coverage" in markdown + assert "| POST /api/orders/{id} | object-level authorization | No issue identified |" in ( + markdown + ) + + +def test_markdown_says_so_when_nothing_was_recorded() -> None: + markdown = render_coverage_markdown(_document(entries=[], agent_graph={})) + + assert "cannot be read as evidence" in markdown + + +def test_markdown_cell_escapes_pipes() -> None: + markdown = render_coverage_markdown(_document(entries=[_entry(surface="a|b")])) + + assert "| a\\|b |" in markdown + + +def test_read_agent_graph_tolerates_a_missing_or_corrupt_snapshot(tmp_path: Path) -> None: + assert read_agent_graph(tmp_path) == {} + + (tmp_path / "agents.json").write_text("{not json", encoding="utf-8") + assert read_agent_graph(tmp_path) == {} + + +def test_read_agent_graph_loads_a_snapshot(tmp_path: Path) -> None: + (tmp_path / "agents.json").write_text(json.dumps(_graph()), encoding="utf-8") + + assert read_agent_graph(tmp_path)["names"] == {"agent-1": "authz-tester"} diff --git a/tests/test_report_writer.py b/tests/test_report_writer.py index 05222796..74d03ed6 100644 --- a/tests/test_report_writer.py +++ b/tests/test_report_writer.py @@ -179,3 +179,38 @@ def test_write_executive_report_writes_markdown(tmp_path: Path) -> None: content = (tmp_path / "penetration_test_report.md").read_text(encoding="utf-8") assert "# Security Penetration Test Report" in content assert "Scan complete. No critical issues." in content + + +def test_render_vulnerability_md_surfaces_calibration_metadata() -> None: + """Confidence, the case against the finding, and retest status are part of + the deliverable — storing them without rendering hides the reasoning.""" + md = render_vulnerability_md( + { + "id": "vuln-0009", + "title": "SSRF in URL preview", + "severity": "high", + "timestamp": "2026-07-02 10:00:00 UTC", + "description": "Fetches user-supplied URLs.", + "confidence": "medium", + "counterevidence": "Egress appears filtered at the network layer.", + "confidence_rationale": "Reproduced once out of three attempts.", + "severity_change_conditions": "Critical if egress filtering is removed.", + "remediation_steps": "Allowlist destinations.", + "fix_verification": "Not retested.", + } + ) + + assert "**Confidence:** Medium" in md + assert "## Counterevidence" in md + assert "Egress appears filtered at the network layer." in md + assert "## Confidence Rationale" in md + assert "## What Would Change This Severity" in md + assert "## Fix Verification" in md + + +def test_write_executive_report_appends_the_coverage_section(tmp_path: Path) -> None: + write_executive_report(tmp_path, "Scan complete.", "# Coverage\n\nNothing tested.\n") + content = (tmp_path / "penetration_test_report.md").read_text(encoding="utf-8") + + assert content.index("Scan complete.") < content.index("# Coverage") + assert "Nothing tested." in content diff --git a/tests/test_sarif.py b/tests/test_sarif.py index 849835b0..61ffc39b 100644 --- a/tests/test_sarif.py +++ b/tests/test_sarif.py @@ -242,3 +242,132 @@ def test_write_sarif_replaces_atomically_no_partial_on_reemit(tmp_path: Path) -> assert leftovers == [] # And it parses as a complete document with both findings. assert len(_read(tmp_path)["runs"][0]["results"]) == 2 + + +def _coverage(*entries: dict[str, Any], **overrides: Any) -> dict[str, Any]: + doc: dict[str, Any] = { + "entries": list(entries), + "completeness": {"complete": True, "caveats": []}, + } + doc.update(overrides) + return doc + + +def _coverage_entry(**overrides: Any) -> dict[str, Any]: + base: dict[str, Any] = { + "surface": "POST /api/orders/{id}", + "risk_area": "SQL injection", + "outcome": "no_issue_found", + "outcome_label": "No issue identified", + "evidence": "14 parameters fuzzed; all queries parameterized.", + "recorded_by": "injection-tester", + "source": "agent_reported", + } + base.update(overrides) + return base + + +def test_cleared_surface_becomes_a_passing_result(tmp_path: Path) -> None: + """ "Tested and clean" is a SARIF pass, not an absent result.""" + write_sarif(tmp_path, [], coverage=_coverage(_coverage_entry())) + results = _read(tmp_path)["runs"][0]["results"] + + assert len(results) == 1 + assert results[0]["kind"] == "pass" + # SARIF requires level "none" on any result that is not a failure. + assert results[0]["level"] == "none" + assert "14 parameters fuzzed" in results[0]["message"]["text"] + + +def test_coverage_outcomes_map_to_their_sarif_kinds(tmp_path: Path) -> None: + write_sarif( + tmp_path, + [], + coverage=_coverage( + _coverage_entry(outcome="ruled_out", risk_area="XSS"), + _coverage_entry(outcome="not_applicable", risk_area="XXE"), + _coverage_entry(outcome="needs_follow_up", risk_area="SSRF"), + ), + ) + kinds = [result["kind"] for result in _read(tmp_path)["runs"][0]["results"]] + + assert kinds == ["pass", "notApplicable", "open"] + + +def test_reported_coverage_is_not_duplicated_as_a_pass(tmp_path: Path) -> None: + """A surface that produced a finding is already in results as a failure.""" + write_sarif( + tmp_path, + [_finding()], + coverage=_coverage(_coverage_entry(outcome="reported")), + ) + results = _read(tmp_path)["runs"][0]["results"] + + assert len(results) == 1 + assert results[0].get("kind", "fail") == "fail" + + +def test_coverage_results_declare_their_own_rules(tmp_path: Path) -> None: + write_sarif( + tmp_path, + [_finding()], + coverage=_coverage( + _coverage_entry(risk_area="SQL injection"), + _coverage_entry(risk_area="SQL injection", surface="GET /search"), + ), + ) + run = _read(tmp_path)["runs"][0] + rules = run["tool"]["driver"]["rules"] + coverage_rules = [rule for rule in rules if rule["id"].startswith("strix-coverage/")] + + # Both entries share one rule, and every result's ruleIndex resolves to it. + assert len(coverage_rules) == 1 + assert coverage_rules[0]["defaultConfiguration"]["level"] == "none" + for result in run["results"]: + assert rules[result["ruleIndex"]]["id"] == result["ruleId"] + + +def test_incomplete_run_is_flagged_on_the_invocation(tmp_path: Path) -> None: + """A scan cut short must not be indistinguishable from a clean one.""" + write_sarif( + tmp_path, + [], + coverage=_coverage( + _coverage_entry(), + completeness={"complete": False, "caveats": ["Budget exhausted."]}, + ), + ) + invocation = _read(tmp_path)["runs"][0]["invocations"][0] + + assert invocation["executionSuccessful"] is False + assert invocation["toolExecutionNotifications"][0]["message"]["text"] == "Budget exhausted." + + +def test_complete_run_reports_a_successful_invocation(tmp_path: Path) -> None: + write_sarif(tmp_path, [], coverage=_coverage(_coverage_entry())) + invocation = _read(tmp_path)["runs"][0]["invocations"][0] + + assert invocation["executionSuccessful"] is True + assert "toolExecutionNotifications" not in invocation + + +def test_calibration_metadata_survives_into_result_properties(tmp_path: Path) -> None: + write_sarif( + tmp_path, + [ + _finding( + confidence="medium", + counterevidence="WAF blocks the naive payload.", + confidence_rationale="Reproduced once out of three attempts.", + severity_change_conditions="Critical if the WAF rule is removed.", + fix_verification="Not retested.", + ) + ], + ) + strix = _read(tmp_path)["runs"][0]["results"][0]["properties"]["strix"] + + assert strix["confidence"] == "medium" + assert strix["counterevidence"] == "WAF blocks the naive payload." + assert strix["confidence_rationale"] == "Reproduced once out of three attempts." + assert strix["severity_change_conditions"] == "Critical if the WAF rule is removed." + assert strix["fix_verification"] == "Not retested." diff --git a/tests/test_state_coverage_artifact.py b/tests/test_state_coverage_artifact.py new file mode 100644 index 00000000..a16a9f7f --- /dev/null +++ b/tests/test_state_coverage_artifact.py @@ -0,0 +1,77 @@ +"""coverage.json is a deliverable artifact, not runtime state.""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING + +import pytest + +from strix.core.paths import runtime_state_dir +from strix.report.state import ReportState +from strix.tools.coverage.tools import _record_impl, hydrate_coverage_from_disk + + +if TYPE_CHECKING: + from pathlib import Path + + +@pytest.fixture +def state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> ReportState: + monkeypatch.chdir(tmp_path) + report_state = ReportState(run_name="run-1") + hydrate_coverage_from_disk(runtime_state_dir(report_state.get_run_dir())) + return report_state + + +def _record_a_cleared_surface() -> None: + _record_impl( + surface="POST /api/orders/{id}", + risk_area="SQL injection", + outcome="no_issue_found", + evidence="14 parameters fuzzed; every query parameterized.", + agent_id="agent-1", + agent_name="injection-tester", + ) + + +def test_coverage_is_written_beside_the_other_artifacts(state: ReportState) -> None: + _record_a_cleared_surface() + + state._save_artifacts() + + document = json.loads((state.get_run_dir() / "coverage.json").read_text(encoding="utf-8")) + assert document["entries"][0]["risk_area"] == "SQL injection" + assert document["summary"]["surfaces_reviewed"] == 1 + + +def test_report_carries_a_coverage_section_from_the_ledger(state: ReportState) -> None: + _record_a_cleared_surface() + state.final_scan_result = "Scan complete." + + state._save_artifacts() + + report = (state.get_run_dir() / "penetration_test_report.md").read_text(encoding="utf-8") + assert "# Coverage" in report + assert "POST /api/orders/{id}" in report + + +def test_cleared_surfaces_reach_sarif(state: ReportState) -> None: + _record_a_cleared_surface() + + state._save_artifacts() + + sarif = json.loads((state.get_run_dir() / "findings.sarif").read_text(encoding="utf-8")) + results = sarif["runs"][0]["results"] + assert [result["kind"] for result in results] == ["pass"] + + +def test_artifacts_still_land_when_coverage_is_empty(state: ReportState) -> None: + state.final_scan_result = "Scan complete." + + state._save_artifacts() + + run_dir = state.get_run_dir() + assert (run_dir / "penetration_test_report.md").is_file() + document = json.loads((run_dir / "coverage.json").read_text(encoding="utf-8")) + assert document["entries"] == [] diff --git a/tests/test_threat_model_tool.py b/tests/test_threat_model_tool.py index 4f83e083..79239d7b 100644 --- a/tests/test_threat_model_tool.py +++ b/tests/test_threat_model_tool.py @@ -307,3 +307,37 @@ def test_repository_subdirectory_shares_the_repository_model(tmp_path: Path) -> _save_impl(str(repo), _MODEL, "root") assert _get_impl(str(repo / "src"))["found"] is True + + +def test_checkout_and_its_clone_url_are_one_identity(tmp_path: Path) -> None: + """The model an agent saves inside the checkout must be visible to an agent + that names the same repository by the URL it was cloned from.""" + repo = _make_repo(tmp_path) + _git(repo, "remote", "add", "origin", "https://github.com/acme/billing.git") + _save_impl(str(repo), _MODEL, "root") + + assert _get_impl("https://github.com/acme/billing")["found"] is True + assert _get_impl("https://github.com/acme/billing.git")["found"] is True + + +def test_ssh_and_https_remotes_are_one_identity(tmp_path: Path) -> None: + """One repository cloned over scp-style SSH and over HTTPS is one target.""" + over_ssh = _make_repo(tmp_path, "ssh-clone") + _git(over_ssh, "remote", "add", "origin", "git@github.com:acme/billing.git") + _save_impl(str(over_ssh), _MODEL, "root") + + over_https = _make_repo(tmp_path, "https-clone") + _git(over_https, "remote", "add", "origin", "https://github.com/acme/billing.git") + + assert _get_impl(str(over_https))["found"] is True + + +def test_different_repositories_on_one_host_stay_separate(tmp_path: Path) -> None: + first = _make_repo(tmp_path, "billing") + _git(first, "remote", "add", "origin", "git@github.com:acme/billing.git") + _save_impl(str(first), _MODEL, "root") + + second = _make_repo(tmp_path, "payments") + _git(second, "remote", "add", "origin", "git@github.com:acme/payments.git") + + assert _get_impl(str(second))["found"] is False