mirror of
https://github.com/usestrix/strix.git
synced 2026-08-16 17:27:26 +02:00
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.
This commit is contained in:
Alex Schapiro
@@ -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())
|
||||
@@ -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
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
+37
-2
@@ -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)")
|
||||
|
||||
+47
-10
@@ -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"]))
|
||||
|
||||
@@ -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,
|
||||
|
||||
+30
-12
@@ -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")
|
||||
|
||||
@@ -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<host>[^:/]+):(?P<path>.+)$", 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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"]
|
||||
@@ -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"}
|
||||
@@ -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
|
||||
|
||||
@@ -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."
|
||||
|
||||
@@ -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"] == []
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user