forkless/strix
1
0
Fork 0
mirror of https://github.com/usestrix/strix.git synced 2026-08-24 03:42:37 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor: nuke gratuitous XML serialization + delete argument_parser

Browse Source 
Argument parser:
- Delete ``strix/tools/argument_parser.py`` and its tests. The SDK
  validates and types tool arguments via Pydantic before they hit our
  wrappers, and the in-container tool server receives JSON-typed
  kwargs over the wire. The string-coercion belt-and-suspenders is no
  longer pulling its weight.

XML → JSON / typed structures:
- ``create_vulnerability_report``: ``cvss_breakdown`` is now a
  ``dict[str, str]`` of the 8 metrics; ``code_locations`` is a
  ``list[dict]``. No more XML parsing in the tool or the renderer.
- ``check_duplicate``: the dedup judge now emits a single JSON object
  instead of an ``<dedupe_result>`` block. Strict JSON parser handles
  optional code-fence wrappers.
- ``agent_finish``: completion report posted to the parent inbox is a
  JSON object (``kind``, ``from``, ``agent_id``, ``success``,
  ``summary``, ``findings``, ``recommendations``) rather than a
  hand-rolled ``<agent_completion_report>`` XML envelope.
- ``create_agent``: identity preamble + inherited-context markers are
  plain bracketed labels rather than ``<agent_delegation>`` /
  ``<inherited_context_from_parent>`` envelopes.
- ``inject_messages_filter``: peer messages get a
  ``[Message from agent <id> | type=... | priority=...]`` header line
  instead of an ``<inter_agent_message>`` envelope.
- Crash + system-warning messages: bracketed labels, no XML.
- System prompt: the inter-agent block now describes the new header
  format and drops the "never echo XML envelope" rule.
- ``strix/llm/utils.py``: deleted. ``clean_content`` collapsed into a
  one-line blank-line normalizer in the agent-message renderer (the
  XML envelope scrub had nothing left to scrub).

