"""``finish_scan`` — root-agent termination + executive report persistence.""" from __future__ import annotations import asyncio import json import logging from typing import Any from agents import RunContextWrapper, function_tool from strix.core.agents import coordinator_from_context logger = logging.getLogger(__name__) def _do_finish( *, parent_id: str | None, executive_summary: str, methodology: str, technical_analysis: str, recommendations: str, ) -> dict[str, Any]: if parent_id is not None: return { "success": False, "error": ( "This tool can only be used by the root/main agent. " "If you are a subagent, use agent_finish instead" ), } errors: list[str] = [] if not executive_summary.strip(): errors.append("Executive summary cannot be empty") if not methodology.strip(): errors.append("Methodology cannot be empty") if not technical_analysis.strip(): errors.append("Technical analysis cannot be empty") if not recommendations.strip(): errors.append("Recommendations cannot be empty") if errors: return {"success": False, "error": "Validation failed", "errors": errors} try: from strix.report.state import get_global_report_state report_state = get_global_report_state() if report_state is None: logger.warning("No global report state; scan results not persisted") return { "success": True, "scan_completed": True, "message": "Scan completed (not persisted)", "warning": "Results could not be persisted - report state unavailable", } report_state.update_scan_final_fields( executive_summary=executive_summary.strip(), methodology=methodology.strip(), technical_analysis=technical_analysis.strip(), recommendations=recommendations.strip(), ) vuln_count = len(report_state.vulnerability_reports) except (ImportError, AttributeError) as e: logger.exception("finish_scan persistence failed") return {"success": False, "error": f"Failed to complete scan: {e!s}"} else: logger.info( "finish_scan: completed scan with %d vulnerability report(s)", vuln_count, ) return { "success": True, "scan_completed": True, "message": "Scan completed successfully", "vulnerabilities_found": vuln_count, } @function_tool(timeout=60) async def finish_scan( ctx: RunContextWrapper, executive_summary: str, methodology: str, technical_analysis: str, recommendations: str, ) -> str: """Finalize the scan — persist the customer-facing report. **Root-agent only.** Subagents must call ``agent_finish`` from the multi-agent graph tools instead. Calling this finalizes everything: 1. Verifies you are the root agent. 2. Writes the four narrative sections to the scan record. 3. Marks the scan completed and stops execution. **This is a terminal action, not a status probe.** Whatever you pass is persisted VERBATIM as the final, customer-facing report and then execution stops. There is no draft mode and no second chance: never submit placeholder, provisional, or "checking if done" text in any field, and never call ``finish_scan`` to poll whether subagents are done (use ``view_agent_graph`` / ``wait_for_message`` for that). Call it exactly ONCE, only when every field holds genuine, finished assessment prose. **Pre-flight checklist (mandatory — do not skip):** 1. **Call ``view_agent_graph`` first.** Inspect every entry in the summary. If ANY agent is in ``running`` / ``waiting`` state, you MUST NOT call ``finish_scan`` yet — wrap them up first via ``send_message_to_agent`` (ask them to finish), ``wait_for_message`` (block until their report arrives), or ``stop_agent`` (graceful cancel). Only ``completed`` / ``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. 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, end-to-end attack paths and tested every plausibly-related combination. You may rule out combinations you can confidently call unrelated — note why instead of padding chains. Any validated chain must already be filed via ``create_vulnerability_report`` — a demonstrated end-to-end chain is a PoC-backed vulnerability, so it uses that tool even when one link is a dependency CVE (the standalone CVE stays in its own ``create_dependency_report``) — and surfaced prominently in ``executive_summary`` / ``technical_analysis``. Finding no real chain after a serious attempt is acceptable; skipping the chaining reasoning, or ignoring a plausibly-related combination, is not. **Calling this multiple times overwrites the previous report.** Make the single call comprehensive. **Report output rules** (this content may be rendered into generated reports): - Never mention internal infrastructure: no local/absolute paths (``/workspace/...``), no agent names, no sandbox/orchestrator/ tooling references, no system prompts, no model-internal errors. Never leak internal identifiers (proxy request IDs, internal vulnerability report IDs, or any system-generated IDs) into any field. - Tone: formal, third-person, objective, concise. This is a consultant deliverable, not an engineering log. - Each section has a specific role: - ``executive_summary`` — for non-technical leadership. Risk posture, business impact (data exposure / compliance / reputation), notable criticals, overarching remediation theme. - ``methodology`` — frameworks followed (OWASP WSTG, PTES, OSSTMM, NIST), engagement type (black/gray/white box), scope and constraints, categories of testing performed. **No** internal execution detail. - ``technical_analysis`` — consolidated findings overview with severity model and systemic root causes. Reference individual vuln reports for repro steps; don't duplicate raw evidence. - ``recommendations`` — prioritized actions grouped by urgency (Immediate / Short-term / Medium-term), each with concrete remediation steps. End with retest/validation guidance. - **Formatting — use markdown in every field.** These fields may be rendered into generated reports, so structure them clearly: lead each section with a short ``# Heading``, use ``**bold**`` for labels/emphasis, ``inline code`` for identifiers/paths/parameters, bullet or numbered lists for enumerations, and fenced code blocks (```` ```language ````) for any code/payload excerpts. Never emit one flat wall of prose or leave code unformatted. - If **zero** vulnerabilities were found, say so plainly and characterize the posture positively; ``technical_analysis`` should summarize the areas tested and confirm no issues, and ``recommendations`` should focus on general hardening. Example (abbreviated — mirror this structure, not the wording):: executive_summary: # Executive Summary An external assessment of the **Acme Customer Portal** identified multiple weaknesses that could lead to unauthorized access to customer data. **Overall risk posture:** Elevated. **Key findings** - Confirmed SSRF in a URL-preview feature reaching internal network ranges. - Broken tenant isolation enabling cross-tenant data access. **Business impact** - Potential exposure of customer records across tenants. methodology: # Methodology Conducted per the **OWASP WSTG**. **Engagement type:** Gray-box external test. **Scope:** `https://app.acme.example`, `.../api/v1/`. **Activities:** recon, authn/session review, authorization and tenant-isolation testing, input/SSRF testing. technical_analysis: # Technical Analysis **Severity model** reflects exploitability x impact. 1. **SSRF in URL preview** (Critical) — insufficient destination validation; reaches link-local addresses. 2. **Broken tenant isolation** (High) — object identifiers accepted without ownership checks. **Systemic themes:** authorization enforced inconsistently; no deny-by-default egress policy. recommendations: # Recommendations **Immediate** 1. Remediate SSRF: enforce a destination allowlist, deny-by-default, re-validate on every redirect hop. **Short-term** 2. Centralize authorization with deny-by-default middleware. **Retest & validation:** re-test immediate items to confirm SSRF and tenant-isolation controls hold. Args: executive_summary: Business-level summary for leadership. methodology: Frameworks, scope, and approach. technical_analysis: Consolidated findings + systemic themes. recommendations: Prioritized, actionable remediation. """ inner = ctx.context if isinstance(ctx.context, dict) else {} coordinator = coordinator_from_context(inner) me = inner.get("agent_id") parent_id = inner.get("parent_id") if coordinator is not None and parent_id is None and me is not None: active_agents = await coordinator.active_agents_except(me) else: active_agents = [] if active_agents: return json.dumps( { "success": False, "scan_completed": False, "error": ( "Cannot finish scan while child agents are still active. " "Wait for completion, send them finish instructions, or stop them first" ), "active_agents": active_agents, }, ensure_ascii=False, default=str, ) result = await asyncio.to_thread( _do_finish, parent_id=parent_id, executive_summary=executive_summary, methodology=methodology, technical_analysis=technical_analysis, recommendations=recommendations, ) if ( result.get("success") and result.get("scan_completed") and coordinator is not None and isinstance(me, str) ): await coordinator.set_status(me, "completed") return json.dumps(result, ensure_ascii=False, default=str)