Compare commits

...
Author SHA1 Message Date
alex s f28b022a07 Update xss.md 2026-10-22 01:09:23 -04:00
theo-at-strix 4205fb4ba5 feat(skills): expand XSS coverage for HTML injection and normalization 2026-10-22 03:13:20 +03:00
alex s a3c31236bd build(deps): enforce a 5-day release cooldown for uv and npm resolution (#1556) 2026-10-21 11:30:08 -04:00
alex s fde37a149f fix(reporting): keep HTTP exchange evidence when the proxy is slow to verify ids (#1551)
Verifying http_exchange_ids fetched every request with its full raw
request and response bodies, serially, and timed out on busy proxy
projects; the ids were then dropped and the finding lost its evidence.

Check existence with an id-only GraphQL query, retry once, and when the
proxy still does not answer keep the ids with a warning instead of
discarding them.
2026-10-20 23:36:38 -04:00
ian-at-strix 3a3855fc4f feat(agents): let callers choose the report tool (#1550)
* feat(agents): let callers remove base tools from scan agents

* feat(agents): let callers choose the report tool instead of removing tools

* refactor(agents): name the report tool via the prompt template, not string replacement
2026-10-20 20:21:31 -04:00
Ian aa2036f229 fix(llm): keep request_log.install(); record tool call timings; both only with STRIX_RUN_RECORD_TIMINGS 2026-10-20 18:26:50 +03:00
Ian f4a7413963 feat(llm): run.json records every provider request's timing, provider and tokens in llm_requests 2026-10-20 18:26:50 +03:00
21 changed files with 332 additions and 57 deletions

No files matched your search

+2
View File
@@ -31,6 +31,8 @@ jobs:
assert not any(name.startswith("strix/interface/tui/cmd/") for name in names)
PY
- name: Install into a clean environment and run the CLI
env:
UV_EXCLUDE_NEWER: 5 days
run: |
uv venv --python 3.12 /tmp/strix-smoke
uv pip install --python /tmp/strix-smoke/bin/python dist/*.whl
+4
View File
@@ -141,6 +141,10 @@ affecting the agents that do the actual testing.
Telemetry toggle. Set to `0`, `false`, `no`, or `off` to disable telemetry (PostHog, Scarf, OTEL).
</ParamField>
<ParamField path="STRIX_RUN_RECORD_TIMINGS" default="false" type="boolean">
Record every LLM request's and tool call's timing in `run.json`, under `llm_requests` and `tool_calls`. Adds about 0.5 MB to a typical scan's `run.json`.
</ParamField>
<ParamField path="TRACELOOP_BASE_URL" type="string">
OTLP/Traceloop base URL for remote OpenTelemetry export. If unset, Strix keeps traces local only.
</ParamField>
+6
View File
@@ -73,6 +73,12 @@ dev = [
[tool.pytest.ini_options]
asyncio_mode = "auto"
[tool.uv]
# Supply-chain cooldown: resolve only distributions that have been on the index for
# at least 5 days, so a poisoned release (litellm 1.107.0, 2026-10-21) is never
# picked up by `uv lock` / `uv add` before it can be caught and yanked.
exclude-newer = "5 days"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
+9 -2
View File
@@ -724,6 +724,7 @@ def build_strix_agent(
instructions_override: str | None = None,
supports_images: bool = True,
finish_tool: Tool = finish_scan,
report_tool: Tool = create_vulnerability_report,
) -> SandboxAgent[Any]:
"""Build a SandboxAgent for either root or child use.
@@ -737,6 +738,8 @@ def build_strix_agent(
instructions_override: Use this verbatim as the system prompt instead
of rendering the built-in scan prompt.
finish_tool: The tool that ends the run, given to the root agent.
report_tool: The tool every agent files findings with, in place of
``create_vulnerability_report``.
"""
if instructions_override is not None:
instructions = instructions_override
@@ -750,16 +753,18 @@ def build_strix_agent(
interactive=interactive,
system_prompt_context=system_prompt_context,
supports_images=supports_images,
report_tool=report_tool.name,
)
agent_tools = [*_EXTRA_TOOLS, *(extra_tools or [])]
if interactive:
# Yielding to the user is only meaningful when one is attached.
agent_tools.append(wait_for_user)
base_tools = [report_tool if t is create_vulnerability_report else t for t in _BASE_TOOLS]
if is_root:
tools: list[Tool] = [*_BASE_TOOLS, *agent_tools, finish_tool]
tools: list[Tool] = [*base_tools, *agent_tools, finish_tool]
else:
tools = [*_BASE_TOOLS, *agent_tools, agent_finish]
tools = [*base_tools, *agent_tools, agent_finish]
_ensure_unique_tool_names(tools)
tools = [
_with_bounded_result(_with_strictness(_with_coerced_arguments(tool), strict_tool_schemas))
@@ -812,6 +817,7 @@ def make_child_factory(
strict_tool_schemas: bool = True,
system_prompt_context: dict[str, Any] | None = None,
supports_images: bool = True,
report_tool: Tool = create_vulnerability_report,
) -> Any:
"""Return the runner-owned builder used by ``spawn_child_agent``.
@@ -833,6 +839,7 @@ def make_child_factory(
strict_tool_schemas=strict_tool_schemas,
system_prompt_context=system_prompt_context,
supports_images=supports_images,
report_tool=report_tool,
)
return _factory
+2
View File
@@ -86,6 +86,7 @@ def render_system_prompt(
system_prompt_context: dict[str, Any] | None = None,
include_scope: bool = True,
supports_images: bool = True,
report_tool: str = "create_vulnerability_report",
) -> str:
"""Render the system prompt. Returns empty string on template failure.
@@ -133,6 +134,7 @@ def render_system_prompt(
is_root=is_root,
system_prompt_context=system_prompt_context or {},
include_scope=include_scope,
report_tool=report_tool,
**skill_content,
)
except Exception:
+6 -6
View File
@@ -119,7 +119,7 @@ WHITE-BOX TESTING (code provided):
- Local execution, unit/integration testing, patch verification, and HTTP requests against locally started in-scope services are normal authorized white-box validation
- If dynamically running the code proves impossible after exhaustive attempts, pivot to comprehensive static analysis.
- Try to infer how to run the code based on its structure and content.
- Derive the code fix as PART OF reporting, not as a separate later pass: create_vulnerability_report already requires the concrete patch inline (`code_locations` with verbatim `fix_before`/`fix_after` and `fix_pr_body`), so the reporting agent that analyzes the root cause is the one that produces the fix. Do NOT spawn a downstream agent afterwards to re-derive/re-apply the same patch.
- Derive the code fix as PART OF reporting, not as a separate later pass: {{ report_tool }} already requires the concrete patch inline (`code_locations` with verbatim `fix_before`/`fix_after` and `fix_pr_body`), so the reporting agent that analyzes the root cause is the one that produces the fix. Do NOT spawn a downstream agent afterwards to re-derive/re-apply the same patch.
- If you also apply and verify the patch in the repo (edit the file, re-test that the vulnerability is gone), do it in the same agent/turn while the analysis is fresh — right before or as part of filing the report — never as a second re-analysis pass.
COMBINED MODE (code + deployed target present):
@@ -205,9 +205,9 @@ VALIDATION REQUIREMENTS:
- COVERAGE: record every surface you assess with `record_coverage` (surface + risk area + outcome + evidence), including the ones that came back clean — a report that only lists findings cannot say what was reviewed and cleared. Use the `needs_follow_up` outcome for anything left in an `open_proof_gap` state, and carry the same items up in `agent_finish(open_items=[...])`. The ledger is shared and mutable: when you resolve a surface another agent left open — or find that a closed one is not — move that entry with `update_coverage` instead of recording a second one for the same surface. The root agent reconciles all of it via `list_coverage` before `finish_scan`.
- THREAT MODEL: before you start testing, call `get_threat_model` on the target you were pointed at — it is the scan's shared answer to who the attacker is, where the trust boundaries sit, and what counts as critical here. It is scoped to this scan and nothing carries over from an earlier run, so `found: false` means no agent on this run has derived one yet. Read it instead of re-deriving trust boundaries yourself; where your testing disproves it — a boundary it calls trusted turns out to be attacker-reachable, a role it did not know about, a host or endpoint it never listed — record that with `amend_threat_model` so the agents after you inherit the correction. Amending is not optional politeness: a model nobody corrects turns the first agent's guesses into everyone's assumptions.
- 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. 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.
- A vulnerability is ONLY considered reported when a reporting agent uses {{ report_tool }} (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 {{ report_tool }} (`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 {{ report_tool }} 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.
- HTTP EVIDENCE: a finding you validated through the proxy is not fully filed until `http_exchange_ids` carries the proxy request ids of the exchanges that prove it — the request that triggers the vulnerability plus the baseline/control request it differs from (an unauthenticated success next to the authenticated one, the payload response next to the benign one). Copy the ids exactly as `list_requests`/`view_request` show them, never invent or guess one, and never omit the field to bypass validation. Leave it out only when there is no captured HTTP exchange at all (static-only code findings, dependency CVEs). If you filed before the proving exchanges existed, attach them afterwards with update_vulnerability_report. Without the ids, the finding ships as prose nobody can replay.
- 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.
- WITHDRAWING A FINDING: when re-testing proves a finding you filed is not a vulnerability at all — the exploit only worked because of your own test setup (a mixed-up session, a self-inflicted state change, a misread response), the behaviour is intended, or the control is in fact enforced — delete it with delete_vulnerability_report (report id + delete_reason). Never leave a disproved finding on file as a zero-CVSS, "retracted", or "false positive" report, and never rewrite its text into a note saying it was wrong: a report that exists is read as a finding. Delete is for a finding that does not exist; a finding that is real but weaker than filed is revised, not deleted.
@@ -227,7 +227,7 @@ Every one of these tools writes to state the rest of the scan reads. Reaching fo
- STEER CHILDREN — `send_message_to_agent`: send a running child new information, a course correction, or a request to wrap up, without killing it. Use it to answer a child's question or narrow its scope mid-run.
- BLOCK ON CHILDREN — `wait_for_agents`: block until named children report back when your next move genuinely depends on their results. If you can keep making progress in parallel, keep working instead of waiting.
- CANCEL CHILDREN — `stop_agent`: gracefully cancel a child whose work is redundant, misdirected, or no longer needed. Prefer `send_message_to_agent` to redirect a child that is merely off-track; reserve `stop_agent` for work that should not continue at all.
- FINISH — subagents call `agent_finish` (with `open_items=[...]` for anything left unresolved); the root agent calls `finish_scan` exactly once, only after every child is wrapped up and coverage is reconciled. `agent_finish`/`finish_scan` are handoffs, not reporting channels — a vulnerability is reported only via `create_vulnerability_report`/`create_dependency_report`.
- FINISH — subagents call `agent_finish` (with `open_items=[...]` for anything left unresolved); the root agent calls `finish_scan` exactly once, only after every child is wrapped up and coverage is reconciled. `agent_finish`/`finish_scan` are handoffs, not reporting channels — a vulnerability is reported only via `{{ report_tool }}`/`create_dependency_report`.
</execution_guidelines>
<vulnerability_focus>
@@ -380,7 +380,7 @@ CRITICAL RULES:
- **REALISTIC OUTCOMES** - Some tests find nothing, some validations fail
- **ONE AGENT = ONE TASK** - Don't let agents do multiple unrelated jobs
- **SPAWN REACTIVELY** - Create new agents based on what you discover
- **ONLY REPORTING AGENTS** can use create_vulnerability_report tool
- **ONLY REPORTING AGENTS** can use {{ report_tool }} tool
- **AGENT SPECIALIZATION MANDATORY** - Each agent must be highly specialized; prefer 1–3 skills, up to 5 for complex contexts
- **NO GENERIC AGENTS** - Avoid creating broad, multi-purpose agents that dilute focus
+4
View File
@@ -705,7 +705,11 @@ def configure_sdk_model_defaults(settings: Settings) -> None:
"""Apply Strix config to SDK-native defaults."""
llm = settings.llm
set_tracing_disabled(True)
from strix.report.state import record_llm_request
request_log.install()
if settings.telemetry.run_record_timings:
request_log.register_sink(record_llm_request)
if codex.subscription_model(llm.model):
return
_configure_litellm_compatibility()
+1
View File
@@ -163,6 +163,7 @@ class TelemetrySettings(BaseSettings):
model_config = _BASE_CONFIG
enabled: bool = Field(default=True, alias="STRIX_TELEMETRY")
run_record_timings: bool = Field(default=False, alias="STRIX_RUN_RECORD_TIMINGS")
WebSearchProvider = Literal["auto", "perplexity", "exa"]
+38 -1
View File
@@ -4,18 +4,21 @@ from __future__ import annotations
import logging
import math
import time
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any
from agents.lifecycle import RunHooks
from strix.core.agents import BudgetPolicy, coordinator_from_context
from strix.report.state import get_global_report_state
from strix.report.state import get_global_report_state, record_tool_call
if TYPE_CHECKING:
from agents import RunContextWrapper
from agents.agent import Agent
from agents.items import ModelResponse, TResponseInputItem
from agents.tool import Tool
logger = logging.getLogger(__name__)
@@ -163,6 +166,7 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
max_turns: int | None = None,
interactive: bool = False,
budget_policy: BudgetPolicy = "stop",
record_tool_timings: bool = False,
) -> None:
_validate_budget(max_budget_usd)
if max_turns is not None and max_turns <= 0:
@@ -175,6 +179,8 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
self._max_turns = max_turns
self._interactive = interactive
self._budget_policy: BudgetPolicy = budget_policy
self._record_tool_timings = record_tool_timings
self._tool_starts: dict[str, tuple[datetime, float]] = {}
@property
def max_budget_usd(self) -> float | None:
@@ -315,6 +321,37 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
)
input_items.append({"role": "user", "content": content})
async def on_tool_start(
self,
context: RunContextWrapper[dict[str, Any]],
agent: Agent[dict[str, Any]], # noqa: ARG002
tool: Tool, # noqa: ARG002
) -> None:
call_id = getattr(context, "tool_call_id", None)
if self._record_tool_timings and isinstance(call_id, str):
self._tool_starts[call_id] = (datetime.now(UTC), time.monotonic())
async def on_tool_end(
self,
context: RunContextWrapper[dict[str, Any]],
agent: Agent[dict[str, Any]], # noqa: ARG002
tool: Tool,
result: object, # noqa: ARG002
) -> None:
call_id = getattr(context, "tool_call_id", None)
start = self._tool_starts.pop(call_id, None) if isinstance(call_id, str) else None
if start is None:
return
started_at, started = start
ctx = context.context if isinstance(context.context, dict) else {}
agent_id = ctx.get("agent_id")
record_tool_call(
started_at=started_at,
agent_id=agent_id if isinstance(agent_id, str) else None,
tool=tool.name,
duration_ms=round((time.monotonic() - started) * 1000),
)
async def on_llm_end(
self,
context: RunContextWrapper[dict[str, Any]],
+10
View File
@@ -54,6 +54,7 @@ from strix.tools.output_store import (
WORKSPACE_SPILL_DIR,
configure_spill_writer,
)
from strix.tools.reporting.tool import create_vulnerability_report
if TYPE_CHECKING:
@@ -172,6 +173,7 @@ def _compose_root_instructions_override(
interactive: bool,
system_prompt_context: dict[str, Any],
supports_images: bool,
report_tool: str,
) -> str | None:
if root_instructions_override is None:
return None
@@ -186,6 +188,7 @@ def _compose_root_instructions_override(
system_prompt_context=system_prompt_context,
include_scope=False,
supports_images=supports_images,
report_tool=report_tool,
)
return (
f"{base_instructions}\n\n"
@@ -218,6 +221,7 @@ async def run_strix_scan(
mcp_connection_requests: list[McpConnectionRequest] | None = None,
mcp_status_sink: McpStatusSink | None = None,
root_finish_tool: Tool = finish_scan,
report_tool: Tool = create_vulnerability_report,
) -> RunResultBase | None:
"""Run or resume one Strix scan against a sandbox.
@@ -241,6 +245,8 @@ async def run_strix_scan(
way the engine does the connecting, so the caller passes inert configs plus
metadata and never live sessions.
``root_finish_tool`` is the tool the root agent ends the run with.
``report_tool`` is the tool every agent files findings with, in place of
``create_vulnerability_report``.
"""
def report(phase: str) -> None:
@@ -419,6 +425,7 @@ async def run_strix_scan(
max_turns=max_turns,
interactive=interactive,
budget_policy=budget_policy,
record_tool_timings=settings.telemetry.run_record_timings,
)
coordinator.set_budget_limit_setter(hooks.set_max_budget_usd)
if interactive and budget_policy != "pause":
@@ -494,6 +501,7 @@ async def run_strix_scan(
interactive=interactive,
system_prompt_context=root_context,
supports_images=supports_images,
report_tool=report_tool.name,
)
root_agent = build_strix_agent(
@@ -510,6 +518,7 @@ async def run_strix_scan(
instructions_override=root_instructions,
supports_images=supports_images,
finish_tool=root_finish_tool,
report_tool=report_tool,
)
if not is_resume:
@@ -530,6 +539,7 @@ async def run_strix_scan(
strict_tool_schemas=strict_tool_schemas,
system_prompt_context=scope_context,
supports_images=supports_images,
report_tool=report_tool,
)
async def spawn_child_agent(**kwargs: Any) -> dict[str, Any]:
+2
View File
@@ -0,0 +1,2 @@
# Supply-chain cooldown: only install versions published at least 5 days ago (npm >= 11.10).
min-release-age=5
+43
View File
@@ -28,6 +28,8 @@ from strix.telemetry import posthog, scarf
if TYPE_CHECKING:
from agents.usage import Usage
from strix.llm.request_log import LlmRequestEvent
logger = logging.getLogger(__name__)
@@ -1019,6 +1021,47 @@ class StreamedOpenRouterCosts:
streamed_openrouter_costs = StreamedOpenRouterCosts()
def record_llm_request(event: "LlmRequestEvent") -> None:
"""Request-log sink: each provider attempt's timing, kept in run.json's ``llm_requests``."""
report_state = get_global_report_state()
if report_state is None:
return
details = event.details or {}
report_state.run_record.setdefault("llm_requests", []).append(
{
"started_at": event.started_at.isoformat(),
"agent_id": event.agent_id,
"model": event.model,
"provider": details.get("upstream_provider") or event.provider,
"outcome": event.outcome,
"status": event.status_code,
"duration_ms": event.duration_ms,
"ttft_ms": event.time_to_first_token_ms,
"input_tokens": event.input_tokens,
"cached_tokens": event.cached_input_tokens,
"output_tokens": event.output_tokens,
"retry_attempt": event.retry_attempt,
}
)
def record_tool_call(
*, started_at: datetime, agent_id: str | None, tool: str, duration_ms: int
) -> None:
"""Each tool call's timing, kept in run.json's ``tool_calls``."""
report_state = get_global_report_state()
if report_state is None:
return
report_state.run_record.setdefault("tool_calls", []).append(
{
"started_at": started_at.isoformat(),
"agent_id": agent_id,
"tool": tool,
"duration_ms": duration_ms,
}
)
def record_openrouter_provider(provider: Any, usage: Any) -> None:
"""Tally which upstream provider served a stream, from its final usage chunk.
+2 -3
View File
@@ -20,7 +20,7 @@ state, and "I moved on" is not one of them.
**1. `confirmed`** — you have a working PoC or, in white-box, a complete
source → control → sink → impact trace plus evidence the path is
reachable. File it with `create_vulnerability_report`.
reachable. File it as a vulnerability report.
**2. `ruled_out`** — you can name the **specific control** that makes the
code safe, at a specific location, and you have checked that the control
@@ -144,8 +144,7 @@ endpoint is broken/unreachable for unrelated reasons".
## Before You File a Report
Run this pass on every finding before calling
`create_vulnerability_report`:
Run this pass on every finding before filing it:
1. **Argue the other side.** Spend real effort building the strongest
case that this is *not* exploitable, or not as severe as you think.
+30 -6
View File
@@ -1,6 +1,6 @@
---
name: xss
description: XSS testing covering reflected, stored, and DOM-based vectors with CSP bypass techniques
description: XSS and HTML injection testing covering reflected, stored, DOM-based, and normalization-dependent vectors
---
# XSS
@@ -11,6 +11,7 @@ Cross-site scripting persists because context, parser, and framework edges are c
**Types**
- Reflected, stored, and DOM-based XSS across web/mobile/desktop shells
- Scriptless HTML injection and second-order render paths
**Contexts**
- HTML, attribute, URL, JS, CSS, SVG/MathML, Markdown, PDF
@@ -20,6 +21,7 @@ Cross-site scripting persists because context, parser, and framework edges are c
**Defenses to Bypass**
- CSP/Trusted Types, DOMPurify, framework auto-escaping
- Decode, normalization, sanitization, and encoding order
## Injection Points
@@ -57,6 +59,24 @@ Cross-site scripting persists because context, parser, and framework edges are c
Distinguish HTML-string input from live DOM objects sanitized with `IN_PLACE`. Resolve the deployed DOMPurify build and check relevant upstream advisories; a controlled observable `nodeName` is one live-node bypass mechanism. Trace same-origin foreign nodes or `adoptNode()` into this mode before selecting the bypass ([example advisory](https://github.com/cure53/DOMPurify/security/advisories/GHSA-x4vx-rjvf-j5p4)).
### Scriptless HTML Injection
Do not require JavaScript execution before you report HTML injection. Prove that the browser creates attacker-controlled markup and a concrete security effect.
Test base or form target changes, external requests, deceptive content, DOM clobbering through `id` or `name`, and dangling markup. Report XSS only when you prove a script-capable execution path. Do not dismiss HTML injection only because CSP blocks scripts.
### Unicode Normalization
Test normalization only when the target performs it. NFKC can fold fullwidth delimiters such as `<` and `>` into HTML syntax.
Map validation, normalization, sanitization, encoding, and parsing in order. Start with the harmless canary `<h1 id="strix-nfkc">marker</h1>`.
Report a bypass only when normalization creates dangerous syntax that reaches an unprotected sink. Load `semantic_confusion` when components interpret the value differently.
### Second-Order Rendering
Follow stored values into previews, administrative pages, email, exports, and hydration. Record which stage transforms and parses the value.
### DOM XSS
**Sources**
@@ -166,7 +186,7 @@ Keep a compact set tuned per context:
## Testing Methodology
1. **Identify sources** - URL/query/hash/referrer, postMessage, storage, WebSocket, server JSON
2. **Trace to sinks** - Map data flow from source to sink
2. **Trace transformations and sinks** - Record decoding, normalization, sanitization, reparsing, and the final sink
3. **Classify context** - HTML node, attribute, URL, script block, event handler, JS eval-like, CSS, SVG
4. **Assess defenses** - Output encoding, sanitizer, CSP, Trusted Types, DOMPurify config
5. **Craft payloads** - Minimal payloads per context with encoding/whitespace/casing variants
@@ -175,9 +195,11 @@ Keep a compact set tuned per context:
## Validation
1. Provide minimal payload and context (sink type) with before/after DOM or network evidence
2. Demonstrate cross-browser execution where relevant or explain parser-specific behavior
3. Show bypass of stated defenses (sanitizer settings, CSP/Trusted Types) with proof
4. Quantify impact beyond alert: data accessed, action performed, persistence achieved
2. Record the original and transformed value when decoding or normalization affects the result
3. Demonstrate cross-browser execution where relevant or explain parser-specific behavior
4. Show bypass of stated defenses (sanitizer settings, CSP/Trusted Types) with proof
5. Separate HTML injection from XSS and state the proven capability
6. Quantify impact beyond alert: data accessed, action performed, persistence achieved
## False Positives
@@ -185,6 +207,8 @@ Keep a compact set tuned per context:
- CSP with nonces/hashes and no inline/event handlers
- Trusted Types enforced on sinks; DOMPurify in strict mode with URI allowlists
- Scriptable contexts disabled (no HTML pass-through, safe URL schemes enforced)
- The target does not perform the claimed normalization or decode
- A normalized value is safely encoded before the final sink
## Impact
@@ -207,4 +231,4 @@ Keep a compact set tuned per context:
## Summary
Context + sink decide execution. Encode for the exact context, verify at runtime with CSP/Trusted Types, and validate every alternative render path. Small payloads with strong evidence beat payload catalogs.
Context, transformation order, and sink decide whether input becomes text, markup, or executable code. Prove HTML injection separately from XSS, and validate every render path. Encode for the exact context and verify at runtime with CSP/Trusted Types. Small payloads with strong evidence beat payload catalogs.
+25 -8
View File
@@ -69,6 +69,15 @@ async def _call[T](client: Client, fn: Callable[[Client], Awaitable[T]]) -> T:
return await fn(client)
_REQUEST_EXISTS_QUERY = "query RequestExists($id: ID!) { request(id: $id) { id } }"
_REQUEST_EXISTS_ATTEMPTS = 2
async def _request_exists(client: Client, request_id: str) -> bool:
data = await client.graphql.query(_REQUEST_EXISTS_QUERY, {"id": request_id})
return bool(data) and data.get("request") is not None
async def existing_request_ids(
ctx: RunContextWrapper,
request_ids: list[str],
@@ -81,16 +90,24 @@ async def existing_request_ids(
if client is None:
raise RuntimeError("Caido client is not available")
# Request IDs are not an HTTPQL field. Resolve each ID through the same
# project-bound lookup as view_request rather than constructing a filter.
# Request IDs are not an HTTPQL field, so each one is resolved by id. Only
# the id is selected: pulling the raw request and response bodies just to
# learn whether a row exists is what made this time out on busy projects.
existing: set[str] = set()
for request_id in request_ids:
result = await _call(
client,
functools.partial(caido_api.get_request_with_client, request_id=request_id),
)
if result is not None:
existing.add(str(result.request.id))
for attempt in range(1, _REQUEST_EXISTS_ATTEMPTS + 1):
try:
exists = await _call(
client,
functools.partial(_request_exists, request_id=request_id),
)
except Exception:
if attempt == _REQUEST_EXISTS_ATTEMPTS:
raise
continue
if exists:
existing.add(request_id)
break
return existing
+7 -7
View File
@@ -222,9 +222,9 @@ def _normalize_http_exchange_ids(raw: Any) -> tuple[list[str] | None, list[str]]
return normalized, errors
_HTTP_EXCHANGE_DROPPED_WARNING = (
"http_exchange_ids were not stored: the proxy project could not be reached to verify "
"them. Attach them with update_vulnerability_report when the proxy responds again."
_HTTP_EXCHANGE_UNVERIFIED_WARNING = (
"http_exchange_ids were stored without verification because the proxy project did "
"not respond in time. Ids that do not exist in the project will not render as evidence."
)
@@ -235,9 +235,9 @@ async def _verify_http_exchange_ids(
"""Verify proxy exchange IDs against the current Caido project.
IDs the project does not know are rejected. When the proxy itself cannot be
queried the IDs are dropped and a warning is returned instead, so a proxy
outage never blocks a finding and unverified IDs are never recorded as
evidence.
queried the IDs are kept and a warning is returned instead: a slow or
unreachable proxy must never strip evidence from a finding, and consumers
resolve each id against the recorded traffic before rendering it.
"""
request_ids, errors = _normalize_http_exchange_ids(raw)
if request_ids is None or errors or not request_ids:
@@ -250,7 +250,7 @@ async def _verify_http_exchange_ids(
"Could not verify HTTP exchange IDs against the current Caido project",
exc_info=True,
)
return None, [], _HTTP_EXCHANGE_DROPPED_WARNING
return request_ids, [], _HTTP_EXCHANGE_UNVERIFIED_WARNING
missing_ids = [request_id for request_id in request_ids if request_id not in existing_ids]
if missing_ids:
+11
View File
@@ -93,6 +93,17 @@ def test_per_call_extra_tools_reject_duplicate_registered_names() -> None:
factory.build_strix_agent(is_root=True, extra_tools=[_tool("same_name")])
def test_report_tool_replaces_create_vulnerability_report() -> None:
for is_root in (True, False):
agent = factory.build_strix_agent(is_root=is_root, report_tool=_tool("file_pr_finding"))
names = [t.name for t in agent.tools]
assert "file_pr_finding" in names
assert "create_vulnerability_report" not in names
assert isinstance(agent.instructions, str)
assert "file_pr_finding" in agent.instructions
assert "create_vulnerability_report" not in agent.instructions
def test_instructions_override_is_used_verbatim() -> None:
custom = "You are a scan agent. Follow the provided scope."
+73
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
import json
import uuid
from datetime import UTC, datetime, timedelta
from types import SimpleNamespace
from typing import TYPE_CHECKING, Any
from unittest.mock import MagicMock, call, patch
@@ -18,11 +19,13 @@ from strix.config.models import (
_configure_litellm_compatibility,
_install_openrouter_stream_cost_capture,
)
from strix.core.hooks import ReportUsageHooks
from strix.llm import request_log
from strix.report.state import (
ReportState,
litellm_cost_callback,
openrouter_stream_cost,
record_llm_request,
set_global_report_state,
streamed_openrouter_costs,
)
@@ -455,3 +458,73 @@ def test_openrouter_records_upstream_of_pre_stream_error() -> None:
request_log._http_reply.reset(token)
assert reply.upstream_provider == "InferenceNet"
assert reply.upstream_error_type is None
def test_request_timings_land_in_run_record() -> None:
started = datetime(2026, 10, 19, 12, 0, tzinfo=UTC)
event = request_log.LlmRequestEvent(
call_id="call-1",
route="litellm",
provider="openrouter",
model="openrouter/deepseek/deepseek-v4.1-flash",
api_host="openrouter.ai",
streaming=True,
outcome="success",
status_code=200,
provider_request_id=None,
response_id="gen-1",
error_type=None,
error_message=None,
started_at=started,
finished_at=started + timedelta(milliseconds=4200),
duration_ms=4200,
input_tokens=90_000,
output_tokens=600,
cached_input_tokens=85_000,
agent_id="agent-1",
time_to_first_token_ms=900,
details={"upstream_provider": "Together"},
)
state = ReportState.__new__(ReportState)
state.run_record = {}
try:
set_global_report_state(state)
record_llm_request(event)
finally:
set_global_report_state(None)
assert state.run_record["llm_requests"] == [
{
"started_at": "2026-10-19T12:00:00+00:00",
"agent_id": "agent-1",
"model": "openrouter/deepseek/deepseek-v4.1-flash",
"provider": "Together",
"outcome": "success",
"status": 200,
"duration_ms": 4200,
"ttft_ms": 900,
"input_tokens": 90_000,
"cached_tokens": 85_000,
"output_tokens": 600,
"retry_attempt": 0,
}
]
async def test_tool_call_timings_land_in_run_record() -> None:
hooks = ReportUsageHooks(model="m", record_tool_timings=True)
context = SimpleNamespace(tool_call_id="call-1", context={"agent_id": "agent-1"})
tool = SimpleNamespace(name="terminal_execute")
state = ReportState.__new__(ReportState)
state.run_record = {}
try:
set_global_report_state(state)
await hooks.on_tool_start(context, MagicMock(), tool) # type: ignore[arg-type]
await hooks.on_tool_end(context, MagicMock(), tool, "ok") # type: ignore[arg-type]
finally:
set_global_report_state(None)
[entry] = state.run_record["tool_calls"]
assert entry["agent_id"] == "agent-1"
assert entry["tool"] == "terminal_execute"
assert entry["duration_ms"] >= 0
+45 -15
View File
@@ -9,7 +9,6 @@ call at a time against the shared client.
from __future__ import annotations
import asyncio
from types import SimpleNamespace
from typing import TYPE_CHECKING, Any, cast
import pytest
@@ -27,6 +26,7 @@ class _FakeClient:
def __init__(self, name: str) -> None:
self.name = name
self.closed = False
self.graphql: Any = None
async def aclose(self) -> None:
self.closed = True
@@ -231,20 +231,23 @@ async def test_ctx_client_degrades_when_bootstrap_failed() -> None:
assert await tools._ctx_client(cast("Any", _Ctx({"caido_client": handle}))) is None
async def test_existing_request_ids_queries_current_project(
monkeypatch: pytest.MonkeyPatch,
) -> None:
class _FakeGraphQL:
def __init__(self, answers: dict[str, list[Any]]) -> None:
self.answers = answers
self.queries: list[tuple[str, dict[str, Any]]] = []
async def query(self, document: str, variables: dict[str, Any]) -> dict[str, Any]:
self.queries.append((document, dict(variables)))
answer = self.answers[variables["id"]].pop(0)
if isinstance(answer, Exception):
raise answer
return {"request": answer}
async def test_existing_request_ids_select_only_the_id() -> None:
client = _FakeClient("host")
looked_up: list[str] = []
async def get_request_with_client(passed_client: Any, request_id: str) -> Any:
assert passed_client is client
looked_up.append(request_id)
if request_id == "1042":
return SimpleNamespace(request=SimpleNamespace(id="1042"))
return None
monkeypatch.setattr(caido_api, "get_request_with_client", get_request_with_client)
graphql = _FakeGraphQL({"1042": [{"id": "1042"}], "1088": [None]})
client.graphql = graphql
existing = await tools.existing_request_ids(
cast("Any", _Ctx({"caido_client": client})),
@@ -252,7 +255,34 @@ async def test_existing_request_ids_queries_current_project(
)
assert existing == {"1042"}
assert looked_up == ["1042", "1088"]
assert [variables["id"] for _, variables in graphql.queries] == ["1042", "1088"]
for document, _ in graphql.queries:
assert "raw" not in document
assert "{ id }" in document
async def test_existing_request_ids_retry_once_before_giving_up() -> None:
client = _FakeClient("host")
graphql = _FakeGraphQL(
{
"1042": [TimeoutError("deadline exceeded"), {"id": "1042"}],
"1088": [TimeoutError("deadline exceeded"), TimeoutError("deadline exceeded")],
}
)
client.graphql = graphql
with pytest.raises(TimeoutError):
await tools.existing_request_ids(
cast("Any", _Ctx({"caido_client": client})),
["1042", "1088"],
)
assert [variables["id"] for _, variables in graphql.queries] == [
"1042",
"1042",
"1088",
"1088",
]
def test_err_logs_caido_errors_as_warnings(caplog: pytest.LogCaptureFixture) -> None:
+8 -9
View File
@@ -1640,7 +1640,7 @@ async def test_http_exchange_ids_must_exist_in_current_proxy_project(
assert warning is None
async def test_http_exchange_ids_are_dropped_when_proxy_cannot_be_queried(
async def test_http_exchange_ids_are_kept_when_proxy_cannot_be_queried(
monkeypatch: pytest.MonkeyPatch,
) -> None:
async def existing_request_ids(
@@ -1656,11 +1656,10 @@ async def test_http_exchange_ids_are_dropped_when_proxy_cannot_be_queried(
["1042", "1042", "1088"],
)
assert request_ids is None
assert request_ids == ["1042", "1088"]
assert errors == []
assert warning is not None
assert "not stored" in warning
assert "update_vulnerability_report" in warning
assert "without verification" in warning
async def test_create_reports_persistence_failure_as_tool_error(
@@ -1679,7 +1678,7 @@ async def test_create_reports_persistence_failure_as_tool_error(
assert report_state.vulnerability_reports == []
async def test_evidence_only_update_reports_proxy_outage_as_retryable(
async def test_evidence_only_update_keeps_ids_when_proxy_cannot_be_queried(
report_state: ReportState,
monkeypatch: pytest.MonkeyPatch,
) -> None:
@@ -1712,11 +1711,11 @@ async def test_evidence_only_update_reports_proxy_outage_as_retryable(
)
result = json.loads(raw)
assert result["success"] is False
assert "No fields to update" not in result["error"]
assert "update_vulnerability_report" in result["error"]
assert result["success"] is True
assert "without verification" in result["warning"]
assert result["report_id"] == "vuln-0009"
assert report_state.vulnerability_reports[0] == original
assert report_state.vulnerability_reports[0]["http_exchange_ids"] == ["204"]
assert report_state.vulnerability_reports[0] != original
def test_update_reports_persistence_failure_as_tool_error(report_state: ReportState) -> None:
Generated
+4
View File
@@ -9,6 +9,10 @@ resolution-markers = [
"python_full_version < '3.14' and sys_platform != 'win32'",
]
[options]
exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for backwards compatibility when using relative exclude-newer values.
exclude-newer-span = "P5D"
[[package]]
name = "aiohappyeyeballs"
version = "2.6.2"