Tests updated to match the new shapes.
This commit is contained in:
0xallam
2026-04-25 12:31:07 -07:00
parent e473b2d6d8
commit 1aeee5dc29
17 changed files with 219 additions and 667 deletions
Download Patch File Download Diff File Expand all files Collapse all files
strix/agents/prompts/system_prompt.jinja
+2 -3
View File
@@ -16,9 +16,8 @@ CLI OUTPUT:
- NEVER use "Strix" or any identifiable names/markers in HTTP requests, payloads, user-agents, or any inputs
INTER-AGENT MESSAGES:
- NEVER echo inter_agent_message or agent_completion_report blocks that are sent to you in your output.
- Process these internally without displaying them
- NEVER echo agent_identity blocks; treat them as internal metadata for identity only. Do not include them in outputs or tool calls.
- Messages from other agents arrive prefixed with a header like `[Message from agent <name> | type=... | priority=...]`. Treat them as internal context — never repeat them verbatim in your own output.
- Treat agent identity / inherited-context preambles as internal metadata; do not echo them in outputs or tool calls.
- Minimize inter-agent messaging: only message when essential for coordination or assistance; avoid routine status updates; batch non-urgent information; prefer parent/child completion flows and shared artifacts over messaging
{% if interactive %}
strix/interface/tool_components/agent_message_renderer.py
+5 -5
View File
@@ -1,3 +1,4 @@
import re
from functools import cache
from typing import Any, ClassVar
@@ -11,6 +12,9 @@ from .base_renderer import BaseToolRenderer
from .registry import register_tool_renderer
_BLANK_LINE_RUNS = re.compile(r"\n\s*\n")
_HEADER_STYLES = [
("###### ", 7, "bold #4ade80"),
("##### ", 6, "bold #22c55e"),
@@ -180,11 +184,7 @@ class AgentMessageRenderer(BaseToolRenderer):
def render_simple(cls, content: str) -> Text:
if not content:
return Text()
from strix.llm.utils import clean_content
cleaned = clean_content(content)
cleaned = _BLANK_LINE_RUNS.sub("\n\n", content).strip()
if not cleaned:
return Text()
return _apply_markdown_styles(cleaned)
strix/interface/tool_components/reporting_renderer.py
+18 -17
View File
@@ -6,17 +6,22 @@ from pygments.styles import get_style_by_name
from rich.text import Text
from textual.widgets import Static
from strix.tools.reporting.tool import (
_parse_code_locations_xml as parse_code_locations_xml,
)
from strix.tools.reporting.tool import (
_parse_cvss_xml as parse_cvss_xml,
)
from .base_renderer import BaseToolRenderer
from .registry import register_tool_renderer
def _coerce_dict(value: Any) -> dict[str, Any]:
if isinstance(value, dict):
return value
return {}
def _coerce_list_of_dicts(value: Any) -> list[dict[str, Any]]:
if isinstance(value, list):
return [item for item in value if isinstance(item, dict)]
return []
@cache
def _get_style_colors() -> dict[Any, str]:
style = get_style_by_name("native")
@@ -94,8 +99,8 @@ class CreateVulnerabilityReportRenderer(BaseToolRenderer):
poc_script_code = args.get("poc_script_code", "")
remediation_steps = args.get("remediation_steps", "")
cvss_breakdown_xml = args.get("cvss_breakdown", "")
code_locations_xml = args.get("code_locations", "")
cvss_breakdown = _coerce_dict(args.get("cvss_breakdown"))
code_locations = _coerce_list_of_dicts(args.get("code_locations"))
endpoint = args.get("endpoint", "")
method = args.get("method", "")
@@ -154,8 +159,7 @@ class CreateVulnerabilityReportRenderer(BaseToolRenderer):
text.append("CWE: ", style=FIELD_STYLE)
text.append(cwe)
parsed_cvss = parse_cvss_xml(cvss_breakdown_xml) if cvss_breakdown_xml else None
if parsed_cvss:
if cvss_breakdown:
text.append("\n\n")
cvss_parts = []
for key, prefix in [
@@ -168,7 +172,7 @@ class CreateVulnerabilityReportRenderer(BaseToolRenderer):
("integrity", "I"),
("availability", "A"),
]:
val = parsed_cvss.get(key)
val = cvss_breakdown.get(key)
if val:
cvss_parts.append(f"{prefix}:{val}")
text.append("CVSS Vector: ", style=FIELD_STYLE)
@@ -192,13 +196,10 @@ class CreateVulnerabilityReportRenderer(BaseToolRenderer):
text.append("\n")
text.append(technical_analysis)
parsed_locations = (
parse_code_locations_xml(code_locations_xml) if code_locations_xml else None
)
if parsed_locations:
if code_locations:
text.append("\n\n")
text.append("Code Locations", style=FIELD_STYLE)
for i, loc in enumerate(parsed_locations):
for i, loc in enumerate(code_locations):
text.append("\n\n")
text.append(f" Location {i + 1}: ", style=DIM_STYLE)
text.append(loc.get("file", "unknown"), style=FILE_STYLE)
strix/llm/dedupe.py
+39 -51
View File
@@ -1,6 +1,5 @@
import json
import logging
import re
from typing import Any
import litellm
@@ -49,30 +48,30 @@ FIELDS TO ANALYZE:
- poc_description: How it's exploited
- impact: What damage it can cause
YOU MUST RESPOND WITH EXACTLY THIS XML FORMAT AND NOTHING ELSE:
Respond with a single JSON object and nothing else:
<dedupe_result>
<is_duplicate>true</is_duplicate>
<duplicate_id>vuln-0001</duplicate_id>
<confidence>0.95</confidence>
<reason>Both reports describe SQL injection in /api/login via the username parameter</reason>
</dedupe_result>
{
"is_duplicate": true,
"duplicate_id": "vuln-0001",
"confidence": 0.95,
"reason": "Both reports describe SQL injection in /api/login via the username parameter"
}
OR if not a duplicate:
Or, if not a duplicate:
<dedupe_result>
<is_duplicate>false</is_duplicate>
<duplicate_id></duplicate_id>
<confidence>0.90</confidence>
<reason>Different endpoints: candidate is /api/search, existing is /api/login</reason>
</dedupe_result>
{
"is_duplicate": false,
"duplicate_id": "",
"confidence": 0.90,
"reason": "Different endpoints: candidate is /api/search, existing is /api/login"
}
RULES:
- is_duplicate MUST be exactly "true" or "false" (lowercase)
- duplicate_id MUST be the exact ID from existing reports or empty if not duplicate
- confidence MUST be a decimal (your confidence level in the decision)
- reason MUST be a specific explanation mentioning endpoint/parameter/root cause
- DO NOT include any text outside the <dedupe_result> tags"""
Rules:
- ``is_duplicate`` is a boolean.
- ``duplicate_id`` is the exact id from existing reports, or "" if not a duplicate.
- ``confidence`` is a number between 0 and 1.
- ``reason`` is a specific explanation mentioning endpoint/parameter/root cause.
- Output ONLY the JSON object — no surrounding prose, no code fences."""
def _prepare_report_for_comparison(report: dict[str, Any]) -> dict[str, Any]:
@@ -99,42 +98,31 @@ def _prepare_report_for_comparison(report: dict[str, Any]) -> dict[str, Any]:
return cleaned
def _extract_xml_field(content: str, field: str) -> str:
pattern = rf"<{field}>(.*?)</{field}>"
match = re.search(pattern, content, re.DOTALL | re.IGNORECASE)
if match:
return match.group(1).strip()
return ""
def _parse_dedupe_response(content: str) -> dict[str, Any]:
result_match = re.search(
r"<dedupe_result>(.*?)</dedupe_result>", content, re.DOTALL | re.IGNORECASE
)
if not result_match:
logger.warning(f"No <dedupe_result> block found in response: {content[:500]}")
raise ValueError("No <dedupe_result> block found in response")
result_content = result_match.group(1)
is_duplicate_str = _extract_xml_field(result_content, "is_duplicate")
duplicate_id = _extract_xml_field(result_content, "duplicate_id")
confidence_str = _extract_xml_field(result_content, "confidence")
reason = _extract_xml_field(result_content, "reason")
is_duplicate = is_duplicate_str.lower() == "true"
text = content.strip()
if text.startswith("```"):
text = text.strip("`")
if text.lower().startswith("json"):
text = text[4:]
text = text.strip()
start = text.find("{")
end = text.rfind("}")
if start == -1 or end == -1 or end <= start:
raise ValueError(f"No JSON object found in dedupe response: {content[:500]}")
parsed = json.loads(text[start : end + 1])
duplicate_id = str(parsed.get("duplicate_id") or "")[:64]
reason = str(parsed.get("reason") or "")[:500]
try:
confidence = float(confidence_str) if confidence_str else 0.0
except ValueError:
confidence = float(parsed.get("confidence", 0.0))
except (TypeError, ValueError):
confidence = 0.0
return {
"is_duplicate": is_duplicate,
"duplicate_id": duplicate_id[:64] if duplicate_id else "",
"is_duplicate": bool(parsed.get("is_duplicate", False)),
"duplicate_id": duplicate_id,
"confidence": confidence,
"reason": reason[:500] if reason else "",
"reason": reason,
}
@@ -165,7 +153,7 @@ def check_duplicate(
"content": (
f"Compare this candidate vulnerability against existing reports:\n\n"
f"{json.dumps(comparison_data, indent=2)}\n\n"
f"Respond with ONLY the <dedupe_result> XML block."
f"Respond with ONLY the JSON object described in the system prompt."
),
},
]
strix/llm/utils.py
-26
View File
@@ -1,26 +0,0 @@
"""Helpers for the TUI message renderer.
The model occasionally echoes inter-agent XML envelopes in plain text
despite the system prompt's "don't echo" rule, so :func:`clean_content`
strips them defensively before display.
"""
import re
_HIDDEN_XML_PATTERNS = [
re.compile(r"<inter_agent_message>.*?</inter_agent_message>", re.DOTALL | re.IGNORECASE),
re.compile(
r"<agent_completion_report>.*?</agent_completion_report>",
re.DOTALL | re.IGNORECASE,
),
]
_BLANK_LINE_RUNS = re.compile(r"\n\s*\n")
def clean_content(content: str) -> str:
if not content:
return ""
for pattern in _HIDDEN_XML_PATTERNS:
content = pattern.sub("", content)
return _BLANK_LINE_RUNS.sub("\n\n", content).strip()
strix/orchestration/filter.py
+7 -12
View File
@@ -23,11 +23,9 @@ logger = logging.getLogger(__name__)
async def inject_messages_filter(data: CallModelData) -> ModelInputData:
"""Drain bus inbox and append messages as user-role items before the LLM call.
Each drained message is wrapped in an ``<inter_agent_message>`` XML envelope
so the system prompt's rules around inter-agent communication apply.
Messages from the literal sender ``"user"`` (a real human via TUI)
skip the XML wrap and are added as plain user messages.
Messages from peer agents are formatted with a labeled header so the
receiving model can attribute them. Messages from the literal sender
``"user"`` (a real human via TUI) are added as plain user messages.
Any exception inside the filter — malformed message dict, bug in
``bus.drain``, etc. — is caught and the original ``data.model_data``
@@ -51,16 +49,13 @@ async def inject_messages_filter(data: CallModelData) -> ModelInputData:
if sender == "user":
new_input.append({"role": "user", "content": content})
else:
msg_type = msg.get("type", "info")
priority = msg.get("priority", "normal")
header = f"[Message from agent {sender} | type={msg_type} | priority={priority}]"
new_input.append(
{
"role": "user",
"content": (
f"<inter_agent_message from='{sender}' "
f"type='{msg.get('type', 'info')}' "
f"priority='{msg.get('priority', 'normal')}'>"
f"{content}"
f"</inter_agent_message>"
),
"content": f"{header}\n{content}",
}
)
return ModelInputData(
strix/orchestration/hooks.py
+9 -12
View File
@@ -26,9 +26,9 @@ class StrixOrchestrationHooks(RunHooks[Any]):
the agent doesn't fire tools before Caido and the tool server
are ready.
4. Subagent crash detection: if ``on_agent_end`` fires without
``agent_finish_called`` being set, posts a synthetic
``<agent_crash>`` message to the parent's inbox so the parent
learns on its next turn instead of waiting forever.
``agent_finish_called`` being set, posts a crash message to the
parent's inbox so the parent learns on its next turn instead of
waiting forever.
"""
async def on_llm_start(
@@ -51,8 +51,8 @@ class StrixOrchestrationHooks(RunHooks[Any]):
{
"role": "user",
"content": (
"<system_warning>You are at 85% of your iteration "
"budget. Begin consolidating findings.</system_warning>"
"[System warning] You are at 85% of your iteration "
"budget. Begin consolidating findings."
),
}
)
@@ -61,9 +61,8 @@ class StrixOrchestrationHooks(RunHooks[Any]):
{
"role": "user",
"content": (
"<system_warning>You have 3 iterations left. Your "
"[System warning] You have 3 iterations left. Your "
"next tool call MUST be the finish tool."
"</system_warning>"
),
}
)
@@ -148,11 +147,9 @@ class StrixOrchestrationHooks(RunHooks[Any]):
{
"from": me,
"content": (
f"<agent_crash agent_id='{me}' "
f"name='{bus.names.get(me, me)}'>"
"Agent terminated without calling agent_finish. "
"Stop waiting on this child."
"</agent_crash>"
f"[Agent crash] {bus.names.get(me, me)} ({me}) "
f"terminated without calling agent_finish. "
f"Stop waiting on this child."
),
"type": "crash",
},
strix/runtime/tool_server.py
+1 -3
View File
@@ -69,7 +69,6 @@ class ToolExecutionResponse(BaseModel):
async def _run_tool(agent_id: str, tool_name: str, kwargs: dict[str, Any]) -> Any:
from strix.tools.argument_parser import convert_arguments
from strix.tools.context import set_current_agent_id
from strix.tools.registry import get_tool_by_name
@@ -79,8 +78,7 @@ async def _run_tool(agent_id: str, tool_name: str, kwargs: dict[str, Any]) -> An
if not tool_func:
raise ValueError(f"Tool '{tool_name}' not found")
converted_kwargs = convert_arguments(tool_func, kwargs)
return await asyncio.to_thread(tool_func, **converted_kwargs)
return await asyncio.to_thread(tool_func, **kwargs)
@app.post("/execute", response_model=ToolExecutionResponse)
strix/tools/agents_graph/tools.py
+18 -22
View File
@@ -371,33 +371,29 @@ async def create_agent(
await bus.register(child_id, name, parent_id)
# Build the child's input. Identity injection mirrors the legacy
# <agent_delegation> envelope so the child's system prompt's existing
# rules around self-identity still apply.
parent_history = inner.get("parent_input_items") if inherit_context else None
initial_input: list[TResponseInputItem] = []
if parent_history:
initial_input.append(
{
"role": "user",
"content": "<inherited_context_from_parent>",
"content": "[Inherited context from parent — read-only history]",
}
)
initial_input.extend(parent_history)
initial_input.append(
{
"role": "user",
"content": "</inherited_context_from_parent>",
"content": "[End of inherited context]",
}
)
initial_input.append(
{
"role": "user",
"content": (
f"<agent_delegation>\n"
f"You are agent {name} ({child_id}). Parent is {parent_id}.\n"
f"Maintain self-identity. Use agent_finish when complete.\n"
f"</agent_delegation>"
f"You are agent {name} ({child_id}); your parent is {parent_id}. "
f"Maintain your own identity. Call agent_finish when your task "
f"is complete."
),
}
)
@@ -471,8 +467,8 @@ async def agent_finish(
this tool refuses to run for root agents. Calling this:
1. Marks the subagent as ``completed``.
2. Posts a structured ``<agent_completion_report>`` to the
parent's inbox (when ``report_to_parent`` is true).
2. Posts a structured completion report to the parent's inbox
(when ``report_to_parent`` is true).
3. Stops this subagent's execution.
**Vulnerability findings must already be filed via
@@ -522,19 +518,19 @@ async def agent_finish(
parent_notified = False
if report_to_parent:
findings_xml = "\n".join(f" <finding>{f}</finding>" for f in (findings or []))
rec_xml = "\n".join(
f" <recommendation>{r}</recommendation>" for r in (final_recommendations or [])
)
async with bus._lock:
agent_name = bus.names.get(me, me)
report = (
f"<agent_completion_report from='{agent_name}' agent_id='{me}' "
f"success='{success}'>\n"
f" <summary>{result_summary}</summary>\n"
f" <findings>\n{findings_xml}\n </findings>\n"
f" <recommendations>\n{rec_xml}\n </recommendations>\n"
f"</agent_completion_report>"
report = json.dumps(
{
"kind": "agent_completion_report",
"from": agent_name,
"agent_id": me,
"success": success,
"summary": result_summary,
"findings": list(findings or []),
"recommendations": list(final_recommendations or []),
},
ensure_ascii=False,
)
await bus.send(
parent_id,
strix/tools/argument_parser.py
-121
View File
@@ -1,121 +0,0 @@
import contextlib
import inspect
import json
import types
from collections.abc import Callable
from typing import Any, Union, get_args, get_origin
class ArgumentConversionError(Exception):
def __init__(self, message: str, param_name: str | None = None) -> None:
self.param_name = param_name
super().__init__(message)
def convert_arguments(func: Callable[..., Any], kwargs: dict[str, Any]) -> dict[str, Any]:
try:
sig = inspect.signature(func)
converted = {}
for param_name, value in kwargs.items():
if param_name not in sig.parameters:
converted[param_name] = value
continue
param = sig.parameters[param_name]
param_type = param.annotation
if param_type == inspect.Parameter.empty or value is None:
converted[param_name] = value
continue
if not isinstance(value, str):
converted[param_name] = value
continue
try:
converted[param_name] = convert_string_to_type(value, param_type)
except (ValueError, TypeError, json.JSONDecodeError) as e:
raise ArgumentConversionError(
f"Failed to convert argument '{param_name}' to type {param_type}: {e}",
param_name=param_name,
) from e
except (ValueError, TypeError, AttributeError) as e:
raise ArgumentConversionError(f"Failed to process function arguments: {e}") from e
return converted
def convert_string_to_type(value: str, param_type: Any) -> Any:
origin = get_origin(param_type)
if origin is Union or isinstance(param_type, types.UnionType):
args = get_args(param_type)
for arg_type in args:
if arg_type is not type(None):
with contextlib.suppress(ValueError, TypeError, json.JSONDecodeError):
return convert_string_to_type(value, arg_type)
return value
if hasattr(param_type, "__args__"):
args = getattr(param_type, "__args__", ())
if len(args) == 2 and type(None) in args:
non_none_type = args[0] if args[1] is type(None) else args[1]
with contextlib.suppress(ValueError, TypeError, json.JSONDecodeError):
return convert_string_to_type(value, non_none_type)
return value
return _convert_basic_types(value, param_type, origin)
def _convert_basic_types(value: str, param_type: Any, origin: Any = None) -> Any:
basic_type_converters: dict[Any, Callable[[str], Any]] = {
int: int,
float: float,
bool: _convert_to_bool,
str: str,
}
if param_type in basic_type_converters:
return basic_type_converters[param_type](value)
if list in (origin, param_type):
return _convert_to_list(value)
if dict in (origin, param_type):
return _convert_to_dict(value)
with contextlib.suppress(json.JSONDecodeError):
return json.loads(value)
return value
def _convert_to_bool(value: str) -> bool:
if value.lower() in ("true", "1", "yes", "on"):
return True
if value.lower() in ("false", "0", "no", "off"):
return False
return bool(value)
def _convert_to_list(value: str) -> list[Any]:
try:
parsed = json.loads(value)
if isinstance(parsed, list):
return parsed
except json.JSONDecodeError:
if "," in value:
return [item.strip() for item in value.split(",")]
return [value]
else:
return [parsed]
def _convert_to_dict(value: str) -> dict[str, Any]:
try:
parsed = json.loads(value)
if isinstance(parsed, dict):
return parsed
except json.JSONDecodeError:
return {}
else:
return {}
strix/tools/reporting/tool.py
+95 -102
View File
@@ -1,15 +1,8 @@
"""``create_vulnerability_report`` — file a vuln finding with dedup + CVSS.
Validates required fields, parses the CVSS-3.1 XML breakdown into a
score, runs LLM-based dedup against existing reports through
``strix.llm.dedupe.check_duplicate``, and persists via the global
:class:`strix.telemetry.tracer.Tracer` instance.
"""
"""``create_vulnerability_report`` — file a vuln finding with dedup + CVSS."""
from __future__ import annotations
import asyncio
import contextlib
import json
import logging
import re
@@ -24,63 +17,29 @@ from strix.tools._decorator import strix_tool
logger = logging.getLogger(__name__)
_CVSS_FIELDS = (
"attack_vector",
"attack_complexity",
"privileges_required",
"user_interaction",
"scope",
"confidentiality",
"integrity",
"availability",
_CVSS_VALID = {
"attack_vector": ["N", "A", "L", "P"],
"attack_complexity": ["L", "H"],
"privileges_required": ["N", "L", "H"],
"user_interaction": ["N", "R"],
"scope": ["U", "C"],
"confidentiality": ["N", "L", "H"],
"integrity": ["N", "L", "H"],
"availability": ["N", "L", "H"],
}
_CODE_LOCATION_FIELDS = (
"file",
"start_line",
"end_line",
"snippet",
"label",
"fix_before",
"fix_after",
)
def _parse_cvss_xml(xml_str: str) -> dict[str, str] | None:
if not xml_str or not xml_str.strip():
return None
result: dict[str, str] = {}
for field in _CVSS_FIELDS:
match = re.search(rf"<{field}>(.*?)</{field}>", xml_str, re.DOTALL)
if match:
result[field] = match.group(1).strip()
return result if result else None
def _parse_code_locations_xml(xml_str: str) -> list[dict[str, Any]] | None:
if not xml_str or not xml_str.strip():
return None
locations: list[dict[str, Any]] = []
for loc_match in re.finditer(r"<location>(.*?)</location>", xml_str, re.DOTALL):
loc: dict[str, Any] = {}
loc_content = loc_match.group(1)
for field in (
"file",
"start_line",
"end_line",
"snippet",
"label",
"fix_before",
"fix_after",
):
field_match = re.search(rf"<{field}>(.*?)</{field}>", loc_content, re.DOTALL)
if field_match:
raw = field_match.group(1)
value = (
raw.strip("\n")
if field in ("snippet", "fix_before", "fix_after")
else raw.strip()
)
if field in ("start_line", "end_line"):
with contextlib.suppress(ValueError, TypeError):
loc[field] = int(value)
elif value:
loc[field] = value
if loc.get("file") and loc.get("start_line") is not None:
locations.append(loc)
return locations if locations else None
def _validate_file_path(path: str) -> str | None:
if not path or not path.strip():
return "file path cannot be empty"
@@ -92,6 +51,36 @@ def _validate_file_path(path: str) -> str | None:
return None
def _normalize_code_locations(
raw: list[dict[str, Any]] | None,
) -> list[dict[str, Any]] | None:
if not raw:
return None
cleaned: list[dict[str, Any]] = []
for loc in raw:
normalized: dict[str, Any] = {}
for field in _CODE_LOCATION_FIELDS:
if field not in loc or loc[field] is None:
continue
value = loc[field]
if field in ("start_line", "end_line"):
try:
normalized[field] = int(value)
except (TypeError, ValueError):
continue
else:
text = (
str(value).strip("\n")
if field in ("snippet", "fix_before", "fix_after")
else str(value).strip()
)
if text:
normalized[field] = text
if normalized.get("file") and normalized.get("start_line") is not None:
cleaned.append(normalized)
return cleaned or None
def _validate_code_locations(locations: list[dict[str, Any]]) -> list[str]:
errors: list[str] = []
for i, loc in enumerate(locations):
@@ -133,15 +122,15 @@ def _validate_cwe(cwe: str) -> str | None:
return None
def _calculate_cvss(**kwargs: str) -> tuple[float, str, str]:
def _calculate_cvss(breakdown: dict[str, str]) -> tuple[float, str, str]:
try:
from cvss import CVSS3
vector = (
f"CVSS:3.1/AV:{kwargs['attack_vector']}/AC:{kwargs['attack_complexity']}/"
f"PR:{kwargs['privileges_required']}/UI:{kwargs['user_interaction']}/"
f"S:{kwargs['scope']}/C:{kwargs['confidentiality']}/"
f"I:{kwargs['integrity']}/A:{kwargs['availability']}"
f"CVSS:3.1/AV:{breakdown['attack_vector']}/AC:{breakdown['attack_complexity']}/"
f"PR:{breakdown['privileges_required']}/UI:{breakdown['user_interaction']}/"
f"S:{breakdown['scope']}/C:{breakdown['confidentiality']}/"
f"I:{breakdown['integrity']}/A:{breakdown['availability']}"
)
c = CVSS3(vector)
score = c.scores()[0]
@@ -165,18 +154,6 @@ _REQUIRED_FIELDS = {
}
_CVSS_VALID = {
"attack_vector": ["N", "A", "L", "P"],
"attack_complexity": ["L", "H"],
"privileges_required": ["N", "L", "H"],
"user_interaction": ["N", "R"],
"scope": ["U", "C"],
"confidentiality": ["N", "L", "H"],
"integrity": ["N", "L", "H"],
"availability": ["N", "L", "H"],
}
def _do_create( # noqa: PLR0912
*,
title: str,
@@ -187,12 +164,12 @@ def _do_create( # noqa: PLR0912
poc_description: str,
poc_script_code: str,
remediation_steps: str,
cvss_breakdown: str,
cvss_breakdown: dict[str, str],
endpoint: str | None,
method: str | None,
cve: str | None,
cwe: str | None,
code_locations: str | None,
code_locations: list[dict[str, Any]] | None,
) -> dict[str, Any]:
errors: list[str] = []
fields = {
@@ -209,16 +186,16 @@ def _do_create( # noqa: PLR0912
if not str(fields.get(name) or "").strip():
errors.append(msg)
parsed_cvss = _parse_cvss_xml(cvss_breakdown)
if not parsed_cvss:
errors.append("cvss: could not parse CVSS breakdown XML")
if not isinstance(cvss_breakdown, dict) or not cvss_breakdown:
errors.append("cvss_breakdown: must be an object with the 8 CVSS metrics")
cvss_breakdown = {}
else:
for name, valid in _CVSS_VALID.items():
value = parsed_cvss.get(name)
value = cvss_breakdown.get(name)
if value not in valid:
errors.append(f"Invalid {name}: {value}. Must be one of: {valid}")
parsed_locations = _parse_code_locations_xml(code_locations) if code_locations else None
parsed_locations = _normalize_code_locations(code_locations)
if parsed_locations:
errors.extend(_validate_code_locations(parsed_locations))
if cve:
@@ -235,8 +212,7 @@ def _do_create( # noqa: PLR0912
if errors:
return {"success": False, "message": "Validation failed", "errors": errors}
assert parsed_cvss is not None
cvss_score, severity, _vector = _calculate_cvss(**parsed_cvss)
cvss_score, severity, _vector = _calculate_cvss(cvss_breakdown)
try:
from strix.telemetry.tracer import get_global_tracer
@@ -294,7 +270,7 @@ def _do_create( # noqa: PLR0912
poc_script_code=poc_script_code,
remediation_steps=remediation_steps,
cvss=cvss_score,
cvss_breakdown=parsed_cvss,
cvss_breakdown=cvss_breakdown,
endpoint=endpoint,
method=method,
cve=cve,
@@ -315,7 +291,9 @@ def _do_create( # noqa: PLR0912
# Generous timeout: the dedup check makes a separate LLM call, and
# large scans can have many existing reports to compare against.
@strix_tool(timeout=180)
# strict_mode=False because cvss_breakdown is a dict[str, str] and
# code_locations is list[dict] — both free-form for the strict schema.
@strix_tool(timeout=180, strict_mode=False)
async def create_vulnerability_report(
ctx: RunContextWrapper,
title: str,
@@ -326,12 +304,12 @@ async def create_vulnerability_report(
poc_description: str,
poc_script_code: str,
remediation_steps: str,
cvss_breakdown: str,
cvss_breakdown: dict[str, str],
endpoint: str | None = None,
method: str | None = None,
cve: str | None = None,
cwe: str | None = None,
code_locations: str | None = None,
code_locations: list[dict[str, Any]] | None = None,
) -> str:
"""File a vulnerability report — one report per fully-verified finding.
@@ -364,13 +342,13 @@ async def create_vulnerability_report(
- Avoid hedging language; be precise and non-vague.
**White-box requirement**: when source is available, you MUST
populate ``code_locations`` with nested XML including
``fix_before`` / ``fix_after`` for proposed fixes. The fix_before
must be a verbatim copy of source at the specified line range — it's
used as a literal GitHub/GitLab PR suggestion block.
populate ``code_locations`` with one entry per affected line range.
The ``fix_before`` field must be a verbatim copy of the source at
the specified line range — it's used as a literal GitHub/GitLab
PR suggestion block.
**CVSS breakdown** is required as nested XML with all 8 metrics
(each a single uppercase letter):
**CVSS breakdown** is an object with all 8 metrics (each a single
uppercase letter):
- ``attack_vector``: ``N`` (Network), ``A`` (Adjacent), ``L``
(Local), ``P`` (Physical)
@@ -381,6 +359,19 @@ async def create_vulnerability_report(
- ``confidentiality`` / ``integrity`` / ``availability``: ``N`` /
``L`` / ``H``
Example::
{
"attack_vector": "N",
"attack_complexity": "L",
"privileges_required": "N",
"user_interaction": "N",
"scope": "U",
"confidentiality": "H",
"integrity": "H",
"availability": "H"
}
**CVE / CWE rules**: pass the bare ID only (``CVE-2024-1234``,
``CWE-89``) — no name, no parenthetical. Be 100% certain; if
unsure, omit. Always prefer the most specific child CWE over a
@@ -397,14 +388,16 @@ async def create_vulnerability_report(
poc_description: Step-by-step reproduction.
poc_script_code: Working PoC (Python preferred).
remediation_steps: Specific, actionable fix.
cvss_breakdown: 8-metric XML block per the format above.
cvss_breakdown: 8-metric object per the format above.
endpoint: API path / Git path (e.g. ``/api/login``).
method: HTTP method when relevant.
cve: ``CVE-YYYY-NNNNN`` if certain, else omit.
cwe: ``CWE-NNN`` (most specific child) if certain, else omit.
code_locations: Required for white-box findings; nested XML
list with ``file``, ``start_line``, ``end_line``,
``snippet``, ``fix_before``, ``fix_after``.
code_locations: Required for white-box findings. List of
objects, each with ``file`` (relative path), ``start_line``,
``end_line``, optional ``snippet``, ``label``,
``fix_before`` (verbatim source), ``fix_after`` (suggested
replacement).
"""
del ctx
result = await asyncio.to_thread(