mirror of
https://github.com/usestrix/strix.git
synced 2026-08-20 02:23:35 +02:00
Add vulnerability report amendments
This commit is contained in:
@@ -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
|
||||
- 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
|
||||
|
||||
|
||||
+53
-1
@@ -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)
|
||||
|
||||
@@ -454,7 +502,11 @@ class ReportState:
|
||||
run_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if self.final_scan_result:
|
||||
write_executive_report(run_dir, self.final_scan_result)
|
||||
write_executive_report(
|
||||
run_dir,
|
||||
self.final_scan_result,
|
||||
self.vulnerability_reports,
|
||||
)
|
||||
|
||||
if self.vulnerability_reports:
|
||||
write_vulnerabilities(run_dir, self.vulnerability_reports, self._saved_vuln_ids)
|
||||
|
||||
+30
-1
@@ -113,12 +113,20 @@ def write_run_record(run_dir: Path, run_record: dict[str, Any]) -> None:
|
||||
)
|
||||
|
||||
|
||||
def write_executive_report(run_dir: Path, final_scan_result: str) -> None:
|
||||
def write_executive_report(
|
||||
run_dir: Path,
|
||||
final_scan_result: str,
|
||||
vulnerability_reports: list[dict[str, Any]] | None = None,
|
||||
) -> None:
|
||||
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")
|
||||
if vulnerability_reports:
|
||||
f.write("\n## Vulnerability Findings\n\n")
|
||||
for report in vulnerability_reports:
|
||||
f.write(f"{render_vulnerability_md(report)}\n")
|
||||
logger.info("Saved final penetration test report to: %s", path)
|
||||
|
||||
|
||||
@@ -197,6 +205,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 +314,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
|
||||
*,
|
||||
@@ -329,6 +347,200 @@ 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,
|
||||
target: str | None = None,
|
||||
cve: str | None = None,
|
||||
dependency_metadata: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Validate and amend one known dynamic vulnerability report."""
|
||||
immutable_attempts = {
|
||||
name: value
|
||||
for name, value in {
|
||||
"target": target,
|
||||
"cve": cve,
|
||||
"dependency_metadata": dependency_metadata,
|
||||
}.items()
|
||||
if value is not None
|
||||
}
|
||||
if immutable_attempts:
|
||||
fields = ", ".join(immutable_attempts)
|
||||
return {
|
||||
"success": False,
|
||||
"error": (
|
||||
f"Cannot amend {fields}. File a new report instead because "
|
||||
"the target and dependency metadata identify the finding."
|
||||
),
|
||||
}
|
||||
|
||||
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",
|
||||
}:
|
||||
updates[field] = str(value).strip()
|
||||
|
||||
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):
|
||||
errors.append("code_locations must be a list")
|
||||
else:
|
||||
parsed_locations = _normalize_code_locations(code_locations)
|
||||
if parsed_locations:
|
||||
errors.extend(_validate_code_locations(parsed_locations))
|
||||
updates["code_locations"] = parsed_locations or []
|
||||
|
||||
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 +589,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
|
||||
@@ -701,6 +915,75 @@ 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,
|
||||
target: str | None = None,
|
||||
cve: str | None = None,
|
||||
dependency_metadata: 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,
|
||||
target=target,
|
||||
cve=cve,
|
||||
dependency_metadata=dependency_metadata,
|
||||
)
|
||||
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,223 @@
|
||||
"""Tests for amending filed vulnerability reports."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
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",
|
||||
)
|
||||
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"}
|
||||
|
||||
|
||||
async def test_update_rejects_target_and_cve_changes() -> None:
|
||||
result = await _do_update(
|
||||
report_id="vuln-0001",
|
||||
update_reason="The chain reached a second asset.",
|
||||
target="https://other.example.com",
|
||||
cve="CVE-2024-12345",
|
||||
)
|
||||
|
||||
assert result["success"] is False
|
||||
assert "File a new report instead" in result["error"]
|
||||
assert "target" in result["error"]
|
||||
assert "cve" in result["error"]
|
||||
|
||||
|
||||
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:
|
||||
report_state.final_scan_result = "# Executive Summary\n\nThe scan is complete."
|
||||
report_state.save_run_data()
|
||||
|
||||
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")
|
||||
executive_md = (run_dir / "penetration_test_report.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 "An attacker can take over an account." in executive_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"
|
||||
|
||||
|
||||
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