mirror of
https://github.com/usestrix/strix.git
synced 2026-08-20 10:33:34 +02:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d3c42e5498 | ||
|
|
9c1a0c9002 | ||
|
|
18556da0cd | ||
|
|
6f88b7d7d5 | ||
|
|
8d3693df8c |
@@ -167,10 +167,15 @@ strix view
|
||||
|
||||
# ...or open a specific run by name
|
||||
strix view my-run-name
|
||||
|
||||
# Expose the viewer on all IPv4 interfaces at a fixed port
|
||||
strix view --host 0.0.0.0 --port 8080 --no-open
|
||||
```
|
||||
|
||||
`strix view` starts a lightweight local server (bound to `127.0.0.1` on a random port) and opens your browser to a private, tokened link. Nothing leaves your machine: the dashboard reads the run's files straight off disk, with no cloud account or upload required. The UI ships prebuilt with Strix, so there is no extra install and no JS build step.
|
||||
|
||||
Use `--host 0.0.0.0` to make the viewer reachable from other machines. Replace `0.0.0.0` in the printed URL with the server's reachable IP or hostname. The token in that URL grants access to the selected run's scan data, history, and steering, so only share it with trusted users and restrict the port with your firewall. Requests without the token-derived session cannot read run data.
|
||||
|
||||
### What's in the dashboard
|
||||
|
||||
- **Overview**: run status, target, and a severity breakdown of everything found so far.
|
||||
|
||||
@@ -48,6 +48,7 @@ from strix.tools.reporting.tool import (
|
||||
create_vulnerability_report,
|
||||
get_report,
|
||||
list_reports,
|
||||
update_vulnerability_report,
|
||||
)
|
||||
from strix.tools.respond.tool import respond_to_user
|
||||
from strix.tools.thinking.tool import think
|
||||
@@ -500,6 +501,7 @@ _BASE_TOOLS: tuple[Tool, ...] = (
|
||||
delete_note,
|
||||
web_search,
|
||||
create_vulnerability_report,
|
||||
update_vulnerability_report,
|
||||
create_dependency_report,
|
||||
list_reports,
|
||||
get_report,
|
||||
|
||||
@@ -214,11 +214,12 @@ VALIDATION REQUIREMENTS:
|
||||
- Treat public metadata, internal-looking identifiers, source maps without secrets, and transport/configuration hygiene as observations unless validation proves unauthorized restricted-data access, modification, or service disruption
|
||||
- Every non-None Confidentiality, Integrity, or Availability metric must map to explicit evidence in the report; use Scope Changed only for a demonstrated crossing of security authorities
|
||||
- Independent verification through subagent
|
||||
- Document complete attack chain
|
||||
- Document complete attack chains. Amend an existing finding when new evidence increases or decreases its impact on the same asset and root cause
|
||||
- Keep going until you find something that matters
|
||||
- A vulnerability is ONLY considered reported when a reporting agent uses create_vulnerability_report (or create_dependency_report for known-CVE dependency/supply-chain findings) with full details. Mentions in agent_finish, finish_scan, or generic messages are NOT sufficient
|
||||
- A vulnerability is ONLY considered reported when a reporting agent uses create_vulnerability_report or update_vulnerability_report (or create_dependency_report for known-CVE dependency/supply-chain findings) with full details. Mentions in agent_finish, finish_scan, or generic messages are NOT sufficient
|
||||
- Reporting and fixing are ONE step, not two: when source is available, the reporting agent derives the concrete fix and files it INLINE via create_vulnerability_report (`code_locations` with `fix_before`/`fix_after` + `fix_pr_body`) — the report is not complete without it. Do NOT report first and then spawn a separate downstream agent to re-derive and re-apply the same patch; that just re-does the analysis and wastes tokens. (Do not silently patch a finding WITHOUT filing a report — the report, with its embedded fix, is the deliverable.)
|
||||
- DEDUPLICATION: The create_vulnerability_report tool uses LLM-based deduplication. If it rejects your report as a duplicate, DO NOT attempt to re-submit the same vulnerability. Accept the rejection and move on to testing other areas. The vulnerability has already been reported by another agent
|
||||
- DEDUPLICATION: The create_vulnerability_report tool uses LLM-based deduplication. If it rejects your report as a duplicate, do not re-submit the same vulnerability. If new validated evidence shows greater or lower impact for the same finding, use update_vulnerability_report with the returned duplicate_of id. Otherwise, move on to testing other areas. The vulnerability has already been reported
|
||||
- DISTINCT ISSUE VS AMENDMENT: File a distinct issue with create_vulnerability_report. Use update_vulnerability_report when new evidence proves greater or lower impact for the same root cause on the same asset. Supply an evidence-backed update_reason and recalculate the rating from cvss_breakdown
|
||||
- REVIEWING FILED FINDINGS (orchestrator/root agent): use list_reports to see every vulnerability filed so far in this scan (by any agent, root or child) — metadata-first with per-severity counts — and get_report to read one finding in full by its id. These are read-only orchestration tools: the root agent uses them to track coverage, avoid dispatching work on already-covered ground, assemble the finish_scan executive summary, and reason about attack-chaining across confirmed findings. Leaf/specialist agents should NOT call them — just do your assigned testing and file findings. Each entry shows which agent filed it (agent_name), and your own entries are flagged by_you. list_notes/get_note do the same for notes.
|
||||
</execution_guidelines>
|
||||
|
||||
@@ -372,7 +373,7 @@ CRITICAL RULES:
|
||||
- **REALISTIC OUTCOMES** - Some tests find nothing, some validations fail
|
||||
- **ONE AGENT = ONE TASK** - Don't let agents do multiple unrelated jobs
|
||||
- **SPAWN REACTIVELY** - Create new agents based on what you discover
|
||||
- **ONLY REPORTING AGENTS** can use create_vulnerability_report tool
|
||||
- **ONLY REPORTING AGENTS** can use create_vulnerability_report or update_vulnerability_report tools
|
||||
- **AGENT SPECIALIZATION MANDATORY** - Each agent must be highly specialized; prefer 1–3 skills, up to 5 for complex contexts
|
||||
- **NO GENERIC AGENTS** - Avoid creating broad, multi-purpose agents that dilute focus
|
||||
|
||||
|
||||
+30
-17
@@ -38,6 +38,35 @@ def _resolve_sandbox_image() -> str:
|
||||
return image
|
||||
|
||||
|
||||
def _configure_report_callbacks(report_state: ReportState, console: Console) -> None:
|
||||
def display_vulnerability(report: dict[str, Any], *, updated: bool = False) -> None:
|
||||
report_id = report.get("id", "unknown")
|
||||
|
||||
vuln_text = format_vulnerability_report(report)
|
||||
|
||||
title = (
|
||||
f"[bold yellow]{report_id.upper()} — UPDATED FINDING"
|
||||
if updated
|
||||
else f"[bold red]{report_id.upper()}"
|
||||
)
|
||||
vuln_panel = Panel(
|
||||
vuln_text,
|
||||
title=title,
|
||||
title_align="left",
|
||||
border_style="yellow" if updated else "red",
|
||||
padding=(1, 2),
|
||||
)
|
||||
|
||||
console.print(vuln_panel)
|
||||
console.print()
|
||||
|
||||
report_state.vulnerability_found_callback = display_vulnerability
|
||||
report_state.vulnerability_updated_callback = lambda report: display_vulnerability(
|
||||
report,
|
||||
updated=True,
|
||||
)
|
||||
|
||||
|
||||
async def run_cli(args: Any) -> None: # noqa: PLR0915
|
||||
console = Console()
|
||||
|
||||
@@ -105,23 +134,7 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
|
||||
report_state.set_scan_config(scan_config)
|
||||
report_state.save_run_data()
|
||||
|
||||
def display_vulnerability(report: dict[str, Any]) -> None:
|
||||
report_id = report.get("id", "unknown")
|
||||
|
||||
vuln_text = format_vulnerability_report(report)
|
||||
|
||||
vuln_panel = Panel(
|
||||
vuln_text,
|
||||
title=f"[bold red]{report_id.upper()}",
|
||||
title_align="left",
|
||||
border_style="red",
|
||||
padding=(1, 2),
|
||||
)
|
||||
|
||||
console.print(vuln_panel)
|
||||
console.print()
|
||||
|
||||
report_state.vulnerability_found_callback = display_vulnerability
|
||||
_configure_report_callbacks(report_state, console)
|
||||
|
||||
def cleanup_on_exit() -> None:
|
||||
report_state.cleanup()
|
||||
|
||||
@@ -102,6 +102,9 @@ class GoTuiRuntime:
|
||||
self.report_state.vulnerability_found_callback = lambda _report: (
|
||||
self.controller.notify_changed()
|
||||
)
|
||||
self.report_state.vulnerability_updated_callback = lambda _report: (
|
||||
self.controller.notify_changed()
|
||||
)
|
||||
self.controller.notify_changed()
|
||||
|
||||
async def start_from_setup(self, verify: bool = True) -> None:
|
||||
|
||||
@@ -45,7 +45,11 @@ def run_view(argv: list[str]) -> None:
|
||||
default=0,
|
||||
help="Port to serve on (default: an available ephemeral port).",
|
||||
)
|
||||
parser.add_argument("--host", default="127.0.0.1", help=argparse.SUPPRESS)
|
||||
parser.add_argument(
|
||||
"--host",
|
||||
default="127.0.0.1",
|
||||
help="Host to bind to (default: 127.0.0.1; use 0.0.0.0 for all IPv4 interfaces).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--no-open",
|
||||
action="store_true",
|
||||
|
||||
@@ -135,8 +135,9 @@ class _ViewerState:
|
||||
# exchanged for a session cookie only when presented on the initial page
|
||||
# load. It is the request-level authorization the review asked for:
|
||||
# reachability of the port (e.g. when bound with ``--host``) is not
|
||||
# enough to steer a live scan, trigger a report, or browse history --
|
||||
# the token is never handed to a caller who merely reaches ``/``.
|
||||
# enough to read run data, steer a live scan, trigger a report, or
|
||||
# browse history -- the token is never handed to a caller who merely
|
||||
# reaches ``/``.
|
||||
self.session_token = secrets.token_urlsafe(32)
|
||||
# Finalized in ``serve()`` once the port is known (the server binds
|
||||
# after this state is constructed); see SESSION_COOKIE_PREFIX.
|
||||
@@ -234,11 +235,11 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
|
||||
self.end_headers()
|
||||
|
||||
def _handle_api(self, path: str, query: dict[str, list[str]]) -> None:
|
||||
# The launched run is always viewable with no verification. The
|
||||
# cross-run history list (/api/runs) unlocks its entries only for a
|
||||
# caller that holds this process's session capability *and* is email
|
||||
# verified, so merely reaching an exposed --host port never leaks the
|
||||
# run list (the payload still advertises the count as a teaser).
|
||||
# The cross-run history list (/api/runs) unlocks its entries only for
|
||||
# a caller that holds this process's session capability *and* is
|
||||
# email verified, so merely reaching an exposed --host port never
|
||||
# leaks the run list (the payload still advertises the count as a
|
||||
# teaser).
|
||||
if path == "/api/runs":
|
||||
unlocked = self._has_session() and auth.is_verified()
|
||||
payload = build_runs_payload(state.base_dir, verified=unlocked)
|
||||
@@ -253,6 +254,13 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
|
||||
self._handle_auth_status()
|
||||
return
|
||||
|
||||
# All remaining GET endpoints expose run metadata or scan output.
|
||||
# Require the capability even for the run used to launch the viewer;
|
||||
# reachability of an exposed --host port must not grant data access.
|
||||
if not self._has_session():
|
||||
self._send_json(HTTPStatus.FORBIDDEN, {"error": "forbidden"})
|
||||
return
|
||||
|
||||
run_values = query.get("run")
|
||||
run_param = run_values[0] if run_values else None
|
||||
run_dir = resolve_run_dir(state.base_dir, run_param, state.run_dir)
|
||||
@@ -260,18 +268,12 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
|
||||
self._send_json(HTTPStatus.NOT_FOUND, {"error": "unknown run"})
|
||||
return
|
||||
|
||||
# The launched run is always viewable. Any *other* run's data is part
|
||||
# of the gated history: it needs this process's session capability
|
||||
# (so merely reaching an exposed --host port is not enough) *and*
|
||||
# email verification -- otherwise knowing a run name would leak its
|
||||
# metadata, vulnerabilities, report, and transcript.
|
||||
if run_dir.resolve() != state.run_dir.resolve():
|
||||
if not self._has_session():
|
||||
self._send_json(HTTPStatus.FORBIDDEN, {"error": "forbidden"})
|
||||
return
|
||||
if not auth.is_verified():
|
||||
self._send_json(HTTPStatus.UNAUTHORIZED, {"error": "unverified"})
|
||||
return
|
||||
# Any run other than the one used to launch the viewer is part of the
|
||||
# email-gated history. The session check above applies to both paths;
|
||||
# verification adds a second gate for historical run data.
|
||||
if run_dir.resolve() != state.run_dir.resolve() and not auth.is_verified():
|
||||
self._send_json(HTTPStatus.UNAUTHORIZED, {"error": "unverified"})
|
||||
return
|
||||
|
||||
if path == "/api/run":
|
||||
self._send_json(HTTPStatus.OK, read_run_summary(run_dir))
|
||||
@@ -385,7 +387,7 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
|
||||
except auth.RelayError as exc:
|
||||
self._send_relay_error(exc)
|
||||
return
|
||||
# The password is returned only to the local (127.0.0.1) browser.
|
||||
# The password is returned only to a session-authorized browser.
|
||||
self._send_json(
|
||||
HTTPStatus.OK,
|
||||
{"ok": True, "password": password, "filename": filename},
|
||||
|
||||
@@ -149,6 +149,7 @@ class ReportState:
|
||||
|
||||
self.caido_url: str | None = None
|
||||
self.vulnerability_found_callback: Callable[[dict[str, Any]], None] | None = None
|
||||
self.vulnerability_updated_callback: Callable[[dict[str, Any]], None] | None = None
|
||||
|
||||
self._sarif_repo_ctx: dict[str, Any] | None = None
|
||||
self._sarif_repo_ctx_ready: bool = False
|
||||
@@ -315,6 +316,53 @@ class ReportState:
|
||||
self.save_run_data()
|
||||
return report_id
|
||||
|
||||
def update_vulnerability_report(
|
||||
self,
|
||||
report_id: str,
|
||||
update_reason: str,
|
||||
**updates: Any,
|
||||
) -> dict[str, Any] | None:
|
||||
"""Amend a filed vulnerability report and persist the updated state."""
|
||||
report = next(
|
||||
(item for item in self.vulnerability_reports if item.get("id") == report_id),
|
||||
None,
|
||||
)
|
||||
if report is None:
|
||||
return None
|
||||
|
||||
timestamp = datetime.now(UTC).isoformat()
|
||||
previous_severity = report.get("severity")
|
||||
previous_cvss = report.get("cvss")
|
||||
changed_fields: list[str] = []
|
||||
for field, value in updates.items():
|
||||
if report.get(field) != value:
|
||||
changed_fields.append(field)
|
||||
report[field] = value
|
||||
|
||||
history_entry: dict[str, Any] = {
|
||||
"timestamp": timestamp,
|
||||
"update_reason": update_reason.strip(),
|
||||
"fields_changed": changed_fields,
|
||||
}
|
||||
if ("severity" in updates and updates.get("severity") != previous_severity) or (
|
||||
"cvss" in updates and updates.get("cvss") != previous_cvss
|
||||
):
|
||||
history_entry["previous_severity"] = previous_severity
|
||||
history_entry["previous_cvss_score"] = previous_cvss
|
||||
|
||||
history = report.setdefault("update_history", [])
|
||||
if not isinstance(history, list):
|
||||
history = []
|
||||
report["update_history"] = history
|
||||
history.append(history_entry)
|
||||
report["updated_at"] = timestamp
|
||||
|
||||
self._saved_vuln_ids.discard(report_id)
|
||||
if self.vulnerability_updated_callback:
|
||||
self.vulnerability_updated_callback(report)
|
||||
self.save_run_data()
|
||||
return report
|
||||
|
||||
def get_existing_vulnerabilities(self) -> list[dict[str, Any]]:
|
||||
return list(self.vulnerability_reports)
|
||||
|
||||
|
||||
@@ -197,6 +197,8 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL
|
||||
f"**Severity:** {report.get('severity', 'unknown').upper()}",
|
||||
f"**Found:** {report.get('timestamp', 'unknown')}",
|
||||
]
|
||||
if report.get("updated_at"):
|
||||
lines.append(f"**Updated:** {report['updated_at']}")
|
||||
|
||||
dep_meta = report.get("dependency_metadata") or {}
|
||||
metadata: list[tuple[str, Any]] = [
|
||||
@@ -304,4 +306,23 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL
|
||||
lines.append(str(report["assumptions"]))
|
||||
lines.append("")
|
||||
|
||||
update_history = report.get("update_history")
|
||||
if isinstance(update_history, list) and update_history:
|
||||
lines.append("## Amendment History\n")
|
||||
for entry in update_history:
|
||||
if not isinstance(entry, dict):
|
||||
continue
|
||||
timestamp = entry.get("timestamp", "unknown")
|
||||
reason = entry.get("update_reason", "")
|
||||
fields = ", ".join(str(field) for field in entry.get("fields_changed", []))
|
||||
lines.append(f"- **{timestamp}:** {reason}")
|
||||
lines.append(f" Changed fields: {fields or 'none'}")
|
||||
if "previous_severity" in entry or "previous_cvss_score" in entry:
|
||||
lines.append(
|
||||
" Previous rating: "
|
||||
f"{str(entry.get('previous_severity', 'unknown')).upper()} "
|
||||
f"(CVSS {entry.get('previous_cvss_score', 'unknown')})"
|
||||
)
|
||||
lines.append("")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
@@ -133,14 +133,15 @@ async def finish_scan(
|
||||
combination. You may rule out combinations you can confidently
|
||||
call unrelated — note why instead of padding chains. Any
|
||||
validated chain must already be filed via
|
||||
``create_vulnerability_report`` — a demonstrated end-to-end chain
|
||||
is a PoC-backed vulnerability, so it uses that tool even when one
|
||||
link is a dependency CVE (the standalone CVE stays in its own
|
||||
``create_dependency_report``) — and surfaced prominently in
|
||||
``create_vulnerability_report`` — or update the existing finding with
|
||||
``update_vulnerability_report`` when the chain amplifies that finding
|
||||
on the same asset and root cause. A demonstrated new chain is a
|
||||
PoC-backed vulnerability, so file it even when one link is a
|
||||
dependency CVE. Keep the standalone CVE in its own
|
||||
``create_dependency_report``. Surface the result prominently in
|
||||
``executive_summary`` / ``technical_analysis``. Finding no real
|
||||
chain after a serious attempt is acceptable; skipping the
|
||||
chaining reasoning, or ignoring a plausibly-related combination,
|
||||
is not.
|
||||
chain after a serious attempt is acceptable. Skipping the chaining
|
||||
reasoning, or ignoring a plausibly-related combination, is not.
|
||||
|
||||
**Calling this multiple times overwrites the previous report.**
|
||||
Make the single call comprehensive.
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
"""Reporting tools — file vuln findings (with dedup + CVSS) and read them back.
|
||||
"""Reporting tools — file findings (with dedup + CVSS) and read them back.
|
||||
|
||||
``create_vulnerability_report`` / ``create_dependency_report`` file findings;
|
||||
``list_reports`` / ``get_report`` let any agent (notably the root orchestrator)
|
||||
review what's been filed so far across the whole scan.
|
||||
``update_vulnerability_report`` amends a known dynamic finding;
|
||||
``list_reports`` / ``get_report`` let the root orchestrator review findings.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -162,6 +162,24 @@ _REQUIRED_FIELDS = {
|
||||
|
||||
_VALID_FIX_EFFORT = frozenset({"trivial", "low", "medium", "high"})
|
||||
|
||||
_AMENDABLE_FIELDS = (
|
||||
"title",
|
||||
"description",
|
||||
"impact",
|
||||
"technical_analysis",
|
||||
"poc_description",
|
||||
"poc_script_code",
|
||||
"remediation_steps",
|
||||
"evidence",
|
||||
"assumptions",
|
||||
"fix_effort",
|
||||
"cvss_breakdown",
|
||||
"endpoint",
|
||||
"method",
|
||||
"cwe",
|
||||
"code_locations",
|
||||
)
|
||||
|
||||
|
||||
async def _do_create( # noqa: PLR0912
|
||||
*,
|
||||
@@ -277,7 +295,10 @@ async def _do_create( # noqa: PLR0912
|
||||
"success": False,
|
||||
"error": (
|
||||
f"Potential duplicate of '{duplicate_title}' "
|
||||
f"(id={duplicate_id[:8]}...) — do not re-report the same vulnerability"
|
||||
f"(id={duplicate_id[:8]}...) — do not re-report the same vulnerability. "
|
||||
f"If new validated evidence shows greater or lower impact than filed, "
|
||||
f"amend this finding with update_vulnerability_report using id "
|
||||
f"'{duplicate_id}' instead."
|
||||
),
|
||||
"duplicate_of": duplicate_id,
|
||||
"duplicate_title": duplicate_title,
|
||||
@@ -329,6 +350,189 @@ async def _do_create( # noqa: PLR0912
|
||||
}
|
||||
|
||||
|
||||
async def _do_update( # noqa: PLR0911, PLR0912, PLR0915
|
||||
*,
|
||||
report_id: str,
|
||||
update_reason: str,
|
||||
title: str | None = None,
|
||||
description: str | None = None,
|
||||
impact: str | None = None,
|
||||
technical_analysis: str | None = None,
|
||||
poc_description: str | None = None,
|
||||
poc_script_code: str | None = None,
|
||||
remediation_steps: str | None = None,
|
||||
evidence: str | None = None,
|
||||
assumptions: str | None = None,
|
||||
fix_effort: str | None = None,
|
||||
cvss_breakdown: dict[str, str] | None = None,
|
||||
endpoint: str | None = None,
|
||||
method: str | None = None,
|
||||
cwe: str | None = None,
|
||||
code_locations: list[dict[str, Any]] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Validate and amend one known dynamic vulnerability report."""
|
||||
if not report_id.strip():
|
||||
return {"success": False, "error": "report_id cannot be empty"}
|
||||
if not update_reason.strip():
|
||||
return {"success": False, "error": "update_reason cannot be empty"}
|
||||
|
||||
raw_updates = {
|
||||
name: value
|
||||
for name, value in {
|
||||
"title": title,
|
||||
"description": description,
|
||||
"impact": impact,
|
||||
"technical_analysis": technical_analysis,
|
||||
"poc_description": poc_description,
|
||||
"poc_script_code": poc_script_code,
|
||||
"remediation_steps": remediation_steps,
|
||||
"evidence": evidence,
|
||||
"assumptions": assumptions,
|
||||
"fix_effort": fix_effort,
|
||||
"cvss_breakdown": cvss_breakdown,
|
||||
"endpoint": endpoint,
|
||||
"method": method,
|
||||
"cwe": cwe,
|
||||
"code_locations": code_locations,
|
||||
}.items()
|
||||
if value is not None and name in _AMENDABLE_FIELDS
|
||||
}
|
||||
if not raw_updates:
|
||||
return {
|
||||
"success": False,
|
||||
"error": "At least one amendable field must be supplied",
|
||||
}
|
||||
|
||||
try:
|
||||
from strix.report.state import get_global_report_state
|
||||
|
||||
report_state = get_global_report_state()
|
||||
if report_state is None:
|
||||
return {
|
||||
"success": False,
|
||||
"error": "Report state is unavailable",
|
||||
}
|
||||
|
||||
existing = report_state.get_existing_vulnerabilities()
|
||||
report = next((item for item in existing if item.get("id") == report_id), None)
|
||||
if report is None:
|
||||
valid_ids = [str(item["id"]) for item in existing if item.get("id")]
|
||||
result: dict[str, Any] = {
|
||||
"success": False,
|
||||
"error": f"Report with id '{report_id}' was not found",
|
||||
}
|
||||
if valid_ids:
|
||||
result["valid_report_ids"] = valid_ids
|
||||
return result
|
||||
|
||||
errors: list[str] = []
|
||||
updates: dict[str, Any] = {}
|
||||
for field, value in raw_updates.items():
|
||||
if field in {
|
||||
"title",
|
||||
"description",
|
||||
"impact",
|
||||
"technical_analysis",
|
||||
"poc_description",
|
||||
"poc_script_code",
|
||||
"remediation_steps",
|
||||
"evidence",
|
||||
"assumptions",
|
||||
"endpoint",
|
||||
"method",
|
||||
}:
|
||||
normalized_value = str(value).strip()
|
||||
if not normalized_value:
|
||||
errors.append(
|
||||
_REQUIRED_FIELDS.get(
|
||||
field,
|
||||
f"{field.replace('_', ' ').capitalize()} cannot be empty",
|
||||
),
|
||||
)
|
||||
else:
|
||||
updates[field] = normalized_value
|
||||
|
||||
if "fix_effort" in raw_updates:
|
||||
normalized_fix_effort = str(fix_effort).strip().lower()
|
||||
if normalized_fix_effort not in _VALID_FIX_EFFORT:
|
||||
errors.append(
|
||||
f"Invalid fix_effort: {normalized_fix_effort!r}. "
|
||||
f"Must be one of: {sorted(_VALID_FIX_EFFORT)}"
|
||||
)
|
||||
else:
|
||||
updates["fix_effort"] = normalized_fix_effort
|
||||
|
||||
if "cvss_breakdown" in raw_updates:
|
||||
if not isinstance(cvss_breakdown, dict) or not cvss_breakdown:
|
||||
errors.append("cvss_breakdown: must be an object with the 8 CVSS metrics")
|
||||
else:
|
||||
for name, valid in _CVSS_VALID.items():
|
||||
metric_value = cvss_breakdown.get(name)
|
||||
if metric_value not in valid:
|
||||
errors.append(
|
||||
f"Invalid {name}: {metric_value}. Must be one of: {valid}",
|
||||
)
|
||||
updates["cvss_breakdown"] = cvss_breakdown
|
||||
|
||||
if "cwe" in raw_updates:
|
||||
parsed_cwe = _extract_cwe(str(cwe))
|
||||
cwe_err = _validate_cwe(parsed_cwe)
|
||||
if cwe_err:
|
||||
errors.append(cwe_err)
|
||||
else:
|
||||
updates["cwe"] = parsed_cwe
|
||||
|
||||
if "code_locations" in raw_updates:
|
||||
if not isinstance(code_locations, list) or not code_locations:
|
||||
errors.append("code_locations must contain at least one location")
|
||||
else:
|
||||
parsed_locations = _normalize_code_locations(code_locations)
|
||||
if not parsed_locations:
|
||||
errors.append("code_locations must contain at least one valid location")
|
||||
else:
|
||||
errors.extend(_validate_code_locations(parsed_locations))
|
||||
updates["code_locations"] = parsed_locations
|
||||
|
||||
if errors:
|
||||
return {"success": False, "error": "Validation failed", "errors": errors}
|
||||
|
||||
if "cvss_breakdown" in updates:
|
||||
try:
|
||||
cvss_score, severity, _vector = _calculate_cvss(updates["cvss_breakdown"])
|
||||
except ValueError as exc:
|
||||
return {"success": False, "error": "Validation failed", "errors": [str(exc)]}
|
||||
updates["cvss"] = cvss_score
|
||||
updates["severity"] = severity
|
||||
|
||||
updated = report_state.update_vulnerability_report(
|
||||
report_id,
|
||||
update_reason,
|
||||
**updates,
|
||||
)
|
||||
if updated is None:
|
||||
return {
|
||||
"success": False,
|
||||
"error": f"Report with id '{report_id}' was not found",
|
||||
}
|
||||
except (AttributeError, KeyError, TypeError, ValueError) as exc:
|
||||
logger.exception("update_vulnerability_report persistence failed")
|
||||
return {"success": False, "error": f"Failed to update vulnerability report: {exc!s}"}
|
||||
|
||||
logger.info(
|
||||
"Vulnerability report updated: id=%s fields=%s",
|
||||
report_id,
|
||||
sorted(raw_updates),
|
||||
)
|
||||
return {
|
||||
"success": True,
|
||||
"message": f"Vulnerability report '{report_id}' updated successfully",
|
||||
"report_id": report_id,
|
||||
"severity": updated.get("severity"),
|
||||
"cvss_score": updated.get("cvss"),
|
||||
"updated_at": updated.get("updated_at"),
|
||||
}
|
||||
|
||||
|
||||
def _caller_identity(ctx: RunContextWrapper) -> tuple[str | None, str | None]:
|
||||
"""Return the (agent_id, agent_name) of the agent invoking this tool."""
|
||||
inner = ctx.context if isinstance(ctx.context, dict) else {}
|
||||
@@ -377,6 +581,8 @@ async def create_vulnerability_report(
|
||||
- Suspicions you haven't confirmed with a PoC.
|
||||
- Tracking multiple vulnerabilities at once — one report per vuln.
|
||||
- Re-reporting something you (or another agent) already filed.
|
||||
- A chain that only amplifies an existing finding's impact on the same
|
||||
asset and root cause. Use ``update_vulnerability_report`` instead.
|
||||
- Known-CVE dependency / supply-chain findings that can't be
|
||||
dynamically PoC'd — a vulnerable dependency version pinned in a
|
||||
lockfile/manifest that matches a published advisory. File those
|
||||
@@ -408,8 +614,10 @@ async def create_vulnerability_report(
|
||||
|
||||
Automatic LLM-based **deduplication** rejects reports that describe
|
||||
the same root cause on the same asset as an existing report. If you
|
||||
get a ``duplicate_of`` response, do NOT retry — move on to other
|
||||
areas.
|
||||
get a ``duplicate_of`` response, do not re-submit the same vulnerability.
|
||||
If new validated evidence shows greater or lower impact for that finding,
|
||||
amend it with ``update_vulnerability_report`` using the returned id.
|
||||
Otherwise, move on to other areas.
|
||||
|
||||
**Report output rules** (this content may be rendered into generated
|
||||
reports):
|
||||
@@ -701,6 +909,69 @@ async def create_vulnerability_report(
|
||||
return json.dumps(result, ensure_ascii=False, default=str)
|
||||
|
||||
|
||||
@function_tool(timeout=180, strict_mode=False)
|
||||
async def update_vulnerability_report(
|
||||
ctx: RunContextWrapper,
|
||||
report_id: str,
|
||||
update_reason: str,
|
||||
title: str | None = None,
|
||||
description: str | None = None,
|
||||
impact: str | None = None,
|
||||
technical_analysis: str | None = None,
|
||||
poc_description: str | None = None,
|
||||
poc_script_code: str | None = None,
|
||||
remediation_steps: str | None = None,
|
||||
evidence: str | None = None,
|
||||
assumptions: str | None = None,
|
||||
fix_effort: str | None = None,
|
||||
cvss_breakdown: dict[str, str] | None = None,
|
||||
endpoint: str | None = None,
|
||||
method: str | None = None,
|
||||
cwe: str | None = None,
|
||||
code_locations: list[dict[str, Any]] | None = None,
|
||||
) -> str:
|
||||
"""Amend a known vulnerability report when new evidence changes its impact.
|
||||
|
||||
Use ``create_vulnerability_report`` when the chain proves a distinct issue.
|
||||
Use this tool when the same root cause on the same asset has greater impact.
|
||||
The tool recomputes severity and score from ``cvss_breakdown``.
|
||||
|
||||
Apply the same report output rules and CVSS calibration discipline as the
|
||||
create tool. An amendment must use evidence from a validated result. Do
|
||||
not use this tool for a speculative upgrade.
|
||||
|
||||
Pass at least one amendable field:
|
||||
|
||||
- ``title``, ``description``, ``impact``, ``technical_analysis``,
|
||||
``poc_description``, ``poc_script_code``, ``remediation_steps``,
|
||||
``evidence``, ``assumptions``, ``fix_effort``, ``cvss_breakdown``,
|
||||
``endpoint``, ``method``, ``cwe``, or ``code_locations``.
|
||||
|
||||
Do not change ``target``, ``cve``, or dependency metadata. File a new
|
||||
report instead when those values must change.
|
||||
"""
|
||||
result = await _do_update(
|
||||
report_id=report_id,
|
||||
update_reason=update_reason,
|
||||
title=title,
|
||||
description=description,
|
||||
impact=impact,
|
||||
technical_analysis=technical_analysis,
|
||||
poc_description=poc_description,
|
||||
poc_script_code=poc_script_code,
|
||||
remediation_steps=remediation_steps,
|
||||
evidence=evidence,
|
||||
assumptions=assumptions,
|
||||
fix_effort=fix_effort,
|
||||
cvss_breakdown=cvss_breakdown,
|
||||
endpoint=endpoint,
|
||||
method=method,
|
||||
cwe=cwe,
|
||||
code_locations=code_locations,
|
||||
)
|
||||
return json.dumps(result, ensure_ascii=False, default=str)
|
||||
|
||||
|
||||
_DEP_SEVERITY_FROM_CVSS = {
|
||||
(9.0, 10.0): "critical",
|
||||
(7.0, 9.0): "high",
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from types import SimpleNamespace
|
||||
from typing import Any, cast
|
||||
from unittest.mock import Mock
|
||||
|
||||
from rich.console import Console
|
||||
from rich.panel import Panel
|
||||
|
||||
from strix.interface import cli
|
||||
|
||||
|
||||
def test_cli_report_callbacks_render_new_and_updated_findings() -> None:
|
||||
report_state = SimpleNamespace(
|
||||
vulnerability_found_callback=None,
|
||||
vulnerability_updated_callback=None,
|
||||
)
|
||||
console = Mock(spec=Console)
|
||||
|
||||
cli._configure_report_callbacks(cast("Any", report_state), console)
|
||||
|
||||
report = {"id": "vuln-0001", "title": "Unsafe redirect"}
|
||||
report_state.vulnerability_found_callback(report)
|
||||
report_state.vulnerability_updated_callback(report)
|
||||
|
||||
panels = [call.args[0] for call in console.print.call_args_list if call.args]
|
||||
assert all(isinstance(panel, Panel) for panel in panels)
|
||||
assert panels[0].title == "[bold red]VULN-0001"
|
||||
assert panels[1].title == "[bold yellow]VULN-0001 — UPDATED FINDING"
|
||||
@@ -12,6 +12,7 @@ import threading
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from typing import Any, cast
|
||||
from unittest.mock import Mock
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -95,6 +96,25 @@ def test_binary_command_ignores_unconstrained_path_sidecar(
|
||||
GoTuiRuntime.binary_command()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_init_run_state_wires_updated_report_callback(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
runtime = GoTuiRuntime(args())
|
||||
notify_changed = Mock()
|
||||
monkeypatch.setattr(runtime.controller, "notify_changed", notify_changed)
|
||||
|
||||
runtime.init_run_state()
|
||||
|
||||
assert runtime.report_state is not None
|
||||
assert runtime.report_state.vulnerability_updated_callback is not None
|
||||
notify_changed.reset_mock()
|
||||
runtime.report_state.vulnerability_updated_callback({"id": "vuln-0001"})
|
||||
notify_changed.assert_called_once_with()
|
||||
|
||||
|
||||
def test_child_environment_excludes_credentials(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("OPENAI_API_KEY", "openai-secret")
|
||||
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "aws-id")
|
||||
|
||||
+49
-7
@@ -11,6 +11,7 @@ from typing import TYPE_CHECKING
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
from strix.core.paths import latest_run_dir, runs_base_dir
|
||||
from strix.interface.viewer.cli import run_view
|
||||
from strix.interface.viewer.server import serve
|
||||
from strix.interface.viewer.transcript import (
|
||||
build_run_state,
|
||||
@@ -48,6 +49,31 @@ def test_latest_run_dir_none_when_no_runs(tmp_path: Path, monkeypatch: pytest.Mo
|
||||
assert runs_base_dir() == tmp_path / "strix_runs"
|
||||
|
||||
|
||||
def test_view_cli_help_includes_host(capsys: pytest.CaptureFixture[str]) -> None:
|
||||
try:
|
||||
run_view(["--help"])
|
||||
except SystemExit as exc:
|
||||
assert exc.code == 0
|
||||
else:
|
||||
raise AssertionError("--help should exit")
|
||||
|
||||
help_text = capsys.readouterr().out
|
||||
assert "--host HOST" in help_text
|
||||
assert "0.0.0.0" in help_text
|
||||
|
||||
|
||||
def test_server_can_bind_all_ipv4_interfaces(tmp_path: Path) -> None:
|
||||
run_dir = _make_run(tmp_path, "remote", status="running", end_time=None)
|
||||
|
||||
httpd, url, _ = serve(run_dir, host="0.0.0.0", open_browser=False)
|
||||
try:
|
||||
assert httpd.server_address[0] == "0.0.0.0"
|
||||
assert url == f"http://0.0.0.0:{httpd.server_address[1]}"
|
||||
finally:
|
||||
httpd.shutdown()
|
||||
httpd.server_close()
|
||||
|
||||
|
||||
def test_latest_run_dir_picks_newest_by_record_mtime(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
@@ -173,14 +199,15 @@ def test_server_serves_api_and_static(tmp_path: Path, monkeypatch: pytest.Monkey
|
||||
(assets / "assets" / "app.js").write_text("console.log(1)", encoding="utf-8")
|
||||
monkeypatch.setattr("strix.interface.viewer.server.bundle_dir", lambda: assets)
|
||||
|
||||
httpd, url, _ = serve(run_dir, open_browser=False)
|
||||
httpd, url, token = serve(run_dir, open_browser=False)
|
||||
try:
|
||||
status, ctype, body = _get(f"{url}/api/run")
|
||||
cookie = _session_cookie(url, token)
|
||||
status, ctype, body = _get(f"{url}/api/run", cookie=cookie)
|
||||
assert status == 200
|
||||
assert "application/json" in ctype
|
||||
assert json.loads(body)["finished"] is True
|
||||
|
||||
status, _, body = _get(f"{url}/api/transcript")
|
||||
status, _, body = _get(f"{url}/api/transcript", cookie=cookie)
|
||||
assert {a["id"] for a in json.loads(body)["agents"]} == {"root", "child"}
|
||||
|
||||
# Real asset is served.
|
||||
@@ -429,6 +456,22 @@ def test_unauthorized_client_cannot_acquire_capability(
|
||||
httpd.server_close()
|
||||
|
||||
|
||||
def test_run_data_requires_session(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
run_dir = _make_run(tmp_path, "private", status="completed", end_time="2026-01-01T00:00:00Z")
|
||||
_bundle(tmp_path, monkeypatch)
|
||||
|
||||
httpd, url, token = serve(run_dir, open_browser=False)
|
||||
try:
|
||||
cookie = _session_cookie(url, token)
|
||||
for path in ("/api/run", "/api/vulnerabilities", "/api/report", "/api/transcript"):
|
||||
assert _get_status(url + path) == 403, path
|
||||
assert _get_status(url + path, cookie=f"{_cookie_name(url)}=wrong") == 403, path
|
||||
assert _get_status(url + path, cookie=cookie) == 200, path
|
||||
finally:
|
||||
httpd.shutdown()
|
||||
httpd.server_close()
|
||||
|
||||
|
||||
def test_auth_status_reflects_expiry(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
run_dir = _make_run(tmp_path, "status", status="running", end_time=None)
|
||||
_bundle(tmp_path, monkeypatch)
|
||||
@@ -561,11 +604,10 @@ def test_historical_run_data_requires_verification(
|
||||
|
||||
httpd, url, token = serve(launched, open_browser=False)
|
||||
try:
|
||||
# The launched run is always viewable, no verification and no cookie.
|
||||
status, _, _ = _get(f"{url}/api/run")
|
||||
assert status == 200
|
||||
|
||||
# The launched run needs the session capability, but not email verification.
|
||||
assert _get_status(f"{url}/api/run") == 403
|
||||
cookie = _session_cookie(url, token)
|
||||
assert _get_status(f"{url}/api/run", cookie=cookie) == 200
|
||||
|
||||
# A different run needs the session capability first: a cookie-less
|
||||
# caller is forbidden even once the machine is verified.
|
||||
|
||||
@@ -0,0 +1,262 @@
|
||||
"""Tests for amending filed vulnerability reports."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import TYPE_CHECKING, Any, cast
|
||||
|
||||
import pytest
|
||||
|
||||
from strix.report.state import ReportState, set_global_report_state
|
||||
from strix.tools.reporting.tool import _do_update
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
_LOW_CVSS = {
|
||||
"attack_vector": "N",
|
||||
"attack_complexity": "H",
|
||||
"privileges_required": "H",
|
||||
"user_interaction": "R",
|
||||
"scope": "U",
|
||||
"confidentiality": "L",
|
||||
"integrity": "N",
|
||||
"availability": "N",
|
||||
}
|
||||
|
||||
_CRITICAL_CVSS = {
|
||||
"attack_vector": "N",
|
||||
"attack_complexity": "L",
|
||||
"privileges_required": "N",
|
||||
"user_interaction": "N",
|
||||
"scope": "U",
|
||||
"confidentiality": "H",
|
||||
"integrity": "H",
|
||||
"availability": "H",
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def report_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> ReportState:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
state = ReportState(run_name="test-run")
|
||||
set_global_report_state(state)
|
||||
state.add_vulnerability_report(
|
||||
title="Unsafe redirect",
|
||||
severity="low",
|
||||
description="The redirect accepts attacker input.",
|
||||
impact="Limited redirect manipulation.",
|
||||
target="https://app.example.com",
|
||||
technical_analysis="The handler does not validate the destination.",
|
||||
poc_description="Send a crafted redirect value.",
|
||||
poc_script_code="GET /redirect?url=https://example.net",
|
||||
remediation_steps="Validate redirect destinations.",
|
||||
evidence="The response contains the attacker-controlled destination.",
|
||||
assumptions="Assumes a victim follows the link.",
|
||||
fix_effort="medium",
|
||||
cvss=3.1,
|
||||
cvss_breakdown=_LOW_CVSS,
|
||||
endpoint="/redirect",
|
||||
method="GET",
|
||||
cwe="CWE-601",
|
||||
code_locations=[
|
||||
{
|
||||
"file": "src/redirect.py",
|
||||
"start_line": 10,
|
||||
"end_line": 12,
|
||||
"snippet": "return redirect(url)",
|
||||
},
|
||||
],
|
||||
)
|
||||
return state
|
||||
|
||||
|
||||
async def test_update_rejects_unknown_report_id() -> None:
|
||||
result = await _do_update(
|
||||
report_id="vuln-9999",
|
||||
update_reason="The validation pass proved broader impact.",
|
||||
impact="Broader impact.",
|
||||
)
|
||||
|
||||
assert result["success"] is False
|
||||
assert "vuln-9999" in result["error"]
|
||||
assert result["valid_report_ids"] == ["vuln-0001"]
|
||||
|
||||
|
||||
async def test_update_requires_an_amendable_field() -> None:
|
||||
result = await _do_update(
|
||||
report_id="vuln-0001",
|
||||
update_reason="The validation pass found no new field to amend.",
|
||||
)
|
||||
|
||||
assert result == {
|
||||
"success": False,
|
||||
"error": "At least one amendable field must be supplied",
|
||||
}
|
||||
|
||||
|
||||
async def test_update_requires_nonempty_reason() -> None:
|
||||
result = await _do_update(
|
||||
report_id="vuln-0001",
|
||||
update_reason=" ",
|
||||
impact="The impact is broader.",
|
||||
)
|
||||
|
||||
assert result == {"success": False, "error": "update_reason cannot be empty"}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("field", "value", "expected_error"),
|
||||
[
|
||||
("impact", "", "Impact cannot be empty"),
|
||||
("impact", " ", "Impact cannot be empty"),
|
||||
("endpoint", "", "Endpoint cannot be empty"),
|
||||
("endpoint", " ", "Endpoint cannot be empty"),
|
||||
],
|
||||
)
|
||||
async def test_update_rejects_blank_text_fields(
|
||||
report_state: ReportState,
|
||||
field: str,
|
||||
value: str,
|
||||
expected_error: str,
|
||||
) -> None:
|
||||
original_value = report_state.vulnerability_reports[0][field]
|
||||
update = cast("dict[str, Any]", {field: value})
|
||||
|
||||
result = await _do_update(
|
||||
report_id="vuln-0001",
|
||||
update_reason="The source review supplied no content for this field.",
|
||||
**update,
|
||||
)
|
||||
|
||||
assert result["success"] is False
|
||||
assert expected_error in result["errors"]
|
||||
assert report_state.vulnerability_reports[0][field] == original_value
|
||||
|
||||
|
||||
async def test_update_changes_impact_only(report_state: ReportState) -> None:
|
||||
result = await _do_update(
|
||||
report_id="vuln-0001",
|
||||
update_reason="The confirmed chain exposes account data.",
|
||||
impact="The chain exposes account data.",
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
report = report_state.vulnerability_reports[0]
|
||||
assert report["impact"] == "The chain exposes account data."
|
||||
assert report["severity"] == "low"
|
||||
assert report["cvss"] == 3.1
|
||||
assert report["update_history"][0]["fields_changed"] == ["impact"]
|
||||
assert "previous_severity" not in report["update_history"][0]
|
||||
|
||||
|
||||
async def test_cvss_update_recomputes_score_and_severity(report_state: ReportState) -> None:
|
||||
result = await _do_update(
|
||||
report_id="vuln-0001",
|
||||
update_reason="The exploit chain proves full account compromise.",
|
||||
cvss_breakdown=_CRITICAL_CVSS,
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
report = report_state.vulnerability_reports[0]
|
||||
assert report["severity"] == "critical"
|
||||
assert report["cvss"] == 9.8
|
||||
history = report["update_history"][0]
|
||||
assert history["fields_changed"] == ["cvss_breakdown", "cvss", "severity"]
|
||||
assert history["previous_severity"] == "low"
|
||||
assert history["previous_cvss_score"] == 3.1
|
||||
|
||||
|
||||
async def test_update_history_is_append_only(report_state: ReportState) -> None:
|
||||
await _do_update(
|
||||
report_id="vuln-0001",
|
||||
update_reason="The chain proves account access.",
|
||||
impact="Account access is possible.",
|
||||
)
|
||||
await _do_update(
|
||||
report_id="vuln-0001",
|
||||
update_reason="The second proof confirms persistent access.",
|
||||
evidence="The second proof confirms persistent access.",
|
||||
)
|
||||
|
||||
history = report_state.vulnerability_reports[0]["update_history"]
|
||||
assert len(history) == 2
|
||||
assert history[0]["update_reason"] == "The chain proves account access."
|
||||
assert history[1]["update_reason"] == "The second proof confirms persistent access."
|
||||
assert all("description" not in entry for entry in history)
|
||||
|
||||
|
||||
async def test_update_callback_fires(report_state: ReportState) -> None:
|
||||
updated: list[dict[str, Any]] = []
|
||||
report_state.vulnerability_updated_callback = updated.append
|
||||
|
||||
await _do_update(
|
||||
report_id="vuln-0001",
|
||||
update_reason="The new proof confirms data exposure.",
|
||||
evidence="The new proof confirms data exposure.",
|
||||
)
|
||||
|
||||
assert len(updated) == 1
|
||||
assert updated[0] is report_state.vulnerability_reports[0]
|
||||
|
||||
|
||||
async def test_update_persists_all_report_artifacts(report_state: ReportState) -> None:
|
||||
result = await _do_update(
|
||||
report_id="vuln-0001",
|
||||
update_reason="The chain proves account takeover.",
|
||||
description="The redirect reaches the account takeover flow.",
|
||||
impact="An attacker can take over an account.",
|
||||
cvss_breakdown=_CRITICAL_CVSS,
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
run_dir = report_state.get_run_dir()
|
||||
finding_md = (run_dir / "vulnerabilities" / "vuln-0001.md").read_text(encoding="utf-8")
|
||||
findings = json.loads((run_dir / "vulnerabilities.json").read_text(encoding="utf-8"))
|
||||
sarif = json.loads((run_dir / "findings.sarif").read_text(encoding="utf-8"))
|
||||
sarif_finding = sarif["runs"][0]["results"][0]
|
||||
|
||||
assert "An attacker can take over an account." in finding_md
|
||||
assert findings[0]["impact"] == "An attacker can take over an account."
|
||||
assert findings[0]["severity"] == "critical"
|
||||
assert sarif_finding["properties"]["strix"]["impact"] == (
|
||||
"An attacker can take over an account."
|
||||
)
|
||||
assert sarif_finding["properties"]["strix"]["severity"] == "critical"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code_locations",
|
||||
[[], [{"file": "../invalid.py", "start_line": 1}]],
|
||||
)
|
||||
async def test_update_rejects_empty_code_locations(
|
||||
report_state: ReportState,
|
||||
code_locations: list[dict[str, Any]],
|
||||
) -> None:
|
||||
original_locations = report_state.vulnerability_reports[0]["code_locations"]
|
||||
|
||||
result = await _do_update(
|
||||
report_id="vuln-0001",
|
||||
update_reason="The source review did not provide a valid location.",
|
||||
code_locations=code_locations,
|
||||
)
|
||||
|
||||
assert result["success"] is False
|
||||
assert any("code_locations" in error for error in result["errors"])
|
||||
assert report_state.vulnerability_reports[0]["code_locations"] == original_locations
|
||||
|
||||
|
||||
async def test_update_fails_without_global_report_state(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.setattr("strix.report.state._global_report_state", None)
|
||||
|
||||
result = await _do_update(
|
||||
report_id="vuln-0001",
|
||||
update_reason="The new proof confirms broader impact.",
|
||||
impact="Broader impact.",
|
||||
)
|
||||
|
||||
assert result == {"success": False, "error": "Report state is unavailable"}
|
||||
Reference in New Issue
Block a user