mirror of
https://github.com/usestrix/strix.git
synced 2026-09-08 22:21:05 +00:00
report: add update_vulnerability_report so an agent can revise a filed finding (#1210)
Co-authored-by: Ahmed Allam <ahmed39652003@gmail.com>
This commit is contained in:
parent
3de9471431
commit
46cf2f52f3
13 changed files with 1288 additions and 35 deletions
|
|
@ -52,6 +52,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
|
||||
|
|
@ -580,6 +581,7 @@ _BASE_TOOLS: tuple[Tool, ...] = (
|
|||
web_search,
|
||||
create_vulnerability_report,
|
||||
create_dependency_report,
|
||||
update_vulnerability_report,
|
||||
list_reports,
|
||||
get_report,
|
||||
list_requests,
|
||||
|
|
|
|||
|
|
@ -239,7 +239,8 @@ VALIDATION REQUIREMENTS:
|
|||
- Before filing any report, run the counterevidence pass: argue the strongest case AGAINST the finding, record what you found in the `counterevidence` field, set `confidence` honestly (a static-only trace you couldn't execute is at best `medium`), and state what evidence would change the severity. See the counterevidence and severity-calibration knowledge above.
|
||||
- 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 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 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. If your evidence proves more than the finding it matched (a working exploit where that one had only a static trace, a chain that raises the impact), revise that finding with update_vulnerability_report using the duplicate_of id — never re-file it.
|
||||
- REVISING A FINDING: use update_vulnerability_report (report id + the fields you want to replace + update_reason) when you learn something a finding already on file does not carry — you built the PoC after filing it, a chain raised its impact, further testing weakened it, or its counterevidence/remediation/code locations were wrong. Editing a finding needs no duplicate verdict, and it is always better than filing a second report for the same issue. Read the finding first with get_report, and pass only the fields that change.
|
||||
- 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.
|
||||
|
||||
STATE & COORDINATION TOOLS (when and how):
|
||||
|
|
|
|||
|
|
@ -105,14 +105,15 @@ 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:
|
||||
def display_vulnerability(report: dict[str, Any], *, updated: bool = False) -> None:
|
||||
report_id = report.get("id", "unknown")
|
||||
|
||||
vuln_text = format_vulnerability_report(report)
|
||||
|
||||
suffix = " (updated)" if updated else ""
|
||||
vuln_panel = Panel(
|
||||
vuln_text,
|
||||
title=f"[bold red]{report_id.upper()}",
|
||||
title=f"[bold red]{report_id.upper()}{suffix}",
|
||||
title_align="left",
|
||||
border_style="red",
|
||||
padding=(1, 2),
|
||||
|
|
@ -122,6 +123,9 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
|
|||
console.print()
|
||||
|
||||
report_state.vulnerability_found_callback = display_vulnerability
|
||||
report_state.vulnerability_updated_callback = lambda report: display_vulnerability(
|
||||
report, updated=True
|
||||
)
|
||||
|
||||
def cleanup_on_exit() -> None:
|
||||
report_state.cleanup()
|
||||
|
|
|
|||
|
|
@ -194,3 +194,28 @@ func TestVulnerabilityReportRendersCalibrationFields(t *testing.T) {
|
|||
"Fix Verification", "bypass review reasoned only",
|
||||
)
|
||||
}
|
||||
|
||||
func TestVulnerabilityReportUpdateRendersReportAndReason(t *testing.T) {
|
||||
out := ansi.Strip(Tool(tool("update_vulnerability_report",
|
||||
map[string]any{
|
||||
"report_id": "vuln-0009",
|
||||
"update_reason": "built a working unauthenticated file write against the endpoint",
|
||||
"poc_script_code": "curl -X PATCH https://target/files/uuid",
|
||||
},
|
||||
map[string]any{
|
||||
"success": true,
|
||||
"action": "updated",
|
||||
"report_id": "vuln-0009",
|
||||
"severity": "critical",
|
||||
"cvss_score": 9.3,
|
||||
"updated_fields": []any{"poc_script_code"},
|
||||
},
|
||||
"completed")))
|
||||
requireContains(t, out,
|
||||
"Vulnerability Report Updated",
|
||||
"vuln-0009",
|
||||
"built a working unauthenticated file write",
|
||||
"CRITICAL",
|
||||
"9.3",
|
||||
)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -84,6 +84,8 @@ func Tool(data map[string]any) string {
|
|||
return renderViewImage(args, result)
|
||||
case "create_vulnerability_report":
|
||||
return renderVulnerabilityReport(args, result)
|
||||
case "update_vulnerability_report":
|
||||
return renderVulnerabilityReportUpdate(args, result)
|
||||
case "create_dependency_report":
|
||||
return renderDependencyReport(args, result)
|
||||
case "list_reports":
|
||||
|
|
|
|||
|
|
@ -12,15 +12,27 @@ import (
|
|||
// ---------------------------------------------------------------------------
|
||||
|
||||
func renderVulnerabilityReport(args map[string]any, result any) string {
|
||||
return renderReport(args, result, "Vulnerability Report", "Creating report...")
|
||||
}
|
||||
|
||||
// A revision names the report it changes and carries only the fields it
|
||||
// replaces, so it renders the same sections with the ones it left alone absent.
|
||||
func renderVulnerabilityReportUpdate(args map[string]any, result any) string {
|
||||
return renderReport(args, result, "Vulnerability Report Updated", "Updating report...")
|
||||
}
|
||||
|
||||
func renderReport(args map[string]any, result any, heading, pending string) string {
|
||||
resultMap, _ := result.(map[string]any)
|
||||
var b strings.Builder
|
||||
b.WriteString("🐞 " + Bold(ReportHdr).Render("Vulnerability Report"))
|
||||
b.WriteString("🐞 " + Bold(ReportHdr).Render(heading))
|
||||
|
||||
field := func(label, value string) {
|
||||
if value != "" {
|
||||
b.WriteString("\n\n" + Bold(Field).Render(label+": ") + value)
|
||||
}
|
||||
}
|
||||
reportID := StringValue(args["report_id"])
|
||||
field("Report", reportID)
|
||||
title := StringValue(args["title"])
|
||||
field("Title", title)
|
||||
|
||||
|
|
@ -59,6 +71,7 @@ func renderVulnerabilityReport(args map[string]any, result any) string {
|
|||
}
|
||||
}
|
||||
|
||||
section("Reason", StringValue(args["update_reason"]))
|
||||
section("Description", StringValue(args["description"]))
|
||||
section("Impact", StringValue(args["impact"]))
|
||||
section("Technical Analysis", StringValue(args["technical_analysis"]))
|
||||
|
|
@ -76,8 +89,8 @@ func renderVulnerabilityReport(args map[string]any, result any) string {
|
|||
// was verified belongs next to it rather than in the artifact alone.
|
||||
section("Fix Verification", StringValue(args["fix_verification"]))
|
||||
|
||||
if title == "" {
|
||||
b.WriteString("\n " + Dim().Render("Creating report..."))
|
||||
if title == "" && reportID == "" {
|
||||
b.WriteString("\n " + Dim().Render(pending))
|
||||
}
|
||||
return "\n\n" + b.String() + "\n\n"
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -116,7 +116,7 @@ const CATEGORY_TOOLS: Record<ToolCategory, readonly string[]> = {
|
|||
filesystem: ["apply_patch", "view_image", "str_replace_editor", "list_files", "search_files"],
|
||||
// Caido proxy tools (legacy: send_request)
|
||||
proxy: ["list_requests", "view_request", "repeat_request", "list_sitemap", "view_sitemap_entry", "scope_rules", "send_request"],
|
||||
reporting: ["create_vulnerability_report", "list_reports", "get_report"],
|
||||
reporting: ["create_vulnerability_report", "update_vulnerability_report", "list_reports", "get_report"],
|
||||
thinking: ["think"],
|
||||
agents: ["create_agent", "agent_finish", "send_message_to_agent", "wait_for_agents", "view_agent_graph", "stop_agent"],
|
||||
search: ["web_search"],
|
||||
|
|
|
|||
|
|
@ -44,6 +44,56 @@ def _strix_version() -> str | None:
|
|||
return None
|
||||
|
||||
|
||||
# Content a revision may replace. The identity of the finding (id, timestamp,
|
||||
# finding_class) and its original author stay put. dependency_metadata is
|
||||
# replaced whole, so a caller carries the package identity over itself.
|
||||
UPDATABLE_REPORT_FIELDS = frozenset(
|
||||
{
|
||||
"title",
|
||||
"dependency_metadata",
|
||||
"severity",
|
||||
"description",
|
||||
"impact",
|
||||
"target",
|
||||
"technical_analysis",
|
||||
"poc_description",
|
||||
"poc_script_code",
|
||||
"remediation_steps",
|
||||
"evidence",
|
||||
"assumptions",
|
||||
"counterevidence",
|
||||
"confidence",
|
||||
"confidence_rationale",
|
||||
"severity_change_conditions",
|
||||
"fix_effort",
|
||||
"cvss",
|
||||
"cvss_breakdown",
|
||||
"endpoint",
|
||||
"method",
|
||||
"cve",
|
||||
"cwe",
|
||||
"code_locations",
|
||||
"fix_verification",
|
||||
"fix_pr_body",
|
||||
}
|
||||
)
|
||||
|
||||
_LOWERCASE_REPORT_FIELDS = frozenset({"severity", "confidence", "fix_effort"})
|
||||
|
||||
# Fields that only describe another field. A revision may raise the rating or
|
||||
# replace the locations without restating the reasoning behind the old one, and
|
||||
# that leftover reasoning then contradicts the finding it annotates
|
||||
# ("confidence: high" beside a rationale calling the evidence unconfirmed). When
|
||||
# the field they describe changes and the update carries no replacement, they
|
||||
# are dropped rather than kept.
|
||||
_DEPENDENT_REPORT_FIELDS: dict[str, tuple[str, ...]] = {
|
||||
"confidence": ("confidence_rationale",),
|
||||
"severity": ("severity_change_conditions",),
|
||||
"cvss": ("cvss_breakdown",),
|
||||
"code_locations": ("fix_verification",),
|
||||
}
|
||||
|
||||
|
||||
def _clean_title(title: str) -> str:
|
||||
"""Return a single-line finding title.
|
||||
|
||||
|
|
@ -169,6 +219,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
|
||||
|
|
@ -236,6 +287,12 @@ class ReportState:
|
|||
)
|
||||
self.vulnerability_reports = [r for r in data if isinstance(r, dict)]
|
||||
for r in self.vulnerability_reports:
|
||||
# A finding written before the class was persisted still carries the
|
||||
# metadata of its class, so name the class it always had.
|
||||
if not r.get("finding_class"):
|
||||
r["finding_class"] = (
|
||||
"dependency_cve" if r.get("dependency_metadata") else "dynamic"
|
||||
)
|
||||
title = r.get("title")
|
||||
stale_md = False
|
||||
if isinstance(title, str):
|
||||
|
|
@ -357,6 +414,100 @@ class ReportState:
|
|||
self.save_run_data()
|
||||
return report_id
|
||||
|
||||
def update_vulnerability_report(
|
||||
self,
|
||||
report_id: str,
|
||||
fields: dict[str, Any],
|
||||
*,
|
||||
update_reason: str | None = None,
|
||||
updated_by_agent_id: str | None = None,
|
||||
updated_by_agent_name: str | None = None,
|
||||
) -> dict[str, Any] | None:
|
||||
"""Apply a revision to an existing report, keeping its id.
|
||||
|
||||
A field that only describes a field this update replaces is dropped when
|
||||
the update carries no replacement for it, so the revised report cannot
|
||||
state a new rating beside the superseded reasoning for the old one.
|
||||
|
||||
Returns the revised report, or ``None`` when the id is unknown or when
|
||||
nothing in ``fields`` changes it.
|
||||
"""
|
||||
report = next((r for r in self.vulnerability_reports if r.get("id") == report_id), None)
|
||||
if report is None:
|
||||
logger.warning("cannot update unknown vulnerability report %s", report_id)
|
||||
return None
|
||||
|
||||
changed: dict[str, Any] = {}
|
||||
for key, raw_value in fields.items():
|
||||
if key not in UPDATABLE_REPORT_FIELDS or raw_value is None:
|
||||
continue
|
||||
value = raw_value
|
||||
if isinstance(value, str):
|
||||
value = _clean_title(value) if key == "title" else value.strip()
|
||||
if key in _LOWERCASE_REPORT_FIELDS:
|
||||
value = value.lower()
|
||||
if not value:
|
||||
continue
|
||||
if report.get(key) == value:
|
||||
continue
|
||||
changed[key] = value
|
||||
|
||||
superseded = {
|
||||
dependent
|
||||
for primary, dependents in _DEPENDENT_REPORT_FIELDS.items()
|
||||
if primary in changed
|
||||
for dependent in dependents
|
||||
if dependent not in changed and report.get(dependent) not in (None, "", [], {})
|
||||
}
|
||||
|
||||
if not changed and not superseded:
|
||||
logger.info("update for %s carried no new content; keeping it as is", report_id)
|
||||
return None
|
||||
|
||||
entry: dict[str, Any] = {
|
||||
"timestamp": datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S UTC"),
|
||||
"fields": sorted(changed),
|
||||
}
|
||||
if superseded:
|
||||
entry["dropped_fields"] = sorted(superseded)
|
||||
if update_reason and update_reason.strip():
|
||||
entry["reason"] = update_reason.strip()[:500]
|
||||
if updated_by_agent_id:
|
||||
entry["agent_id"] = updated_by_agent_id
|
||||
if updated_by_agent_name:
|
||||
entry["agent_name"] = updated_by_agent_name
|
||||
for key in ("severity", "cvss", "confidence"):
|
||||
if key in changed and report.get(key) is not None:
|
||||
entry[f"previous_{key}"] = report[key]
|
||||
|
||||
raw_history = report.get("update_history")
|
||||
history: list[dict[str, Any]] = (
|
||||
[e for e in raw_history if isinstance(e, dict)] if isinstance(raw_history, list) else []
|
||||
)
|
||||
history.append(entry)
|
||||
|
||||
report.update(changed)
|
||||
for dependent in superseded:
|
||||
report.pop(dependent, None)
|
||||
report["update_history"] = history
|
||||
report["updated_at"] = entry["timestamp"]
|
||||
|
||||
# The markdown on disk still shows the superseded evidence, so let the
|
||||
# writer re-render it.
|
||||
self._saved_vuln_ids.discard(report_id)
|
||||
|
||||
logger.info(
|
||||
"Updated vulnerability report %s (%s)",
|
||||
report_id,
|
||||
", ".join(entry["fields"]) or "no field replaced",
|
||||
)
|
||||
|
||||
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)
|
||||
|
||||
|
|
|
|||
|
|
@ -356,4 +356,41 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL
|
|||
lines.append(str(report["assumptions"]))
|
||||
lines.append("")
|
||||
|
||||
lines.extend(render_update_history(report.get("update_history")))
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def render_update_history(history: Any) -> list[str]:
|
||||
"""Render the audit trail of every revision a report has received."""
|
||||
if not isinstance(history, list):
|
||||
return []
|
||||
entries: list[dict[str, Any]] = [
|
||||
cast("dict[str, Any]", e) for e in history if isinstance(e, dict)
|
||||
]
|
||||
if not entries:
|
||||
return []
|
||||
|
||||
lines = ["## Update History\n"]
|
||||
for entry in entries:
|
||||
author = str(entry.get("agent_name") or entry.get("agent_id") or "an agent")
|
||||
raw_fields = entry.get("fields")
|
||||
fields: list[Any] = raw_fields if isinstance(raw_fields, list) else []
|
||||
changed = ", ".join(str(field) for field in fields)
|
||||
timestamp = str(entry.get("timestamp") or "unknown")
|
||||
lines.append(f"**{timestamp}** — {author} updated: {changed}")
|
||||
raw_dropped = entry.get("dropped_fields")
|
||||
if isinstance(raw_dropped, list) and raw_dropped:
|
||||
dropped = ", ".join(str(field) for field in raw_dropped)
|
||||
lines.append(f" Dropped as superseded: {dropped}")
|
||||
for key, label in (
|
||||
("previous_severity", "severity"),
|
||||
("previous_cvss", "CVSS"),
|
||||
("previous_confidence", "confidence"),
|
||||
):
|
||||
if entry.get(key) is not None:
|
||||
lines.append(f" Previous {label}: {entry[key]}")
|
||||
if entry.get("reason"):
|
||||
lines.append(f" Reason: {entry['reason']}")
|
||||
lines.append("")
|
||||
return lines
|
||||
|
|
|
|||
|
|
@ -12,13 +12,17 @@ import json
|
|||
import logging
|
||||
import re
|
||||
from pathlib import PurePosixPath
|
||||
from typing import Any
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from agents import RunContextWrapper, function_tool
|
||||
|
||||
from strix.tools.nullish import clean_optional
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from strix.report.state import ReportState
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
|
|
@ -257,6 +261,329 @@ def _validate_fix_verification(
|
|||
]
|
||||
|
||||
|
||||
def _finding_class_of(report: dict[str, Any]) -> str:
|
||||
"""Resolve the class of a stored finding.
|
||||
|
||||
A finding filed before ``finding_class`` was persisted still carries the
|
||||
metadata of its class. A record with dependency metadata is a dependency
|
||||
finding even when the field is absent, so read the metadata before falling
|
||||
back to dynamic.
|
||||
"""
|
||||
declared = str(report.get("finding_class") or "").lower()
|
||||
if declared:
|
||||
return declared
|
||||
if report.get("dependency_metadata"):
|
||||
return "dependency_cve"
|
||||
return "dynamic"
|
||||
|
||||
|
||||
_UPDATE_TEXT_FIELDS = (
|
||||
"title",
|
||||
"description",
|
||||
"impact",
|
||||
"target",
|
||||
"technical_analysis",
|
||||
"poc_description",
|
||||
"poc_script_code",
|
||||
"remediation_steps",
|
||||
"evidence",
|
||||
"assumptions",
|
||||
"counterevidence",
|
||||
"confidence_rationale",
|
||||
"severity_change_conditions",
|
||||
"endpoint",
|
||||
"method",
|
||||
"fix_verification",
|
||||
"fix_pr_body",
|
||||
"contextual_cvss_reasoning",
|
||||
)
|
||||
|
||||
|
||||
def _collect_update_changes( # noqa: PLR0912
|
||||
fields: dict[str, Any],
|
||||
) -> tuple[dict[str, Any], list[str]]:
|
||||
"""Validate the fields a revision replaces and return them with any errors."""
|
||||
errors: list[str] = []
|
||||
changes: dict[str, Any] = {}
|
||||
|
||||
for name in _UPDATE_TEXT_FIELDS:
|
||||
value = clean_optional(fields.get(name))
|
||||
if value is not None:
|
||||
changes[name] = value
|
||||
|
||||
confidence = clean_optional(fields.get("confidence"))
|
||||
if confidence is not None:
|
||||
confidence = confidence.lower()
|
||||
if confidence not in _VALID_CONFIDENCE:
|
||||
errors.append(
|
||||
f"Invalid confidence: {confidence!r}. Must be one of: {sorted(_VALID_CONFIDENCE)}"
|
||||
)
|
||||
else:
|
||||
changes["confidence"] = confidence
|
||||
|
||||
fix_effort = clean_optional(fields.get("fix_effort"))
|
||||
if fix_effort is not None:
|
||||
fix_effort = fix_effort.lower()
|
||||
if fix_effort not in _VALID_FIX_EFFORT:
|
||||
errors.append(
|
||||
f"Invalid fix_effort: {fix_effort!r}. Must be one of: {sorted(_VALID_FIX_EFFORT)}"
|
||||
)
|
||||
else:
|
||||
changes["fix_effort"] = fix_effort
|
||||
|
||||
breakdown = fields.get("cvss_breakdown")
|
||||
if breakdown is not None:
|
||||
breakdown_errors = _validate_cvss_breakdown(breakdown)
|
||||
errors.extend(breakdown_errors)
|
||||
if not breakdown_errors:
|
||||
try:
|
||||
cvss_score, severity, _vector = _calculate_cvss(breakdown)
|
||||
except ValueError as exc:
|
||||
errors.append(str(exc))
|
||||
else:
|
||||
# The rating belongs to the vector, so a revised vector carries
|
||||
# its own score and severity rather than leaving the old ones.
|
||||
changes["cvss_breakdown"] = breakdown
|
||||
changes["cvss"] = cvss_score
|
||||
changes["severity"] = severity
|
||||
|
||||
raw_locations = fields.get("code_locations")
|
||||
locations = _normalize_code_locations(raw_locations)
|
||||
if locations:
|
||||
errors.extend(_validate_code_locations(locations))
|
||||
errors.extend(_validate_fix_verification(locations, changes.get("fix_verification")))
|
||||
changes["code_locations"] = locations
|
||||
elif raw_locations:
|
||||
errors.append(
|
||||
"code_locations were dropped as unusable - every location needs a relative "
|
||||
"'file' and an integer 'start_line'"
|
||||
)
|
||||
|
||||
cve, cwe, identifier_errors = _validate_identifiers(
|
||||
clean_optional(fields.get("cve")), clean_optional(fields.get("cwe"))
|
||||
)
|
||||
errors.extend(identifier_errors)
|
||||
if cve:
|
||||
changes["cve"] = cve
|
||||
if cwe:
|
||||
changes["cwe"] = cwe
|
||||
|
||||
return changes, errors
|
||||
|
||||
|
||||
# Evidence that only a dynamic finding carries. A dependency finding describes a
|
||||
# package, not a request against an endpoint.
|
||||
_DYNAMIC_ONLY_UPDATE_FIELDS = (
|
||||
"endpoint",
|
||||
"method",
|
||||
"poc_description",
|
||||
"poc_script_code",
|
||||
)
|
||||
|
||||
# A dependency finding is rated in the context of the codebase that pins it, and
|
||||
# that rating is only shown with the reasoning behind it.
|
||||
_DEPENDENCY_ONLY_UPDATE_FIELDS = ("contextual_cvss_reasoning",)
|
||||
|
||||
|
||||
def _reject_cross_class_revision(
|
||||
report_id: str,
|
||||
matched_class: str,
|
||||
offending: list[str],
|
||||
) -> dict[str, Any]:
|
||||
logger.info(
|
||||
"Revision of %s carries fields (%s) a %s finding does not hold; rejecting",
|
||||
report_id,
|
||||
", ".join(offending),
|
||||
matched_class,
|
||||
)
|
||||
return {
|
||||
"success": False,
|
||||
"error": (
|
||||
f"Report '{report_id}' is a {matched_class} finding, so it cannot carry "
|
||||
f"{', '.join(offending)}. File your proof as its own vulnerability report "
|
||||
"instead of writing it onto this one."
|
||||
),
|
||||
"report_id": report_id,
|
||||
"finding_class": matched_class,
|
||||
"rejected_fields": offending,
|
||||
}
|
||||
|
||||
|
||||
def _rate_dependency_revision(
|
||||
report_id: str,
|
||||
matched: dict[str, Any],
|
||||
changes: dict[str, Any],
|
||||
) -> dict[str, Any] | None:
|
||||
"""Turn a replacement ``cvss_breakdown`` into the contextual rating of a dependency.
|
||||
|
||||
A dependency record keeps its rating as ``cvss``/``severity`` plus the
|
||||
contextual breakdown, vector and reasoning inside ``dependency_metadata``.
|
||||
The package identity in that metadata is copied over untouched. A new
|
||||
breakdown needs its own reasoning. The reasoning alone can be corrected
|
||||
when the record already carries the breakdown it explains.
|
||||
"""
|
||||
breakdown = changes.pop("cvss_breakdown", None)
|
||||
reasoning = changes.pop("contextual_cvss_reasoning", None)
|
||||
if breakdown is None and reasoning is None:
|
||||
return None
|
||||
|
||||
metadata = dict(matched.get("dependency_metadata") or {})
|
||||
if breakdown is None and not metadata.get("contextual_cvss_breakdown"):
|
||||
return {
|
||||
"success": False,
|
||||
"error": "Validation failed",
|
||||
"errors": [
|
||||
"cvss_breakdown is required: this dependency finding carries no "
|
||||
"contextual rating yet, so contextual_cvss_reasoning has nothing to explain"
|
||||
],
|
||||
"report_id": report_id,
|
||||
}
|
||||
if reasoning is None:
|
||||
return {
|
||||
"success": False,
|
||||
"error": "Validation failed",
|
||||
"errors": [
|
||||
"contextual_cvss_reasoning is required: a dependency finding is re-rated "
|
||||
"with the cvss_breakdown observed in this codebase together with the "
|
||||
"reasoning a reader can check"
|
||||
],
|
||||
"report_id": report_id,
|
||||
}
|
||||
|
||||
if breakdown is not None:
|
||||
score, _severity, vector = _calculate_cvss(breakdown)
|
||||
metadata["contextual_cvss_breakdown"] = breakdown
|
||||
metadata["contextual_cvss_score"] = score
|
||||
metadata["contextual_cvss_vector"] = vector
|
||||
metadata["contextual_cvss_reasoning"] = reasoning[:_MAX_CONTEXTUAL_REASONING_CHARS]
|
||||
changes["dependency_metadata"] = metadata
|
||||
return None
|
||||
|
||||
|
||||
def _fit_revision_to_class(
|
||||
report_state: ReportState,
|
||||
report_id: str,
|
||||
changes: dict[str, Any],
|
||||
) -> dict[str, Any] | None:
|
||||
"""Keep a revision inside the class of the finding it names.
|
||||
|
||||
A finding keeps its class and the metadata that belongs to it. Writing an
|
||||
exploit onto a dependency record would leave it carrying a package pin next
|
||||
to a request against an endpoint, so the proof belongs in its own dynamic
|
||||
finding instead. A dependency finding is still re-rated, through the
|
||||
contextual CVSS it was filed with.
|
||||
"""
|
||||
matched = next(
|
||||
(r for r in report_state.get_existing_vulnerabilities() if r.get("id") == report_id),
|
||||
None,
|
||||
)
|
||||
if matched is None:
|
||||
return None
|
||||
|
||||
matched_class = _finding_class_of(matched)
|
||||
foreign = (
|
||||
_DEPENDENCY_ONLY_UPDATE_FIELDS
|
||||
if matched_class == "dynamic"
|
||||
else _DYNAMIC_ONLY_UPDATE_FIELDS
|
||||
)
|
||||
offending = [name for name in foreign if name in changes]
|
||||
if offending:
|
||||
return _reject_cross_class_revision(report_id, matched_class, offending)
|
||||
if matched_class == "dynamic":
|
||||
return None
|
||||
return _rate_dependency_revision(report_id, matched, changes)
|
||||
|
||||
|
||||
def _read_revision(
|
||||
report_id: str, update_reason: str, fields: dict[str, Any]
|
||||
) -> tuple[dict[str, Any], dict[str, Any] | None]:
|
||||
"""Return the changes a revision asks for, or the reason it cannot be acted on."""
|
||||
if not report_id or not str(update_reason or "").strip():
|
||||
missing = "report_id" if not report_id else "update_reason"
|
||||
return {}, {
|
||||
"success": False,
|
||||
"error": (
|
||||
f"{missing} cannot be empty - name the report you are revising and state "
|
||||
"what you learned that it does not yet carry"
|
||||
),
|
||||
}
|
||||
|
||||
changes, errors = _collect_update_changes(fields)
|
||||
if errors:
|
||||
return {}, {"success": False, "error": "Validation failed", "errors": errors}
|
||||
if not changes:
|
||||
return {}, {
|
||||
"success": False,
|
||||
"error": "No fields to update - pass at least one field you want to replace",
|
||||
}
|
||||
return changes, None
|
||||
|
||||
|
||||
def _do_update(
|
||||
*,
|
||||
report_id: str,
|
||||
update_reason: str,
|
||||
fields: dict[str, Any],
|
||||
agent_id: str | None = None,
|
||||
agent_name: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Apply an agent's own revision to a report it can name.
|
||||
|
||||
Editing a finding is its own operation and the only way a filed finding
|
||||
changes. Deduplication never reaches this path: it only decides whether a
|
||||
new candidate is a finding already on file.
|
||||
"""
|
||||
report_id = (report_id or "").strip()
|
||||
changes, rejection = _read_revision(report_id, update_reason, fields)
|
||||
if rejection is not None:
|
||||
return rejection
|
||||
|
||||
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 unavailable - no reports have been filed yet",
|
||||
}
|
||||
|
||||
class_error = _fit_revision_to_class(report_state, report_id, changes)
|
||||
if class_error is not None:
|
||||
return class_error
|
||||
|
||||
updated = report_state.update_vulnerability_report(
|
||||
report_id,
|
||||
changes,
|
||||
update_reason=update_reason,
|
||||
updated_by_agent_id=agent_id,
|
||||
updated_by_agent_name=agent_name,
|
||||
)
|
||||
if updated is None:
|
||||
known = [r.get("id") for r in report_state.get_existing_vulnerabilities()]
|
||||
if report_id not in known:
|
||||
error = f"Report with id '{report_id}' not found"
|
||||
else:
|
||||
error = f"Report '{report_id}' already says this - nothing in your update changes it"
|
||||
return {"success": False, "error": error, "report_id": report_id}
|
||||
|
||||
logger.info(
|
||||
"Vulnerability report %s revised by its author: severity=%s cvss=%s fields=%s",
|
||||
report_id,
|
||||
updated.get("severity"),
|
||||
updated.get("cvss"),
|
||||
", ".join(sorted(changes)),
|
||||
)
|
||||
return {
|
||||
"success": True,
|
||||
"action": "updated",
|
||||
"message": f"Report '{report_id}' now carries your revision. Do not file it again.",
|
||||
"report_id": report_id,
|
||||
"updated_fields": sorted(changes),
|
||||
"severity": updated.get("severity"),
|
||||
"cvss_score": updated.get("cvss"),
|
||||
}
|
||||
|
||||
|
||||
async def _do_create(
|
||||
*,
|
||||
title: str,
|
||||
|
|
@ -359,9 +686,37 @@ async def _do_create(
|
|||
"endpoint": endpoint,
|
||||
"method": method,
|
||||
}
|
||||
report_fields: dict[str, Any] = {
|
||||
"title": title,
|
||||
"description": description,
|
||||
"severity": severity,
|
||||
"impact": impact,
|
||||
"target": target,
|
||||
"technical_analysis": technical_analysis,
|
||||
"poc_description": poc_description,
|
||||
"poc_script_code": poc_script_code,
|
||||
"remediation_steps": remediation_steps,
|
||||
"evidence": evidence,
|
||||
"assumptions": assumptions,
|
||||
"counterevidence": counterevidence,
|
||||
"confidence": confidence,
|
||||
"confidence_rationale": confidence_rationale,
|
||||
"severity_change_conditions": severity_change_conditions,
|
||||
"fix_effort": fix_effort,
|
||||
"cvss": cvss_score,
|
||||
"cvss_breakdown": cvss_breakdown,
|
||||
"endpoint": endpoint,
|
||||
"method": method,
|
||||
"cve": cve,
|
||||
"cwe": cwe,
|
||||
"code_locations": parsed_locations,
|
||||
"fix_verification": fix_verification,
|
||||
"fix_pr_body": fix_pr_body,
|
||||
}
|
||||
|
||||
dedupe = await check_duplicate(candidate, existing)
|
||||
if dedupe.get("is_duplicate"):
|
||||
duplicate_id = dedupe.get("duplicate_id", "")
|
||||
duplicate_id = str(dedupe.get("duplicate_id") or "")
|
||||
duplicate_title = next(
|
||||
(r.get("title", "Unknown") for r in existing if r.get("id") == duplicate_id),
|
||||
"",
|
||||
|
|
@ -379,31 +734,7 @@ async def _do_create(
|
|||
}
|
||||
|
||||
report_id = report_state.add_vulnerability_report(
|
||||
title=title,
|
||||
description=description,
|
||||
severity=severity,
|
||||
impact=impact,
|
||||
target=target,
|
||||
technical_analysis=technical_analysis,
|
||||
poc_description=poc_description,
|
||||
poc_script_code=poc_script_code,
|
||||
remediation_steps=remediation_steps,
|
||||
evidence=evidence,
|
||||
assumptions=assumptions,
|
||||
counterevidence=counterevidence,
|
||||
confidence=confidence,
|
||||
confidence_rationale=confidence_rationale,
|
||||
severity_change_conditions=severity_change_conditions,
|
||||
fix_effort=fix_effort,
|
||||
cvss=cvss_score,
|
||||
cvss_breakdown=cvss_breakdown,
|
||||
endpoint=endpoint,
|
||||
method=method,
|
||||
cve=cve,
|
||||
cwe=cwe,
|
||||
code_locations=parsed_locations,
|
||||
fix_verification=fix_verification,
|
||||
fix_pr_body=fix_pr_body,
|
||||
**report_fields,
|
||||
agent_id=agent_id if isinstance(agent_id, str) else None,
|
||||
agent_name=agent_name if isinstance(agent_name, str) else None,
|
||||
)
|
||||
|
|
@ -512,7 +843,9 @@ 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.
|
||||
areas. When you have learned something a filed finding does not yet
|
||||
carry, revise that finding with ``update_vulnerability_report``
|
||||
instead of filing this report again.
|
||||
|
||||
**Counterevidence pass (required before filing)**: actively build the
|
||||
strongest case that this finding is NOT exploitable, or less severe
|
||||
|
|
@ -886,6 +1219,147 @@ async def create_vulnerability_report(
|
|||
return json.dumps(result, ensure_ascii=False, default=str)
|
||||
|
||||
|
||||
@function_tool(timeout=60, 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,
|
||||
target: 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,
|
||||
counterevidence: str | None = None,
|
||||
confidence: str | None = None,
|
||||
confidence_rationale: str | None = None,
|
||||
severity_change_conditions: str | None = None,
|
||||
fix_effort: str | None = None,
|
||||
cvss_breakdown: dict[str, str] | None = None,
|
||||
endpoint: str | None = None,
|
||||
method: str | None = None,
|
||||
cve: str | None = None,
|
||||
cwe: str | None = None,
|
||||
code_locations: list[dict[str, Any]] | None = None,
|
||||
fix_verification: str | None = None,
|
||||
fix_pr_body: str | None = None,
|
||||
contextual_cvss_reasoning: str | None = None,
|
||||
) -> str:
|
||||
"""Revise a vulnerability report that is already filed, keeping its id.
|
||||
|
||||
Use this when you learn something a filed finding does not yet carry:
|
||||
|
||||
- You built the working exploit after filing the finding on static
|
||||
evidence, so the PoC and the confidence change.
|
||||
- You chained the finding with another one and the real impact is
|
||||
higher, so the impact narrative and the CVSS vector change.
|
||||
- Further testing narrowed or weakened the finding, so the severity
|
||||
must come down.
|
||||
- Counterevidence, remediation, or a code location was wrong or
|
||||
incomplete.
|
||||
|
||||
This is not deduplication. You do not need a duplicate verdict to
|
||||
revise your own finding, and you must not file a second report for a
|
||||
finding you can revise. Call ``list_reports`` or ``get_report`` first
|
||||
to find the id and read what the report already says.
|
||||
|
||||
Pass only the fields you want to replace. Every other field stays as
|
||||
it is. Reporting rules of ``create_vulnerability_report`` apply to
|
||||
every field you pass, including the markdown and tone rules.
|
||||
|
||||
Notes on specific fields:
|
||||
|
||||
- ``cvss_breakdown`` replaces the whole vector. The score and the
|
||||
severity are recalculated from it, so pass all 8 metrics. On a
|
||||
dependency finding it replaces the contextual rating and needs
|
||||
``contextual_cvss_reasoning`` with it. Pass the reasoning alone to
|
||||
correct only the explanation of the rating already on file.
|
||||
- A dependency finding never carries ``endpoint``, ``method`` or a PoC.
|
||||
File a proven exploit of the package as its own report.
|
||||
- A field that only explains another field is dropped when the field
|
||||
it explains changes and you pass no replacement. Pass
|
||||
``confidence_rationale`` with a new ``confidence``, and
|
||||
``severity_change_conditions`` with a new ``cvss_breakdown``.
|
||||
- ``code_locations`` replaces the whole list. A location carrying
|
||||
``fix_after`` needs ``fix_verification``.
|
||||
|
||||
The report keeps its id, its original author, and its filing time. The
|
||||
revision is recorded in the report as update history, so state the
|
||||
reason plainly.
|
||||
|
||||
Args:
|
||||
report_id: Id of the report to revise (format ``vuln-NNNN``).
|
||||
update_reason: What you learned that the report does not yet
|
||||
carry, in one or two sentences.
|
||||
title: Replacement title.
|
||||
description: Replacement overview.
|
||||
impact: Replacement impact narrative.
|
||||
target: Replacement affected asset.
|
||||
technical_analysis: Replacement technical details.
|
||||
poc_description: Replacement PoC steps (no code).
|
||||
poc_script_code: Replacement exploit script or payload.
|
||||
remediation_steps: Replacement remediation prose (no code).
|
||||
evidence: Replacement evidence.
|
||||
assumptions: Replacement exploitability prerequisites.
|
||||
counterevidence: Replacement case against the finding.
|
||||
confidence: ``high`` / ``medium`` / ``low``.
|
||||
confidence_rationale: The gap behind a confidence below ``high``.
|
||||
severity_change_conditions: What would move the severity now.
|
||||
fix_effort: ``trivial`` / ``low`` / ``medium`` / ``high``.
|
||||
cvss_breakdown: All 8 CVSS metrics. Replaces the score and the
|
||||
severity too.
|
||||
endpoint: Replacement endpoint.
|
||||
method: Replacement HTTP method.
|
||||
cve: Replacement CVE id.
|
||||
cwe: Replacement CWE id.
|
||||
code_locations: Replacement code locations.
|
||||
fix_verification: Verification statement for an applyable fix.
|
||||
fix_pr_body: Replacement fix PR body.
|
||||
contextual_cvss_reasoning: Dependency findings only. What you
|
||||
observed in this codebase that justifies the contextual
|
||||
``cvss_breakdown``.
|
||||
"""
|
||||
agent_id, agent_name = _caller_identity(ctx)
|
||||
result = await asyncio.to_thread(
|
||||
_do_update,
|
||||
report_id=report_id,
|
||||
update_reason=update_reason,
|
||||
fields={
|
||||
"title": title,
|
||||
"description": description,
|
||||
"impact": impact,
|
||||
"target": target,
|
||||
"technical_analysis": technical_analysis,
|
||||
"poc_description": poc_description,
|
||||
"poc_script_code": poc_script_code,
|
||||
"remediation_steps": remediation_steps,
|
||||
"evidence": evidence,
|
||||
"assumptions": assumptions,
|
||||
"counterevidence": counterevidence,
|
||||
"confidence": confidence,
|
||||
"confidence_rationale": confidence_rationale,
|
||||
"severity_change_conditions": severity_change_conditions,
|
||||
"fix_effort": fix_effort,
|
||||
"cvss_breakdown": cvss_breakdown,
|
||||
"endpoint": endpoint,
|
||||
"method": method,
|
||||
"cve": cve,
|
||||
"cwe": cwe,
|
||||
"code_locations": code_locations,
|
||||
"fix_verification": fix_verification,
|
||||
"fix_pr_body": fix_pr_body,
|
||||
"contextual_cvss_reasoning": contextual_cvss_reasoning,
|
||||
},
|
||||
agent_id=agent_id,
|
||||
agent_name=agent_name,
|
||||
)
|
||||
return json.dumps(result, ensure_ascii=False, default=str)
|
||||
|
||||
|
||||
_DEP_SEVERITY_FROM_CVSS = {
|
||||
(9.0, 10.0): "critical",
|
||||
(7.0, 9.0): "high",
|
||||
|
|
|
|||
|
|
@ -73,6 +73,41 @@ def test_hydrate_from_run_dir_strips_control_chars_from_title(
|
|||
assert md_path.read_text(encoding="utf-8").startswith("# XSS in search form\n")
|
||||
|
||||
|
||||
def test_hydrate_names_the_class_a_legacy_record_always_had(
|
||||
report_state: ReportState,
|
||||
) -> None:
|
||||
# A run started before the class was persisted still holds the package metadata
|
||||
# of a dependency finding, and resume must not read it as a dynamic one.
|
||||
(report_state.get_run_dir() / "vulnerabilities.json").write_text(
|
||||
json.dumps(
|
||||
[
|
||||
{
|
||||
"id": "vuln-0001",
|
||||
"title": "Directus 11.5.1 is affected by CVE-2025-55746",
|
||||
"severity": "medium",
|
||||
"timestamp": "2026-01-01 00:00:00 UTC",
|
||||
"dependency_metadata": {
|
||||
"package_name": "directus",
|
||||
"installed_version": "11.5.1",
|
||||
},
|
||||
},
|
||||
{
|
||||
"id": "vuln-0002",
|
||||
"title": "Reflected XSS in search",
|
||||
"severity": "medium",
|
||||
"timestamp": "2026-01-01 00:00:00 UTC",
|
||||
},
|
||||
]
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
report_state.hydrate_from_run_dir()
|
||||
|
||||
assert report_state.vulnerability_reports[0]["finding_class"] == "dependency_cve"
|
||||
assert report_state.vulnerability_reports[1]["finding_class"] == "dynamic"
|
||||
|
||||
|
||||
def _seed(state: ReportState) -> None:
|
||||
state.add_vulnerability_report(
|
||||
title="Reflected XSS in search",
|
||||
|
|
|
|||
|
|
@ -16,8 +16,10 @@ from strix.tools.finish.tool import finish_scan
|
|||
from strix.tools.reporting.tool import (
|
||||
_do_create,
|
||||
_do_create_dependency,
|
||||
_do_update,
|
||||
create_dependency_report,
|
||||
create_vulnerability_report,
|
||||
update_vulnerability_report,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -1251,3 +1253,507 @@ async def test_dependency_report_rejects_contextual_breakdown_without_reasoning(
|
|||
assert result["success"] is False
|
||||
assert any("contextual_cvss_reasoning is required" in error for error in result["errors"])
|
||||
assert report_state.vulnerability_reports == []
|
||||
|
||||
|
||||
_CONFIRMED_KWARGS: dict[str, Any] = {
|
||||
"title": "Unauthenticated file write on /files/{id}",
|
||||
"description": "A multipart PATCH writes attacker content before the permission check.",
|
||||
"impact": "Any anonymous user overwrites stored files and serves attacker content.",
|
||||
"target": "https://cms.example.com",
|
||||
"technical_analysis": "disk.write runs before the authorization guard.",
|
||||
"poc_description": "1. PATCH /files/<uuid> with a multipart body as an anonymous user.",
|
||||
"poc_script_code": "PATCH /files/2f1c HTTP/1.1\n\n--x\nowned\n--x--",
|
||||
"remediation_steps": "Authorize before the write.",
|
||||
"evidence": "The stored file returns the injected payload after the 403 response.",
|
||||
"assumptions": "Assumes the uuid of one existing file is known.",
|
||||
"counterevidence": "The endpoint answers 403, yet the write already landed.",
|
||||
"confidence": "HIGH",
|
||||
"confidence_rationale": "The write was observed end to end against the live host.",
|
||||
"severity_change_conditions": "A guard before disk.write would remove the impact.",
|
||||
"fix_effort": "MEDIUM",
|
||||
"cvss_breakdown": _CVSS,
|
||||
"endpoint": "/files/{id}",
|
||||
"method": "PATCH",
|
||||
"cve": "CVE-2025-55746",
|
||||
"cwe": "CWE-863",
|
||||
"code_locations": None,
|
||||
}
|
||||
|
||||
|
||||
def _seed_weak_report(report_state: ReportState) -> None:
|
||||
"""A version-based, unproven entry for the same issue, as an earlier agent files it."""
|
||||
report_state.vulnerability_reports.append(
|
||||
{
|
||||
"id": "vuln-0009",
|
||||
"title": "Directus 11.5.1 exposed on public host (in scope for CVE-2025-55746)",
|
||||
"severity": "medium",
|
||||
"timestamp": "2026-01-01 00:00:00 UTC",
|
||||
"description": "The banner reports a version affected by CVE-2025-55746.",
|
||||
"target": "https://cms.example.com",
|
||||
"confidence": "low",
|
||||
"evidence": "The version banner only.",
|
||||
"cvss": 5.3,
|
||||
"finding_class": "dynamic",
|
||||
"agent_id": "aaaa1111",
|
||||
}
|
||||
)
|
||||
report_state._saved_vuln_ids.add("vuln-0009")
|
||||
|
||||
|
||||
async def test_duplicate_verdict_rejects_without_touching_the_existing_report(
|
||||
report_state: ReportState, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Deduplication only answers identity. A duplicate is rejected and points at the
|
||||
finding it matched; revising that finding is a separate, explicit operation."""
|
||||
_seed_weak_report(report_state)
|
||||
|
||||
async def fake_check_duplicate(
|
||||
_candidate: dict[str, Any], _existing: list[dict[str, Any]]
|
||||
) -> dict[str, Any]:
|
||||
return {
|
||||
"is_duplicate": True,
|
||||
"duplicate_id": "vuln-0009",
|
||||
"confidence": 0.9,
|
||||
"reason": "Same root cause on the same endpoint.",
|
||||
}
|
||||
|
||||
monkeypatch.setattr("strix.report.dedupe.check_duplicate", fake_check_duplicate)
|
||||
|
||||
result = await _do_create(**_CONFIRMED_KWARGS, agent_id="834f79fb", agent_name="Validation")
|
||||
|
||||
assert result["success"] is False
|
||||
assert result["duplicate_of"] == "vuln-0009"
|
||||
assert "action" not in result
|
||||
assert len(report_state.vulnerability_reports) == 1
|
||||
report = report_state.vulnerability_reports[0]
|
||||
assert report["severity"] == "medium", "a duplicate verdict never edits the matched finding"
|
||||
assert "poc_script_code" not in report
|
||||
assert "update_history" not in report
|
||||
|
||||
|
||||
def test_update_vulnerability_report_records_chained_impact(report_state: ReportState) -> None:
|
||||
"""Attack chaining raises the impact of a finding already on file."""
|
||||
_seed_weak_report(report_state)
|
||||
|
||||
updated = report_state.update_vulnerability_report(
|
||||
"vuln-0009",
|
||||
{
|
||||
"severity": "CRITICAL",
|
||||
"cvss": 9.8,
|
||||
"impact": "The overwritten file loads in an admin session and takes over the account.",
|
||||
"id": "vuln-9999",
|
||||
"finding_class": "static",
|
||||
},
|
||||
update_reason="A chained admin takeover follows the file write.",
|
||||
)
|
||||
|
||||
assert updated is not None
|
||||
assert updated["id"] == "vuln-0009", "identity fields are not updatable"
|
||||
assert updated["finding_class"] == "dynamic"
|
||||
assert updated["severity"] == "critical"
|
||||
assert updated["updated_at"]
|
||||
assert report_state.update_vulnerability_report("vuln-0404", {"severity": "high"}) is None
|
||||
|
||||
|
||||
def test_update_vulnerability_report_ignores_identical_content(report_state: ReportState) -> None:
|
||||
_seed_weak_report(report_state)
|
||||
assert report_state.update_vulnerability_report("vuln-0009", {"severity": "medium"}) is None
|
||||
assert "update_history" not in report_state.vulnerability_reports[0]
|
||||
|
||||
|
||||
def test_update_drops_reasoning_left_behind_by_the_field_it_describes(
|
||||
report_state: ReportState,
|
||||
) -> None:
|
||||
"""A rating the update replaces must not keep the rationale for the old one."""
|
||||
_seed_weak_report(report_state)
|
||||
report = report_state.vulnerability_reports[0]
|
||||
report["confidence_rationale"] = "Nothing was executed; the version banner is the only signal."
|
||||
report["cvss_breakdown"] = {"attack_vector": "network", "user_interaction": "required"}
|
||||
report["severity_change_conditions"] = "Confirming the write would raise this."
|
||||
|
||||
updated = report_state.update_vulnerability_report(
|
||||
"vuln-0009",
|
||||
{
|
||||
"confidence": "high",
|
||||
"severity": "critical",
|
||||
"cvss": 9.8,
|
||||
"severity_change_conditions": "A guard before the write would remove the impact.",
|
||||
},
|
||||
)
|
||||
|
||||
assert updated is not None
|
||||
assert "confidence_rationale" not in updated, "the superseded rationale must not survive"
|
||||
assert "cvss_breakdown" not in updated
|
||||
assert updated["severity_change_conditions"].startswith("A guard"), (
|
||||
"a replacement the update supplies is kept, not dropped"
|
||||
)
|
||||
assert updated["update_history"][0]["dropped_fields"] == [
|
||||
"confidence_rationale",
|
||||
"cvss_breakdown",
|
||||
]
|
||||
|
||||
run_dir = report_state._run_dir
|
||||
assert run_dir is not None
|
||||
markdown = (run_dir / "vulnerabilities" / "vuln-0009.md").read_text(encoding="utf-8")
|
||||
assert "version banner is the only signal" not in markdown
|
||||
assert "Dropped as superseded: confidence_rationale, cvss_breakdown" in markdown
|
||||
|
||||
|
||||
def test_agent_revises_its_own_report_without_a_duplicate_verdict(
|
||||
report_state: ReportState,
|
||||
) -> None:
|
||||
"""Editing a finding is its own operation: no dedupe verdict is involved."""
|
||||
_seed_weak_report(report_state)
|
||||
|
||||
result = _do_update(
|
||||
report_id="vuln-0009",
|
||||
update_reason="An unauthenticated PATCH wrote the file, so the finding is confirmed.",
|
||||
fields={
|
||||
"poc_script_code": "PATCH /files/2f1c HTTP/1.1",
|
||||
"confidence": "HIGH",
|
||||
"confidence_rationale": "The write was replayed twice.",
|
||||
"cvss_breakdown": _CVSS,
|
||||
"severity_change_conditions": "A guard before the write would remove the impact.",
|
||||
},
|
||||
agent_id="834f79fb",
|
||||
agent_name="Directus CVE-2025-55746 Validation Agent",
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["action"] == "updated"
|
||||
assert result["severity"] == "critical"
|
||||
assert result["cvss_score"] == pytest.approx(9.8)
|
||||
assert "cvss" in result["updated_fields"], "a new vector carries its own score"
|
||||
assert len(report_state.vulnerability_reports) == 1
|
||||
|
||||
report = report_state.vulnerability_reports[0]
|
||||
assert report["id"] == "vuln-0009"
|
||||
assert report["confidence"] == "high"
|
||||
assert report["agent_id"] == "aaaa1111", "the original reporter stays on the finding"
|
||||
history = report["update_history"]
|
||||
assert history[0]["agent_name"] == "Directus CVE-2025-55746 Validation Agent"
|
||||
assert history[0]["reason"].startswith("An unauthenticated PATCH")
|
||||
|
||||
run_dir = report_state._run_dir
|
||||
assert run_dir is not None
|
||||
markdown = (run_dir / "vulnerabilities" / "vuln-0009.md").read_text(encoding="utf-8")
|
||||
assert "PATCH /files/2f1c" in markdown
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("report_id", "update_reason", "fields", "expected"),
|
||||
[
|
||||
(" ", "reason", {"impact": "x"}, "report_id cannot be empty"),
|
||||
("vuln-0009", " ", {"impact": "x"}, "update_reason cannot be empty"),
|
||||
("vuln-0009", "reason", {}, "No fields to update"),
|
||||
("vuln-0404", "reason", {"impact": "x"}, "not found"),
|
||||
],
|
||||
)
|
||||
def test_update_rejects_a_call_it_cannot_act_on(
|
||||
report_state: ReportState,
|
||||
report_id: str,
|
||||
update_reason: str,
|
||||
fields: dict[str, Any],
|
||||
expected: str,
|
||||
) -> None:
|
||||
_seed_weak_report(report_state)
|
||||
|
||||
result = _do_update(report_id=report_id, update_reason=update_reason, fields=fields)
|
||||
|
||||
assert result["success"] is False
|
||||
assert expected in result["error"]
|
||||
assert "update_history" not in report_state.vulnerability_reports[0]
|
||||
|
||||
|
||||
def test_update_reports_every_invalid_field_at_once(report_state: ReportState) -> None:
|
||||
_seed_weak_report(report_state)
|
||||
|
||||
result = _do_update(
|
||||
report_id="vuln-0009",
|
||||
update_reason="Raising the rating.",
|
||||
fields={
|
||||
"confidence": "very high",
|
||||
"fix_effort": "weeks",
|
||||
"cvss_breakdown": {**_CVSS, "attack_vector": "X"},
|
||||
"cve": "CVE-BAD",
|
||||
},
|
||||
)
|
||||
|
||||
assert result["success"] is False
|
||||
joined = " ".join(result["errors"])
|
||||
assert "confidence" in joined
|
||||
assert "fix_effort" in joined
|
||||
assert "attack_vector" in joined
|
||||
assert "CVE" in joined
|
||||
assert report_state.vulnerability_reports[0]["confidence"] == "low", "nothing was applied"
|
||||
|
||||
|
||||
def test_update_wants_verification_for_a_fix_it_would_apply(
|
||||
report_state: ReportState,
|
||||
) -> None:
|
||||
_seed_weak_report(report_state)
|
||||
|
||||
result = _do_update(
|
||||
report_id="vuln-0009",
|
||||
update_reason="Adding the file the write lands in.",
|
||||
fields={
|
||||
"code_locations": [
|
||||
{
|
||||
"file": "api/src/controllers/files.ts",
|
||||
"start_line": 42,
|
||||
"fix_before": "await storage.write(id, body)",
|
||||
"fix_after": "await assertPermission(req); await storage.write(id, body)",
|
||||
}
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
assert result["success"] is False
|
||||
assert any("fix_verification" in error for error in result["errors"])
|
||||
|
||||
|
||||
def test_update_says_so_when_the_report_already_carries_it(
|
||||
report_state: ReportState,
|
||||
) -> None:
|
||||
_seed_weak_report(report_state)
|
||||
|
||||
result = _do_update(
|
||||
report_id="vuln-0009",
|
||||
update_reason="Restating the severity.",
|
||||
fields={"confidence": "low"},
|
||||
)
|
||||
|
||||
assert result["success"] is False
|
||||
assert "already says this" in result["error"]
|
||||
assert result["report_id"] == "vuln-0009"
|
||||
|
||||
|
||||
def test_update_tool_asks_for_the_report_and_the_reason() -> None:
|
||||
schema = update_vulnerability_report.params_json_schema
|
||||
assert set(schema["required"]) >= {"report_id", "update_reason"}
|
||||
assert "cvss_breakdown" in schema["properties"]
|
||||
assert "id" not in schema["properties"], "identity fields are not editable"
|
||||
description = update_vulnerability_report.description
|
||||
assert "not deduplication" in description
|
||||
|
||||
|
||||
def test_update_keeps_an_exploit_out_of_a_dependency_finding(report_state: ReportState) -> None:
|
||||
"""A dependency record is rated from its advisory, so a revision must not write a
|
||||
PoC and a dynamic rating onto it. The proof belongs in its own finding."""
|
||||
_seed_weak_report(report_state)
|
||||
dependency_report = report_state.vulnerability_reports[0]
|
||||
dependency_report["finding_class"] = "dependency_cve"
|
||||
dependency_report["dependency_metadata"] = {
|
||||
"package_name": "directus",
|
||||
"installed_version": "11.5.1",
|
||||
}
|
||||
|
||||
result = _do_update(
|
||||
report_id="vuln-0009",
|
||||
update_reason="An unauthenticated PATCH wrote the file.",
|
||||
fields={
|
||||
"poc_script_code": "PATCH /files/2f1c HTTP/1.1",
|
||||
"endpoint": "/files/{uuid}",
|
||||
"cvss_breakdown": _CVSS,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["success"] is False
|
||||
assert "dependency_cve" in result["error"]
|
||||
assert set(result["rejected_fields"]) == {"endpoint", "poc_script_code"}
|
||||
assert dependency_report["severity"] == "medium"
|
||||
assert "poc_script_code" not in dependency_report
|
||||
assert "update_history" not in dependency_report
|
||||
|
||||
|
||||
def _seed_dependency_report(report_state: ReportState) -> dict[str, Any]:
|
||||
_seed_weak_report(report_state)
|
||||
dependency_report = report_state.vulnerability_reports[0]
|
||||
dependency_report["finding_class"] = "dependency_cve"
|
||||
dependency_report["dependency_metadata"] = {
|
||||
"package_name": "directus",
|
||||
"installed_version": "11.5.1",
|
||||
"manifest_path": "package-lock.json",
|
||||
"advisory_cvss": 9.8,
|
||||
"contextual_cvss_breakdown": {**_CVSS, "confidentiality": "L"},
|
||||
"contextual_cvss_score": 5.3,
|
||||
"contextual_cvss_vector": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:L/I:N/A:N",
|
||||
"contextual_cvss_reasoning": "The vulnerable API is imported but never called.",
|
||||
}
|
||||
return dependency_report
|
||||
|
||||
|
||||
def test_update_re_rates_a_dependency_finding_through_its_contextual_cvss(
|
||||
report_state: ReportState,
|
||||
) -> None:
|
||||
"""A dependency finding is rated in the context of the codebase. A revised
|
||||
breakdown replaces that contextual rating, with the reasoning a reader can
|
||||
check, and leaves the package identity alone."""
|
||||
dependency_report = _seed_dependency_report(report_state)
|
||||
|
||||
result = _do_update(
|
||||
report_id="vuln-0009",
|
||||
update_reason="A call path from the upload handler to the vulnerable API was found.",
|
||||
fields={
|
||||
"cvss_breakdown": _CVSS,
|
||||
"contextual_cvss_reasoning": (
|
||||
"routes/upload.ts:88 reaches the affected parser with user input."
|
||||
),
|
||||
},
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["severity"] == "critical"
|
||||
assert dependency_report["severity"] == "critical"
|
||||
assert dependency_report["cvss"] == 9.8
|
||||
assert "cvss_breakdown" not in dependency_report
|
||||
assert "contextual_cvss_reasoning" not in dependency_report
|
||||
metadata = dependency_report["dependency_metadata"]
|
||||
assert metadata["package_name"] == "directus"
|
||||
assert metadata["installed_version"] == "11.5.1"
|
||||
assert metadata["manifest_path"] == "package-lock.json"
|
||||
assert metadata["advisory_cvss"] == 9.8
|
||||
assert metadata["contextual_cvss_breakdown"] == _CVSS
|
||||
assert metadata["contextual_cvss_score"] == 9.8
|
||||
assert metadata["contextual_cvss_vector"] == "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H"
|
||||
assert metadata["contextual_cvss_reasoning"].startswith("routes/upload.ts:88")
|
||||
assert dependency_report["finding_class"] == "dependency_cve"
|
||||
history = dependency_report["update_history"]
|
||||
assert history[-1]["previous_severity"] == "medium"
|
||||
assert set(history[-1]["fields"]) == {"cvss", "dependency_metadata", "severity"}
|
||||
|
||||
|
||||
def test_update_wants_the_reasoning_behind_a_dependency_re_rating(
|
||||
report_state: ReportState,
|
||||
) -> None:
|
||||
dependency_report = _seed_dependency_report(report_state)
|
||||
|
||||
result = _do_update(
|
||||
report_id="vuln-0009",
|
||||
update_reason="The parser is reachable.",
|
||||
fields={"cvss_breakdown": _CVSS},
|
||||
)
|
||||
|
||||
assert result["success"] is False
|
||||
assert any("contextual_cvss_reasoning" in error for error in result["errors"])
|
||||
assert dependency_report["severity"] == "medium"
|
||||
assert dependency_report["dependency_metadata"]["contextual_cvss_score"] == 5.3
|
||||
assert "update_history" not in dependency_report
|
||||
|
||||
|
||||
def test_update_corrects_the_reasoning_behind_a_dependency_rating_alone(
|
||||
report_state: ReportState,
|
||||
) -> None:
|
||||
"""The rating on file stays; only its explanation is replaced."""
|
||||
dependency_report = _seed_dependency_report(report_state)
|
||||
|
||||
result = _do_update(
|
||||
report_id="vuln-0009",
|
||||
update_reason="The reasoning named the wrong module.",
|
||||
fields={"contextual_cvss_reasoning": "lib/parser.ts imports it; no call site reaches it."},
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["updated_fields"] == ["dependency_metadata"]
|
||||
assert dependency_report["severity"] == "medium"
|
||||
assert dependency_report["cvss"] == 5.3
|
||||
metadata = dependency_report["dependency_metadata"]
|
||||
assert metadata["contextual_cvss_breakdown"] == {**_CVSS, "confidentiality": "L"}
|
||||
assert metadata["contextual_cvss_score"] == 5.3
|
||||
assert metadata["contextual_cvss_reasoning"].startswith("lib/parser.ts")
|
||||
assert metadata["package_name"] == "directus"
|
||||
|
||||
|
||||
def test_update_wants_a_rating_before_reasoning_about_one(
|
||||
report_state: ReportState,
|
||||
) -> None:
|
||||
_seed_weak_report(report_state)
|
||||
dependency_report = report_state.vulnerability_reports[0]
|
||||
dependency_report["finding_class"] = "dependency_cve"
|
||||
dependency_report["dependency_metadata"] = {
|
||||
"package_name": "directus",
|
||||
"installed_version": "11.5.1",
|
||||
}
|
||||
|
||||
result = _do_update(
|
||||
report_id="vuln-0009",
|
||||
update_reason="Explaining the rating.",
|
||||
fields={"contextual_cvss_reasoning": "Reachable."},
|
||||
)
|
||||
|
||||
assert result["success"] is False
|
||||
assert any("cvss_breakdown is required" in error for error in result["errors"])
|
||||
assert "contextual_cvss_reasoning" not in dependency_report["dependency_metadata"]
|
||||
|
||||
|
||||
def test_update_keeps_contextual_reasoning_off_a_dynamic_finding(
|
||||
report_state: ReportState,
|
||||
) -> None:
|
||||
_seed_weak_report(report_state)
|
||||
|
||||
result = _do_update(
|
||||
report_id="vuln-0009",
|
||||
update_reason="Re-rating.",
|
||||
fields={"cvss_breakdown": _CVSS, "contextual_cvss_reasoning": "Reachable."},
|
||||
)
|
||||
|
||||
assert result["success"] is False
|
||||
assert result["rejected_fields"] == ["contextual_cvss_reasoning"]
|
||||
assert report_state.vulnerability_reports[0]["severity"] == "medium"
|
||||
|
||||
|
||||
def test_update_reads_a_legacy_dependency_record_by_its_metadata(
|
||||
report_state: ReportState,
|
||||
) -> None:
|
||||
"""A dependency finding filed before finding_class was persisted still carries
|
||||
package metadata, so its class is read from that, not defaulted to dynamic."""
|
||||
_seed_weak_report(report_state)
|
||||
dependency_report = report_state.vulnerability_reports[0]
|
||||
dependency_report.pop("finding_class", None)
|
||||
dependency_report["dependency_metadata"] = {
|
||||
"package_name": "directus",
|
||||
"installed_version": "11.5.1",
|
||||
}
|
||||
|
||||
result = _do_update(
|
||||
report_id="vuln-0009",
|
||||
update_reason="An unauthenticated PATCH wrote the file.",
|
||||
fields={"poc_script_code": "PATCH /files/2f1c HTTP/1.1", "cvss_breakdown": _CVSS},
|
||||
)
|
||||
|
||||
assert result["success"] is False
|
||||
assert "dependency_cve" in result["error"]
|
||||
assert "poc_script_code" not in dependency_report
|
||||
|
||||
|
||||
def test_update_still_corrects_the_prose_of_a_dependency_finding(
|
||||
report_state: ReportState,
|
||||
) -> None:
|
||||
"""Fields every class carries stay editable on a dependency record."""
|
||||
_seed_weak_report(report_state)
|
||||
dependency_report = report_state.vulnerability_reports[0]
|
||||
dependency_report["finding_class"] = "dependency_cve"
|
||||
|
||||
result = _do_update(
|
||||
report_id="vuln-0009",
|
||||
update_reason="The advisory names a later fixed release than the report says.",
|
||||
fields={"remediation_steps": "Upgrade to 11.5.2 or later."},
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
assert dependency_report["remediation_steps"] == "Upgrade to 11.5.2 or later."
|
||||
assert dependency_report["finding_class"] == "dependency_cve"
|
||||
|
||||
|
||||
def test_update_refuses_code_locations_it_cannot_use(report_state: ReportState) -> None:
|
||||
"""A location without a usable file and line is reported, not dropped in silence."""
|
||||
_seed_weak_report(report_state)
|
||||
|
||||
result = _do_update(
|
||||
report_id="vuln-0009",
|
||||
update_reason="Naming the vulnerable handler.",
|
||||
fields={"code_locations": [{"label": "the file write"}]},
|
||||
)
|
||||
|
||||
assert result["success"] is False
|
||||
assert any("start_line" in error for error in result["errors"])
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue