Compare commits

..
Author SHA1 Message Date
bearsyankees 4e50a8e32c Add semantic browser and Electron security skills 2026-08-19 12:05:02 -04:00
15 changed files with 62 additions and 786 deletions
-5
View File
@@ -167,15 +167,10 @@ 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.
-2
View File
@@ -48,7 +48,6 @@ 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
@@ -501,7 +500,6 @@ _BASE_TOOLS: tuple[Tool, ...] = (
delete_note,
web_search,
create_vulnerability_report,
update_vulnerability_report,
create_dependency_report,
list_reports,
get_report,
+4 -5
View File
@@ -214,12 +214,11 @@ 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 chains. Amend an existing finding when new evidence increases or decreases its impact on the same asset and root cause
- Document complete attack chain
- Keep going until you find something that matters
- 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
- 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
- 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 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
- 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
- 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>
@@ -373,7 +372,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 or update_vulnerability_report tools
- **ONLY REPORTING AGENTS** can use create_vulnerability_report tool
- **AGENT SPECIALIZATION MANDATORY** - Each agent must be highly specialized; prefer 13 skills, up to 5 for complex contexts
- **NO GENERIC AGENTS** - Avoid creating broad, multi-purpose agents that dilute focus
+17 -30
View File
@@ -38,35 +38,6 @@ 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()
@@ -134,7 +105,23 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
report_state.set_scan_config(scan_config)
report_state.save_run_data()
_configure_report_callbacks(report_state, console)
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
def cleanup_on_exit() -> None:
report_state.cleanup()
-3
View File
@@ -102,9 +102,6 @@ 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:
+1 -5
View File
@@ -45,11 +45,7 @@ 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="Host to bind to (default: 127.0.0.1; use 0.0.0.0 for all IPv4 interfaces).",
)
parser.add_argument("--host", default="127.0.0.1", help=argparse.SUPPRESS)
parser.add_argument(
"--no-open",
action="store_true",
+20 -22
View File
@@ -135,9 +135,8 @@ 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 read run data, steer a live scan, trigger a report, or
# browse history -- the token is never handed to a caller who merely
# reaches ``/``.
# enough to 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.
@@ -235,11 +234,11 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
self.end_headers()
def _handle_api(self, path: str, query: dict[str, list[str]]) -> None:
# 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 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).
if path == "/api/runs":
unlocked = self._has_session() and auth.is_verified()
payload = build_runs_payload(state.base_dir, verified=unlocked)
@@ -254,13 +253,6 @@ 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)
@@ -268,12 +260,18 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
self._send_json(HTTPStatus.NOT_FOUND, {"error": "unknown run"})
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
# 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
if path == "/api/run":
self._send_json(HTTPStatus.OK, read_run_summary(run_dir))
@@ -387,7 +385,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 a session-authorized browser.
# The password is returned only to the local (127.0.0.1) browser.
self._send_json(
HTTPStatus.OK,
{"ok": True, "password": password, "filename": filename},
-48
View File
@@ -149,7 +149,6 @@ 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
@@ -316,53 +315,6 @@ 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)
-21
View File
@@ -197,8 +197,6 @@ 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]] = [
@@ -306,23 +304,4 @@ 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)
+7 -8
View File
@@ -133,15 +133,14 @@ 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`` — 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
``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
``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.
+6 -277
View File
@@ -1,8 +1,8 @@
"""Reporting tools — file findings (with dedup + CVSS) and read them back.
"""Reporting tools — file vuln findings (with dedup + CVSS) and read them back.
``create_vulnerability_report`` / ``create_dependency_report`` file findings;
``update_vulnerability_report`` amends a known dynamic finding;
``list_reports`` / ``get_report`` let the root orchestrator review findings.
``list_reports`` / ``get_report`` let any agent (notably the root orchestrator)
review what's been filed so far across the whole scan.
"""
from __future__ import annotations
@@ -162,24 +162,6 @@ _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
*,
@@ -295,10 +277,7 @@ 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"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."
f"(id={duplicate_id[:8]}...) — do not re-report the same vulnerability"
),
"duplicate_of": duplicate_id,
"duplicate_title": duplicate_title,
@@ -350,189 +329,6 @@ 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 {}
@@ -581,8 +377,6 @@ 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
@@ -614,10 +408,8 @@ 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 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.
get a ``duplicate_of`` response, do NOT retry — move on to other
areas.
**Report output rules** (this content may be rendered into generated
reports):
@@ -909,69 +701,6 @@ 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",
-29
View File
@@ -1,29 +0,0 @@
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"
-20
View File
@@ -12,7 +12,6 @@ import threading
from pathlib import Path
from types import SimpleNamespace
from typing import Any, cast
from unittest.mock import Mock
import pytest
@@ -96,25 +95,6 @@ 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")
+7 -49
View File
@@ -11,7 +11,6 @@ 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,
@@ -49,31 +48,6 @@ 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:
@@ -199,15 +173,14 @@ 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, token = serve(run_dir, open_browser=False)
httpd, url, _ = serve(run_dir, open_browser=False)
try:
cookie = _session_cookie(url, token)
status, ctype, body = _get(f"{url}/api/run", cookie=cookie)
status, ctype, body = _get(f"{url}/api/run")
assert status == 200
assert "application/json" in ctype
assert json.loads(body)["finished"] is True
status, _, body = _get(f"{url}/api/transcript", cookie=cookie)
status, _, body = _get(f"{url}/api/transcript")
assert {a["id"] for a in json.loads(body)["agents"]} == {"root", "child"}
# Real asset is served.
@@ -456,22 +429,6 @@ 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)
@@ -604,10 +561,11 @@ def test_historical_run_data_requires_verification(
httpd, url, token = serve(launched, open_browser=False)
try:
# The launched run needs the session capability, but not email verification.
assert _get_status(f"{url}/api/run") == 403
# The launched run is always viewable, no verification and no cookie.
status, _, _ = _get(f"{url}/api/run")
assert status == 200
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.
-262
View File
@@ -1,262 +0,0 @@
"""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"}