feat(reporting): add read-only list_reports + get_report tools (#889)

This commit is contained in:
devin-ai-integration[bot]
2026-07-26 14:05:53 -07:00
committed by GitHub
parent 8169e177de
commit d2fbcb726d
16 changed files with 1192 additions and 183 deletions
+8 -1
View File
@@ -41,7 +41,12 @@ from strix.tools.proxy.tools import (
view_request,
view_sitemap_entry,
)
from strix.tools.reporting.tool import create_dependency_report, create_vulnerability_report
from strix.tools.reporting.tool import (
create_dependency_report,
create_vulnerability_report,
get_report,
list_reports,
)
from strix.tools.thinking.tool import think
from strix.tools.todo.tools import (
create_todo,
@@ -343,6 +348,8 @@ _BASE_TOOLS: tuple[Tool, ...] = (
web_search,
create_vulnerability_report,
create_dependency_report,
list_reports,
get_report,
list_requests,
view_request,
repeat_request,
+1
View File
@@ -215,6 +215,7 @@ VALIDATION REQUIREMENTS:
- 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
- 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>
<vulnerability_focus>
@@ -7,6 +7,13 @@ from .base_renderer import BaseToolRenderer
from .registry import register_tool_renderer
def _author_label(note: dict[str, Any]) -> str:
if note.get("by_you"):
return "you"
agent_name = note.get("agent_name")
return str(agent_name).strip() if agent_name else ""
@register_tool_renderer
class CreateNoteRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "create_note"
@@ -123,6 +130,9 @@ class ListNotesRenderer(BaseToolRenderer):
text.append("\n - ")
text.append(title)
text.append(f" ({category})", style="dim")
author = _author_label(note)
if author:
text.append(f" by {author}", style="dim")
if note_content:
text.append("\n ")
@@ -156,6 +166,9 @@ class GetNoteRenderer(BaseToolRenderer):
text.append("\n ")
text.append(title)
text.append(f" ({category})", style="dim")
author = _author_label(note)
if author:
text.append(f" by {author}", style="dim")
if content:
text.append("\n ")
text.append(content, style="dim")
@@ -431,3 +431,117 @@ class CreateDependencyReportRenderer(BaseToolRenderer):
css_classes = cls.get_css_classes("completed")
return Static(padded, classes=css_classes)
_LIST_SEVERITY_COLORS = {
"critical": "#dc2626",
"high": "#ea580c",
"medium": "#d97706",
"low": "#65a30d",
"info": "#0284c7",
"none": "#6b7280",
}
def _severity_style(severity: Any) -> str:
return _LIST_SEVERITY_COLORS.get(str(severity or "").lower(), "#d97706")
def _author_label(report: dict[str, Any]) -> str:
if report.get("by_you"):
return "you"
agent_name = report.get("agent_name")
return str(agent_name).strip() if agent_name else ""
@register_tool_renderer
class ListReportsRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "list_reports"
css_classes: ClassVar[list[str]] = ["tool-call", "reporting-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
result = _coerce_dict(tool_data.get("result"))
text = Text()
text.append("", style="#ef4444")
text.append("reports", style="dim")
if isinstance(tool_data.get("result"), str) and str(tool_data["result"]).strip():
text.append("\n ")
text.append(str(tool_data["result"]).strip(), style="dim")
elif result.get("success"):
total = result.get("total_count", 0)
reports = _coerce_list_of_dicts(result.get("reports"))
counts = _coerce_dict(result.get("severity_counts"))
text.append(f" ({total})", style="dim")
for sev, count in counts.items():
text.append(" ")
text.append(f"{sev} {count}", style=_severity_style(sev))
if not reports:
text.append("\n ")
text.append("No reports filed yet", style="dim")
else:
for report in reports:
rid = str(report.get("id", "")).strip()
title = str(report.get("title", "")).strip() or "(untitled)"
severity = str(report.get("severity", "")).strip()
text.append("\n - ")
if severity:
text.append(severity.upper(), style=f"bold {_severity_style(severity)}")
text.append(" ")
if rid:
text.append(f"{rid} ", style="dim")
text.append(title)
author = _author_label(report)
if author:
text.append(f" ({author})", style="dim")
else:
text.append("\n ")
text.append("Loading...", style="dim")
css_classes = cls.get_css_classes("completed")
return Static(text, classes=css_classes)
@register_tool_renderer
class GetReportRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "get_report"
css_classes: ClassVar[list[str]] = ["tool-call", "reporting-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
result = _coerce_dict(tool_data.get("result"))
text = Text()
text.append("", style="#ef4444")
text.append("report read", style="dim")
report = _coerce_dict(result.get("report")) if result.get("success") else {}
if report:
rid = str(report.get("id", "")).strip()
title = str(report.get("title", "")).strip() or "(untitled)"
severity = str(report.get("severity", "")).strip()
text.append("\n ")
if severity:
text.append(severity.upper(), style=f"bold {_severity_style(severity)}")
text.append(" ")
if rid:
text.append(f"{rid} ", style="dim")
text.append(title)
author = _author_label(report)
if author:
text.append(f" ({author})", style="dim")
target = str(report.get("target", "")).strip()
if target:
text.append("\n ")
text.append(target, style="dim")
else:
text.append("\n ")
detail = result.get("error") if result.get("success") is False else None
text.append(str(detail) if detail else "Loading...", style="dim")
css_classes = cls.get_css_classes("completed")
return Static(text, classes=css_classes)
@@ -49,6 +49,9 @@ export default function NotesRenderer({ toolName, args, result }: ToolRendererPr
<div className="mt-1.5 text-[#999] text-[13px]">
{note.title ?? "(untitled)"}
<span className="text-[#555] ml-1">({note.category ?? "general"})</span>
{(note.by_you || note.agent_name) && (
<span className="text-[#666] ml-1 text-xs">by {note.by_you ? "you" : note.agent_name}</span>
)}
</div>
{note.content && <div className="mt-1"><Markdown text={note.content} /></div>}
</>
@@ -74,6 +77,9 @@ export default function NotesRenderer({ toolName, args, result }: ToolRendererPr
<span className="text-[#555] mr-1">-</span>
<span className="text-[#999]">{n.title ?? "(untitled)"}</span>
<span className="text-[#555] ml-1">({n.category ?? "general"})</span>
{(n.by_you || n.agent_name) && (
<span className="text-[#666] ml-1 text-xs">by {n.by_you ? "you" : n.agent_name}</span>
)}
{n.content && <div className="ml-3"><Markdown text={n.content} /></div>}
</div>
))}
@@ -0,0 +1,121 @@
"use client";
import type { ToolRendererProps } from "@/types/events";
import { TruncatedText } from "./ToolCard";
import Markdown from "./Markdown";
const SEVERITY_COLORS: Record<string, string> = {
critical: "text-red-400", high: "text-orange-400", medium: "text-yellow-400",
low: "text-blue-400", info: "text-cyan-400", none: "text-[#888]",
};
interface ReportEntry {
id?: string;
title?: string;
severity?: string;
cvss?: number;
cve?: string;
cwe?: string;
target?: string;
endpoint?: string;
method?: string;
description_preview?: string;
description?: string;
agent_name?: string;
by_you?: boolean;
}
function authorTag(r: ReportEntry) {
if (!r.agent_name && !r.by_you) return null;
const label = r.by_you ? "you" : r.agent_name;
return <span className="text-[#666] text-xs ml-1.5">({label})</span>;
}
function sevBadge(severity: string | undefined) {
const sev = String(severity ?? "").toLowerCase();
const color = SEVERITY_COLORS[sev] ?? "text-yellow-400";
return <span className={`font-semibold text-[13px] ${color}`}>{sev.toUpperCase() || "—"}</span>;
}
export default function ReportListRenderer({ toolName, result }: ToolRendererProps) {
const res = result as Record<string, unknown> | null;
const ok = res != null && typeof res === "object" && res.success === true;
if (toolName === "get_report") {
const report = ok ? (res.report as ReportEntry | undefined) : undefined;
return (
<div>
<span className="text-red-400/80 font-semibold text-sm">report</span>
{report ? (
<div className="mt-1.5 space-y-2">
<div className="flex items-center gap-2 flex-wrap">
{sevBadge(report.severity)}
{report.cvss != null && <span className="text-[#888] text-[13px]">CVSS {report.cvss}</span>}
{report.id && <span className="text-[#555] font-mono text-[13px]">{report.id}</span>}
{report.cve && <span className="text-[#888] font-mono text-[13px]">{report.cve}</span>}
{report.cwe && <span className="text-[#888] font-mono text-[13px]">{report.cwe}</span>}
{(report.agent_name || report.by_you) && (
<span className="text-[#666] text-[13px]">{report.by_you ? "you" : report.agent_name}</span>
)}
</div>
{report.title && <div className="text-[15px] text-white/80 font-semibold">{report.title}</div>}
{(report.target || report.endpoint) && (
<div className="text-[13px] text-[#888] font-mono">
{report.target}{report.endpoint ? ` ${report.method ?? ""} ${report.endpoint}` : ""}
</div>
)}
{report.description && <TruncatedText text={report.description} maxLines={20} />}
</div>
) : (
<div className="mt-1 text-[#555] text-xs">
{(res && typeof res === "object" && (res.error as string)) || "Report not found"}
</div>
)}
</div>
);
}
// list_reports
const rawReports = ok ? res.reports : null;
const reports: ReportEntry[] = Array.isArray(rawReports) ? (rawReports as ReportEntry[]) : [];
const total = ok && typeof res.total_count === "number" ? (res.total_count as number) : reports.length;
const counts = ok && res.severity_counts && typeof res.severity_counts === "object"
? (res.severity_counts as Record<string, number>)
: {};
const countEntries = Object.entries(counts);
return (
<div>
<div className="flex items-center gap-2 flex-wrap">
<span className="text-red-400/80 font-semibold text-sm">reports</span>
<span className="text-[#555] text-[13px]">({total})</span>
{countEntries.map(([sev, n]) => (
<span key={sev} className="text-[13px]">
{sevBadge(sev)}<span className="text-[#888] ml-0.5">{n}</span>
</span>
))}
</div>
{reports.length > 0 ? (
<div className="mt-1.5 space-y-1">
{reports.map((r, i) => (
<div key={r.id ?? i} className="text-[13px]">
<span className="text-[#555] mr-1">-</span>
{sevBadge(r.severity)}
{r.id && <span className="text-[#555] font-mono ml-1.5">{r.id}</span>}
<span className="text-[#999] ml-1.5">{r.title ?? "(untitled)"}</span>
{authorTag(r)}
{(r.target || r.endpoint) && (
<div className="ml-3 text-[#666] font-mono text-xs">
{r.target}{r.endpoint ? ` ${r.method ?? ""} ${r.endpoint}` : ""}
</div>
)}
{r.description_preview && (
<div className="ml-3"><Markdown text={r.description_preview} /></div>
)}
</div>
))}
</div>
) : <div className="mt-1 text-[#555] text-xs">No reports filed yet</div>}
</div>
);
}
@@ -12,6 +12,7 @@ import FileEditRenderer from "./FileEditRenderer";
import ApplyPatchRenderer from "./ApplyPatchRenderer";
import ViewImageRenderer from "./ViewImageRenderer";
import VulnReportRenderer from "./VulnReportRenderer";
import ReportListRenderer from "./ReportListRenderer";
import ProxyRenderer from "./ProxyRenderer";
import ThinkRenderer from "./ThinkRenderer";
import AgentCommsRenderer from "./AgentCommsRenderer";
@@ -101,7 +102,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"],
reporting: ["create_vulnerability_report", "list_reports", "get_report"],
thinking: ["think"],
agents: ["create_agent", "agent_finish", "send_message_to_agent", "wait_for_message", "view_agent_graph", "stop_agent"],
search: ["web_search"],
@@ -128,6 +129,8 @@ const RENDERER_OVERRIDES: Partial<Record<string, ComponentType<ToolRendererProps
finish_scan: FinishRenderer,
apply_patch: ApplyPatchRenderer,
view_image: ViewImageRenderer,
list_reports: ReportListRenderer,
get_report: ReportListRenderer,
};
/**
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
+2 -2
View File
@@ -6,8 +6,8 @@
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<meta name="color-scheme" content="dark" />
<title>Strix Results</title>
<script type="module" crossorigin src="./assets/index-Dd1cyttN.js"></script>
<link rel="stylesheet" crossorigin href="./assets/index-vV8wxCG6.css">
<script type="module" crossorigin src="./assets/index-DzvI_0HX.js"></script>
<link rel="stylesheet" crossorigin href="./assets/index-C3kQ5kk8.css">
</head>
<body>
<div id="root"></div>
+10 -6
View File
@@ -116,12 +116,16 @@ async def finish_scan(
/ ``crashed`` / ``stopped`` agents are safe to leave behind.
Calling ``finish_scan`` while children are alive orphans their
work and produces an incomplete report.
2. All vulnerabilities you found are filed via
``create_vulnerability_report`` — or, for known-CVE dependency
findings, ``create_dependency_report`` (un-reported findings are
not tracked and not credited). A dependency CVE already filed via
``create_dependency_report`` counts as reported; it does NOT need
re-filing here and does NOT block finishing.
2. It's a good idea to call ``list_reports`` before finishing to
review every finding filed in this scan (use ``get_report`` for
full detail on any of them) so your ``executive_summary`` /
``technical_analysis`` are grounded in what was actually reported
— don't invent or omit findings. All vulnerabilities you found are
filed via ``create_vulnerability_report`` — or, for known-CVE
dependency findings, ``create_dependency_report`` (un-reported
findings are not tracked and not credited). A dependency CVE
already filed via ``create_dependency_report`` counts as reported;
it does NOT need re-filing here and does NOT block finishing.
3. Don't double-report — one report per distinct vulnerability.
4. **Attack-chaining gate.** Do NOT finish until you have genuinely
considered chaining the confirmed findings into higher-impact,
+63 -6
View File
@@ -27,6 +27,21 @@ _NOTE_ID_GENERATION_ATTEMPTS = 1024
_notes_path: Path | None = None
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 {}
raw_agent_id = inner.get("agent_id")
agent_id = raw_agent_id if isinstance(raw_agent_id, str) else None
agent_name: str | None = None
coordinator = inner.get("coordinator")
if agent_id is not None and coordinator is not None:
names = getattr(coordinator, "names", {})
if isinstance(names, dict):
raw_agent_name = names.get(agent_id)
agent_name = raw_agent_name if isinstance(raw_agent_name, str) else None
return agent_id, agent_name
def _generate_note_id() -> str | None:
for _ in range(_NOTE_ID_GENERATION_ATTEMPTS):
note_id = uuid.uuid4().hex[:6]
@@ -117,10 +132,26 @@ def _filter_notes(
return filtered
def _mark_authorship(
entry: dict[str, Any], note: dict[str, Any], caller_agent_id: str | None
) -> dict[str, Any]:
"""Attach the note's author and flag whether the caller wrote it."""
agent_name = note.get("agent_name")
if agent_name:
entry["agent_name"] = agent_name
agent_id = note.get("agent_id")
if agent_id:
entry["agent_id"] = agent_id
if caller_agent_id is not None and agent_id == caller_agent_id:
entry["by_you"] = True
return entry
def _to_note_listing_entry(
note: dict[str, Any],
*,
include_content: bool = False,
caller_agent_id: str | None = None,
) -> dict[str, Any]:
entry = {
"note_id": note.get("note_id"),
@@ -138,7 +169,7 @@ def _to_note_listing_entry(
entry["content_preview"] = f"{content[:_DEFAULT_CONTENT_PREVIEW_CHARS].rstrip()}..."
else:
entry["content_preview"] = content
return entry
return _mark_authorship(entry, note, caller_agent_id)
def _create_note_impl(
@@ -146,6 +177,8 @@ def _create_note_impl(
content: str,
category: str = "general",
tags: list[str] | None = None,
agent_id: str | None = None,
agent_name: str | None = None,
) -> dict[str, Any]:
with _notes_lock:
try:
@@ -179,6 +212,10 @@ def _create_note_impl(
"created_at": timestamp,
"updated_at": timestamp,
}
if agent_id:
note["agent_id"] = agent_id
if agent_name:
note["agent_name"] = agent_name
_notes_storage[note_id] = note
except (ValueError, TypeError) as e:
return {"success": False, "error": f"Failed to create note: {e}", "note_id": None}
@@ -197,11 +234,17 @@ def _list_notes_impl(
tags: list[str] | None = None,
search: str | None = None,
include_content: bool = False,
caller_agent_id: str | None = None,
) -> dict[str, Any]:
with _notes_lock:
try:
filtered = _filter_notes(category=category, tags=tags, search_query=search)
notes = [_to_note_listing_entry(n, include_content=include_content) for n in filtered]
notes = [
_to_note_listing_entry(
n, include_content=include_content, caller_agent_id=caller_agent_id
)
for n in filtered
]
except (ValueError, TypeError) as e:
return {
"success": False,
@@ -218,7 +261,7 @@ def _list_notes_impl(
}
def _get_note_impl(note_id: str) -> dict[str, Any]:
def _get_note_impl(note_id: str, caller_agent_id: str | None = None) -> dict[str, Any]:
with _notes_lock:
try:
if not note_id or not note_id.strip():
@@ -232,6 +275,7 @@ def _get_note_impl(note_id: str) -> dict[str, Any]:
}
note_with_id = note.copy()
note_with_id["note_id"] = note_id
_mark_authorship(note_with_id, note, caller_agent_id)
except (ValueError, TypeError) as e:
return {"success": False, "error": f"Failed to get note: {e}", "note": None}
else:
@@ -304,7 +348,9 @@ async def create_note(
Notes are visible to every agent in the same scan for the lifetime
of the run; they live in-memory only and are cleared when the
process exits.
process exits. Each note records the agent that wrote it, so
``list_notes`` / ``get_note`` show the author (``agent_name``) and
flag your own notes with ``by_you``.
For actionable tasks, use ``todo`` instead — notes are for capturing
information, todos are for tracking work.
@@ -329,8 +375,11 @@ async def create_note(
category: One of the categories above. Default ``"general"``.
tags: Optional free-form tags.
"""
agent_id, agent_name = _caller_identity(ctx)
return json.dumps(
await asyncio.to_thread(_create_note_impl, title, content, category, tags),
await asyncio.to_thread(
_create_note_impl, title, content, category, tags, agent_id, agent_name
),
ensure_ascii=False,
default=str,
)
@@ -355,6 +404,9 @@ async def list_notes(
when you need to scan many notes; expensive in tokens for large
notes.
Each entry also carries the author (``agent_name``) and, for notes
you wrote yourself, ``by_you: true``.
Args:
category: Filter by category.
tags: Filter to notes that have any of these tags.
@@ -362,6 +414,7 @@ async def list_notes(
include_content: When False (default) entries have a preview;
when True the full ``content`` is included.
"""
caller_agent_id, _ = _caller_identity(ctx)
return json.dumps(
await asyncio.to_thread(
_list_notes_impl,
@@ -369,6 +422,7 @@ async def list_notes(
tags=tags,
search=search,
include_content=include_content,
caller_agent_id=caller_agent_id,
),
ensure_ascii=False,
default=str,
@@ -382,8 +436,11 @@ async def get_note(ctx: RunContextWrapper, note_id: str) -> str:
Args:
note_id: Note id from ``create_note`` or a ``list_notes`` entry.
"""
caller_agent_id, _ = _caller_identity(ctx)
return json.dumps(
await asyncio.to_thread(_get_note_impl, note_id), ensure_ascii=False, default=str
await asyncio.to_thread(_get_note_impl, note_id, caller_agent_id),
ensure_ascii=False,
default=str,
)
+309 -21
View File
@@ -1,7 +1,13 @@
"""``create_vulnerability_report`` — file a vuln finding with dedup + CVSS."""
"""Reporting tools — file vuln 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.
"""
from __future__ import annotations
import asyncio
import json
import logging
import re
@@ -318,6 +324,21 @@ async def _do_create( # noqa: PLR0912
}
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 {}
raw_agent_id = inner.get("agent_id")
agent_id = raw_agent_id if isinstance(raw_agent_id, str) else None
agent_name: str | None = None
coordinator = inner.get("coordinator")
if agent_id is not None and coordinator is not None:
names = getattr(coordinator, "names", {})
if isinstance(names, dict):
raw_agent_name = names.get(agent_id)
agent_name = raw_agent_name if isinstance(raw_agent_name, str) else None
return agent_id, agent_name
@function_tool(timeout=180, strict_mode=False)
async def create_vulnerability_report(
ctx: RunContextWrapper,
@@ -604,16 +625,7 @@ async def create_vulnerability_report(
template engine's auto-escaping over string interpolation.
fix_effort: "low"
"""
inner = ctx.context if isinstance(ctx.context, dict) else {}
raw_agent_id = inner.get("agent_id")
agent_id = raw_agent_id if isinstance(raw_agent_id, str) else None
agent_name = None
coordinator = inner.get("coordinator")
if agent_id is not None and coordinator is not None:
names = getattr(coordinator, "names", {})
if isinstance(names, dict):
raw_agent_name = names.get(agent_id)
agent_name = raw_agent_name if isinstance(raw_agent_name, str) else None
agent_id, agent_name = _caller_identity(ctx)
result = await _do_create(
title=title,
@@ -918,16 +930,7 @@ async def create_dependency_report(
fix_effort: One of ``trivial`` / ``low`` / ``medium`` / ``high``
(dependency upgrades are usually ``trivial``/``low``).
"""
inner = ctx.context if isinstance(ctx.context, dict) else {}
raw_agent_id = inner.get("agent_id")
agent_id = raw_agent_id if isinstance(raw_agent_id, str) else None
agent_name = None
coordinator = inner.get("coordinator")
if agent_id is not None and coordinator is not None:
names = getattr(coordinator, "names", {})
if isinstance(names, dict):
raw_agent_name = names.get(agent_id)
agent_name = raw_agent_name if isinstance(raw_agent_name, str) else None
agent_id, agent_name = _caller_identity(ctx)
result = await _do_create_dependency(
title=title,
@@ -949,3 +952,288 @@ async def create_dependency_report(
agent_name=agent_name,
)
return json.dumps(result, ensure_ascii=False, default=str)
_SEVERITY_ORDER = {
"critical": 0,
"high": 1,
"medium": 2,
"low": 3,
"info": 4,
"none": 5,
}
_VALID_SEVERITIES = frozenset(_SEVERITY_ORDER)
_VALID_FINDING_CLASSES = frozenset({"dynamic", "dependency_cve"})
_REPORT_DESCRIPTION_PREVIEW_CHARS = 280
# Compact, listing-safe fields — no full bodies / PoC code / evidence.
_REPORT_SUMMARY_FIELDS = (
"id",
"title",
"severity",
"cvss",
"finding_class",
"cve",
"cwe",
"target",
"endpoint",
"method",
"fix_effort",
"agent_name",
"timestamp",
)
def _report_severity_rank(report: dict[str, Any]) -> int:
return _SEVERITY_ORDER.get(str(report.get("severity", "")).lower(), 99)
def _report_matches_filters(
report: dict[str, Any],
*,
severity: str | None,
finding_class: str | None,
target: str | None,
search: str | None,
) -> bool:
if severity and str(report.get("severity", "")).lower() != severity:
return False
if finding_class and str(report.get("finding_class", "dynamic")).lower() != finding_class:
return False
if target:
target_lower = target.lower()
haystack = f"{report.get('target', '')} {report.get('endpoint', '')}".lower()
if target_lower not in haystack:
return False
if search:
search_lower = search.lower()
title_match = search_lower in str(report.get("title", "")).lower()
desc_match = search_lower in str(report.get("description", "")).lower()
if not (title_match or desc_match):
return False
return True
def _mark_authorship(
entry: dict[str, Any], report: dict[str, Any], caller_agent_id: str | None
) -> dict[str, Any]:
"""Flag whether ``report`` was filed by the agent making this call."""
if caller_agent_id is not None and report.get("agent_id") == caller_agent_id:
entry["by_you"] = True
return entry
def _to_report_summary_entry(
report: dict[str, Any], caller_agent_id: str | None = None
) -> dict[str, Any]:
entry = {
field: report[field] for field in _REPORT_SUMMARY_FIELDS if report.get(field) is not None
}
description = str(report.get("description", "")).strip()
if description:
if len(description) > _REPORT_DESCRIPTION_PREVIEW_CHARS:
entry["description_preview"] = (
f"{description[:_REPORT_DESCRIPTION_PREVIEW_CHARS].rstrip()}..."
)
else:
entry["description_preview"] = description
return _mark_authorship(entry, report, caller_agent_id)
def _severity_counts(reports: list[dict[str, Any]]) -> dict[str, int]:
counts: dict[str, int] = {}
for report in reports:
sev = str(report.get("severity", "")).lower() or "none"
counts[sev] = counts.get(sev, 0) + 1
return {sev: counts[sev] for sev in _SEVERITY_ORDER if sev in counts}
async def _run_report_reader(fn: Any, *args: Any, **kwargs: Any) -> dict[str, Any]:
try:
return await asyncio.to_thread(fn, *args, **kwargs)
except (ImportError, AttributeError) as e:
logger.exception("report reader failed")
return {"success": False, "error": f"Failed to read reports: {e!s}"}
def _do_list_reports(
*,
severity: str | None,
finding_class: str | None,
target: str | None,
search: str | None,
include_details: bool,
caller_agent_id: str | None = None,
) -> dict[str, Any]:
errors: list[str] = []
severity = (severity or "").strip().lower() or None
if severity and severity not in _VALID_SEVERITIES:
errors.append(
f"Invalid severity: {severity!r}. Must be one of: {sorted(_VALID_SEVERITIES)}"
)
finding_class = (finding_class or "").strip().lower() or None
if finding_class and finding_class not in _VALID_FINDING_CLASSES:
errors.append(
f"Invalid finding_class: {finding_class!r}. "
f"Must be one of: {sorted(_VALID_FINDING_CLASSES)}"
)
if errors:
return {"success": False, "error": "Validation failed", "errors": errors}
from strix.report.state import get_global_report_state
report_state = get_global_report_state()
if report_state is None:
return {
"success": True,
"reports": [],
"filtered_count": 0,
"total_count": 0,
"severity_counts": {},
"warning": "Report state unavailable - no reports have been filed yet",
}
all_reports = report_state.get_existing_vulnerabilities()
matched = [
r
for r in all_reports
if _report_matches_filters(
r,
severity=severity,
finding_class=finding_class,
target=(target or "").strip() or None,
search=(search or "").strip() or None,
)
]
matched.sort(key=lambda r: (_report_severity_rank(r), str(r.get("id", ""))))
reports = [
_mark_authorship(dict(r), r, caller_agent_id)
if include_details
else _to_report_summary_entry(r, caller_agent_id)
for r in matched
]
return {
"success": True,
"reports": reports,
"filtered_count": len(reports),
"total_count": len(all_reports),
"severity_counts": _severity_counts(all_reports),
}
def _do_get_report(report_id: str, caller_agent_id: str | None = None) -> dict[str, Any]:
report_id = (report_id or "").strip()
if not report_id:
return {"success": False, "error": "report_id cannot be empty", "report": None}
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",
"report": None,
}
for report in report_state.get_existing_vulnerabilities():
if report.get("id") == report_id:
return {
"success": True,
"report": _mark_authorship(dict(report), report, caller_agent_id),
}
return {
"success": False,
"error": f"Report with id '{report_id}' not found",
"report": None,
}
@function_tool(timeout=30)
async def list_reports(
ctx: RunContextWrapper,
severity: str | None = None,
finding_class: str | None = None,
target: str | None = None,
search: str | None = None,
include_details: bool = False,
) -> str:
"""List vulnerability reports filed so far in this scan — metadata-first.
**For the orchestrator / root agent.** This is an orchestration tool
for tracking scan-wide coverage and assembling the final report — leaf
/ specialist agents do their own testing and file findings; they should
NOT call this. If you are a subagent, ignore it and focus on your task.
Reports are shared across **every** agent in the scan, so this returns
findings filed by any agent (root or child), not just your own. As the
root agent, use it to track progress, avoid dispatching work on
already-covered ground, reason about attack-chaining across confirmed
findings, and build the ``finish_scan`` executive summary.
By default each entry is compact: ``id``, ``title``, ``severity``,
``cvss``, ``finding_class``, ``cve`` / ``cwe``, ``target`` /
``endpoint``, ``fix_effort``, ``agent_name`` (who filed it), ``timestamp``,
plus a 280-char ``description_preview``. Entries you filed yourself are
flagged ``by_you: true``. The response also carries
``total_count`` and ``severity_counts`` (counts per severity across all
reports, ignoring filters). Set ``include_details=True`` for full report
bodies (PoC, evidence, remediation, code_locations) — token-expensive;
prefer ``get_report`` to drill into a single finding.
Filters compose (all must match): ``severity`` and ``finding_class``
match exactly, ``target`` is a substring match against target/endpoint,
and ``search`` is a substring match against title/description. Results
are ordered by severity (critical -> info), then report id.
This is read-only — it never files or dedupes anything.
Args:
severity: Filter to one of ``critical`` / ``high`` / ``medium`` /
``low`` / ``info`` / ``none``.
finding_class: Filter to ``dynamic`` (PoC-backed) or
``dependency_cve`` (known-CVE supply-chain).
target: Substring match against a report's target / endpoint.
search: Substring match against title and description.
include_details: When False (default) entries are compact; when
True full report bodies are returned.
"""
caller_agent_id, _ = _caller_identity(ctx)
return json.dumps(
await _run_report_reader(
_do_list_reports,
severity=severity,
finding_class=finding_class,
target=target,
search=search,
include_details=include_details,
caller_agent_id=caller_agent_id,
),
ensure_ascii=False,
default=str,
)
@function_tool(timeout=30)
async def get_report(ctx: RunContextWrapper, report_id: str) -> str:
"""Fetch one vulnerability report by its id (e.g. ``vuln-0001``).
Returns the full report body — description, impact, technical analysis,
PoC, evidence, remediation, CVSS breakdown, and any ``code_locations``.
Use ``list_reports`` first to find ids; this is the cheap way to read a
single finding in full without pulling every body.
Read-only.
Args:
report_id: Report id from ``list_reports`` or a
``create_vulnerability_report`` / ``create_dependency_report``
response (format ``vuln-NNNN``).
"""
caller_agent_id, _ = _caller_identity(ctx)
return json.dumps(
await _run_report_reader(_do_get_report, report_id, caller_agent_id),
ensure_ascii=False,
default=str,
)
+359
View File
@@ -0,0 +1,359 @@
"""Tests for the read-only list_reports / get_report tools."""
from __future__ import annotations
from typing import TYPE_CHECKING
import pytest
from strix.report.state import ReportState, set_global_report_state
from strix.tools.reporting.tool import (
_do_get_report,
_do_list_reports,
get_report,
list_reports,
)
if TYPE_CHECKING:
from pathlib import Path
@pytest.fixture
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)
return state
def _seed(state: ReportState) -> None:
state.add_vulnerability_report(
title="Reflected XSS in search",
severity="medium",
description="q reflects unencoded input.",
target="https://app.example.com",
endpoint="/search",
method="GET",
cwe="CWE-79",
cvss=6.1,
agent_name="XSS Agent",
)
state.add_vulnerability_report(
title="SQL Injection in login",
severity="critical",
description="Login parameter is injectable.",
target="https://app.example.com",
endpoint="/api/login",
cwe="CWE-89",
cvss=9.8,
agent_name="SQLi Agent",
)
state.add_vulnerability_report(
title="CVE-2021-23337 in lodash 4.17.20",
severity="high",
description="Command injection via template.",
target="repo/package.json",
cve="CVE-2021-23337",
cvss=7.2,
finding_class="dependency_cve",
)
@pytest.mark.usefixtures("report_state")
def test_list_reports_empty() -> None:
result = _do_list_reports(
severity=None, finding_class=None, target=None, search=None, include_details=False
)
assert result["success"] is True
assert result["reports"] == []
assert result["total_count"] == 0
assert result["severity_counts"] == {}
def test_list_reports_metadata_first_and_sorted(report_state: ReportState) -> None:
_seed(report_state)
result = _do_list_reports(
severity=None, finding_class=None, target=None, search=None, include_details=False
)
assert result["success"] is True
assert result["total_count"] == 3
# sorted by severity: critical, high, medium
titles = [r["title"] for r in result["reports"]]
assert titles == [
"SQL Injection in login",
"CVE-2021-23337 in lodash 4.17.20",
"Reflected XSS in search",
]
assert result["severity_counts"] == {"critical": 1, "high": 1, "medium": 1}
# compact entries carry a preview, never full-body fields
first = result["reports"][0]
assert "description_preview" in first
assert "poc_script_code" not in first
assert "evidence" not in first
def test_list_reports_filter_severity(report_state: ReportState) -> None:
_seed(report_state)
result = _do_list_reports(
severity="critical", finding_class=None, target=None, search=None, include_details=False
)
assert result["filtered_count"] == 1
assert result["reports"][0]["title"] == "SQL Injection in login"
# severity_counts reflect ALL reports, not the filtered subset
assert result["total_count"] == 3
def test_list_reports_filter_finding_class(report_state: ReportState) -> None:
_seed(report_state)
result = _do_list_reports(
severity=None,
finding_class="dependency_cve",
target=None,
search=None,
include_details=False,
)
assert result["filtered_count"] == 1
assert result["reports"][0]["cve"] == "CVE-2021-23337"
def test_list_reports_filter_target_and_search(report_state: ReportState) -> None:
_seed(report_state)
by_target = _do_list_reports(
severity=None, finding_class=None, target="/api/login", search=None, include_details=False
)
assert [r["title"] for r in by_target["reports"]] == ["SQL Injection in login"]
by_search = _do_list_reports(
severity=None, finding_class=None, target=None, search="lodash", include_details=False
)
assert [r["cve"] for r in by_search["reports"]] == ["CVE-2021-23337"]
def test_list_reports_include_details(report_state: ReportState) -> None:
_seed(report_state)
result = _do_list_reports(
severity="medium", finding_class=None, target=None, search=None, include_details=True
)
entry = result["reports"][0]
assert entry["description"] == "q reflects unencoded input."
assert "description_preview" not in entry
@pytest.mark.usefixtures("report_state")
def test_list_reports_rejects_invalid_filters() -> None:
result = _do_list_reports(
severity="spicy", finding_class="bogus", target=None, search=None, include_details=False
)
assert result["success"] is False
joined = " ".join(result["errors"])
assert "severity" in joined
assert "finding_class" in joined
def test_get_report_success(report_state: ReportState) -> None:
_seed(report_state)
result = _do_get_report("vuln-0002")
assert result["success"] is True
assert result["report"]["title"] == "SQL Injection in login"
assert result["report"]["cwe"] == "CWE-89"
def test_get_report_not_found(report_state: ReportState) -> None:
_seed(report_state)
result = _do_get_report("vuln-9999")
assert result["success"] is False
assert result["report"] is None
@pytest.mark.usefixtures("report_state")
def test_get_report_empty_id() -> None:
result = _do_get_report(" ")
assert result["success"] is False
def test_read_tools_are_read_only_and_stateless(report_state: ReportState) -> None:
_seed(report_state)
before = list(report_state.vulnerability_reports)
_do_list_reports(
severity=None, finding_class=None, target=None, search=None, include_details=True
)
_do_get_report("vuln-0001")
assert report_state.vulnerability_reports == before
def test_tool_descriptions_mention_read_only() -> None:
assert "read-only" in list_reports.description.lower()
assert "get_report" in list_reports.description
assert "read-only" in get_report.description.lower()
def test_list_reports_flags_callers_own_reports(report_state: ReportState) -> None:
report_state.add_vulnerability_report(
title="Mine", severity="high", target="t", agent_id="agent-1", agent_name="Agent One"
)
report_state.add_vulnerability_report(
title="Theirs", severity="low", target="t", agent_id="agent-2", agent_name="Agent Two"
)
result = _do_list_reports(
severity=None,
finding_class=None,
target=None,
search=None,
include_details=False,
caller_agent_id="agent-1",
)
by_title = {r["title"]: r for r in result["reports"]}
assert by_title["Mine"].get("by_you") is True
assert by_title["Mine"]["agent_name"] == "Agent One"
assert "by_you" not in by_title["Theirs"]
assert by_title["Theirs"]["agent_name"] == "Agent Two"
def test_list_reports_no_caller_marks_nothing(report_state: ReportState) -> None:
report_state.add_vulnerability_report(
title="Mine", severity="high", target="t", agent_id="agent-1", agent_name="Agent One"
)
result = _do_list_reports(
severity=None, finding_class=None, target=None, search=None, include_details=False
)
assert "by_you" not in result["reports"][0]
def test_get_report_flags_caller_ownership(report_state: ReportState) -> None:
report_state.add_vulnerability_report(
title="Mine", severity="high", target="t", agent_id="agent-1", agent_name="Agent One"
)
mine = _do_get_report("vuln-0001", caller_agent_id="agent-1")
assert mine["report"].get("by_you") is True
theirs = _do_get_report("vuln-0001", caller_agent_id="agent-9")
assert "by_you" not in theirs["report"]
def test_list_reports_docstring_scopes_to_orchestrator() -> None:
desc = list_reports.description.lower()
assert "orchestrator" in desc or "root agent" in desc
@pytest.mark.parametrize("sev", ["CRITICAL", "Critical", "critical"])
def test_list_reports_severity_filter_case_insensitive(report_state: ReportState, sev: str) -> None:
_seed(report_state)
result = _do_list_reports(
severity=sev, finding_class=None, target=None, search=None, include_details=False
)
assert [r["title"] for r in result["reports"]] == ["SQL Injection in login"]
def test_list_reports_finding_class_filter_case_insensitive(report_state: ReportState) -> None:
_seed(report_state)
result = _do_list_reports(
severity=None,
finding_class="Dependency_CVE",
target=None,
search=None,
include_details=False,
)
assert result["filtered_count"] == 1
def test_list_reports_target_filter_matches_domain_and_is_case_insensitive(
report_state: ReportState,
) -> None:
_seed(report_state)
# substring of the `target` field (not the endpoint), mixed case
result = _do_list_reports(
severity=None,
finding_class=None,
target="APP.EXAMPLE.COM",
search=None,
include_details=False,
)
assert {r["title"] for r in result["reports"]} == {
"Reflected XSS in search",
"SQL Injection in login",
}
def test_list_reports_search_matches_title_case_insensitive(report_state: ReportState) -> None:
_seed(report_state)
result = _do_list_reports(
severity=None, finding_class=None, target=None, search="INJECTION", include_details=False
)
# matches title "SQL Injection in login" and description "Command injection..."
assert {r["title"] for r in result["reports"]} == {
"SQL Injection in login",
"CVE-2021-23337 in lodash 4.17.20",
}
def test_list_reports_filters_compose(report_state: ReportState) -> None:
_seed(report_state)
result = _do_list_reports(
severity="high",
finding_class="dependency_cve",
target="package.json",
search="lodash",
include_details=False,
)
assert [r["cve"] for r in result["reports"]] == ["CVE-2021-23337"]
def test_list_reports_no_match_returns_empty_success(report_state: ReportState) -> None:
_seed(report_state)
result = _do_list_reports(
severity=None,
finding_class=None,
target=None,
search="nonexistent-xyz",
include_details=False,
)
assert result["success"] is True
assert result["filtered_count"] == 0
assert result["reports"] == []
# counts still reflect all reports
assert result["total_count"] == 3
def test_list_reports_description_preview_truncated(report_state: ReportState) -> None:
long_desc = "A" * 400
report_state.add_vulnerability_report(
title="Long", severity="low", description=long_desc, target="t"
)
result = _do_list_reports(
severity=None, finding_class=None, target=None, search=None, include_details=False
)
preview = result["reports"][0]["description_preview"]
assert preview.endswith("...")
assert len(preview) <= 284 # 280 chars + "..."
def test_list_reports_severity_counts_ordered_with_none_bucket(
report_state: ReportState,
) -> None:
report_state.add_vulnerability_report(title="A", severity="low", target="t")
report_state.add_vulnerability_report(title="B", severity="critical", target="t")
report_state.add_vulnerability_report(title="C", severity="", target="t")
result = _do_list_reports(
severity=None, finding_class=None, target=None, search=None, include_details=False
)
# ordered critical -> ... -> none
assert list(result["severity_counts"].keys()) == ["critical", "low", "none"]
@pytest.mark.usefixtures("report_state")
def test_list_reports_no_state_returns_warning(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr("strix.report.state.get_global_report_state", lambda: None)
result = _do_list_reports(
severity=None, finding_class=None, target=None, search=None, include_details=False
)
assert result["success"] is True
assert result["reports"] == []
assert "warning" in result
@pytest.mark.usefixtures("report_state")
def test_get_report_no_state_returns_error(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr("strix.report.state.get_global_report_state", lambda: None)
result = _do_get_report("vuln-0001")
assert result["success"] is False
assert result["report"] is None
+39 -3
View File
@@ -32,7 +32,7 @@ def test_create_note_retries_on_note_id_collision(monkeypatch: pytest.MonkeyPatc
uuid.UUID("12345600-0000-4000-8000-000000000000"),
]
)
monkeypatch.setattr(notes_tools.uuid, "uuid4", lambda: next(generated_ids))
monkeypatch.setattr("strix.tools.notes.tools.uuid.uuid4", lambda: next(generated_ids))
first = notes_tools._create_note_impl("first", "original content")
second = notes_tools._create_note_impl("second", "new content")
@@ -51,8 +51,7 @@ def test_create_note_returns_error_after_repeated_note_id_collisions(
) -> None:
monkeypatch.setattr(notes_tools, "_NOTE_ID_GENERATION_ATTEMPTS", 2)
monkeypatch.setattr(
notes_tools.uuid,
"uuid4",
"strix.tools.notes.tools.uuid.uuid4",
lambda: uuid.UUID("abcdef00-0000-4000-8000-000000000000"),
)
notes_tools._notes_storage["abcdef"] = {"content": "existing"}
@@ -65,3 +64,40 @@ def test_create_note_returns_error_after_repeated_note_id_collisions(
"note_id": None,
}
assert notes_tools._notes_storage == {"abcdef": {"content": "existing"}}
def test_create_note_records_author() -> None:
result = notes_tools._create_note_impl("t", "c", agent_id="agent-1", agent_name="Agent One")
note = notes_tools._notes_storage[result["note_id"]]
assert note["agent_id"] == "agent-1"
assert note["agent_name"] == "Agent One"
def test_list_notes_exposes_author_and_flags_caller() -> None:
notes_tools._create_note_impl("mine", "c", agent_id="agent-1", agent_name="Agent One")
notes_tools._create_note_impl("theirs", "c", agent_id="agent-2", agent_name="Agent Two")
result = notes_tools._list_notes_impl(caller_agent_id="agent-1")
by_title = {n["title"]: n for n in result["notes"]}
assert by_title["mine"]["agent_name"] == "Agent One"
assert by_title["mine"].get("by_you") is True
assert by_title["theirs"]["agent_name"] == "Agent Two"
assert "by_you" not in by_title["theirs"]
def test_list_notes_without_author_has_no_attribution() -> None:
notes_tools._create_note_impl("anon", "c")
entry = notes_tools._list_notes_impl(caller_agent_id="agent-1")["notes"][0]
assert "agent_name" not in entry
assert "by_you" not in entry
def test_get_note_flags_caller_ownership() -> None:
note_id = notes_tools._create_note_impl(
"mine", "c", agent_id="agent-1", agent_name="Agent One"
)["note_id"]
mine = notes_tools._get_note_impl(note_id, caller_agent_id="agent-1")
assert mine["note"].get("by_you") is True
assert mine["note"]["agent_name"] == "Agent One"
theirs = notes_tools._get_note_impl(note_id, caller_agent_id="agent-9")
assert "by_you" not in theirs["note"]