mortal/strix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Proxy tool sweep: drop send_request, fix Caido SDK gotchas

Browse Source 
send_request was a thin wrapper over the Caido Replay API that the model
could replicate with a one-liner `curl` via exec_command. The sandbox's
HTTP_PROXY env captures all such traffic for free, so the tool was
adding bugs (duplicate dispatch, dropped responses) without adding
capability. Removed across factory, tools module, sandbox-importable
caido_api helper, TUI renderer, prompt template, skill doc, and public
docs. repeat_request stays — it operates on captured request IDs with
structured modifications, which curl can't replicate cleanly.

Three caido-sdk-client workarounds that were hitting us through both
send_request and repeat_request:

- replay_send_raw used to pass CreateReplaySessionFromRaw to
  sessions.create(), which seeds a stored entry server-side, then
  called send() — producing two history rows per call. Empty-create +
  send produces one dispatched request.
- The same helper read result.entry.response_raw, an attribute that
  doesn't exist on ReplayEntry, so response bytes were silently
  dropped. Fixed to walk result.entry.response.raw with proper None
  guards.
- get_request_with_client passed include_request_raw / include_response_raw
  based on the requested part, but the SDK's generated pydantic models
  declare raw as required even though the GraphQL fragment makes it
  conditional via @include. Passing False crashed view_request with a
  pydantic validation error. Always request both raw bodies; the caller
  picks which to surface.

Also wrapped replay.send() in asyncio.wait_for(30s) so a stalled Caido
dispatch (notably loopback targets that don't route cleanly through the
sandbox proxy) fails fast with a model-readable error instead of
hanging the agent until the function_tool 120s budget expires.

Finally, list_requests now omits the roundtrip_ms field when Caido
reports 0 — proxy-captured unscoped traffic consistently reports 0
while scoped/replay traffic carries real measurements, so the absence
of the field is now informative ("Caido didn't measure this") rather
than misleading ("this request took 0ms").

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
0xallam
2026-05-24 19:16:06 -07:00
parent 940319f28a
commit 5921fec16c
7 changed files with 85 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/tools/proxy.mdx
+5 -1
View File
@@ -66,9 +66,13 @@ asyncio.run(main())
| `list_requests()` | Query captured traffic with HTTPQL filters |
| `view_request()` | Get full request/response details |
| `repeat_request()` | Replay a request with modifications |
| `send_request()` | Send a new HTTP request |
| `scope_rules()` | Manage proxy scope (allowlist/denylist) |
For one-off arbitrary requests, use shell tooling like `curl` — the
sandbox's `HTTP_PROXY` env routes the traffic through Caido
automatically, so it lands in `list_requests` and can be replayed via
`repeat_request`.
### Example: Automated IDOR Testing
```python
strix/agents/factory.py
-2
View File
@@ -48,7 +48,6 @@ from strix.tools.proxy.tools import (
list_requests,
repeat_request,
scope_rules,
send_request,
view_request,
)
from strix.tools.reporting.tool import create_vulnerability_report
@@ -268,7 +267,6 @@ _BASE_TOOLS: tuple[Tool, ...] = (
# Caido HTTP/HTTPS proxy
list_requests,
view_request,
send_request,
repeat_request,
scope_rules,
# Multi-agent graph tools (the coordinator is in ctx.context)
strix/agents/prompts/system_prompt.jinja
+1 -1
View File
@@ -171,7 +171,7 @@ EFFICIENCY TACTICS:
`python3 -c` or a here-document is acceptable.
- For Caido proxy automation inside Python, explicitly import from
`caido_api`:
`from caido_api import list_requests, view_request, send_request, repeat_request, scope_rules`
`from caido_api import list_requests, view_request, repeat_request, scope_rules`
- Prefer established fuzzers/scanners where applicable: ffuf, sqlmap, zaproxy, nuclei, wapiti, arjun, httpx, katana, semgrep, bandit, trufflehog, nmap. Use scripts mainly to coordinate or validate around them, not to replace them without reason
- For trial-heavy vectors (SQLi, XSS, XXE, SSRF, RCE, auth/JWT, deserialization), DO NOT iterate payloads manually in the browser. Always spray payloads via Python scripts through `exec_command` or terminal tools.
- When using established fuzzers/scanners, use the proxy for inspection where helpful
strix/interface/tui/renderers/proxy_renderer.py
-74
View File
@@ -195,80 +195,6 @@ class ViewRequestRenderer(BaseToolRenderer):
return Static(text, classes=css_classes)
@register_tool_renderer
class SendRequestRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "send_request"
css_classes: ClassVar[list[str]] = ["tool-call", "proxy-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static: # noqa: PLR0912, PLR0915
args = tool_data.get("args", {})
result = tool_data.get("result")
status = tool_data.get("status", "running")
method = args.get("method", "GET")
url = args.get("url", "")
req_headers = args.get("headers")
req_body = args.get("body", "")
text = Text()
text.append(PROXY_ICON, style="dim")
text.append(" sending request", style="#06b6d4")
text.append("\n")
text.append(" >> ", style="#3b82f6")
text.append(method, style="#a78bfa")
text.append(f" {_truncate(url, 180)}", style="dim")
if req_headers and isinstance(req_headers, dict):
for k, v in list(req_headers.items())[:5]:
text.append("\n")
text.append(" >> ", style="#3b82f6")
text.append(f"{k}: ", style="dim")
text.append(_sanitize(str(v), 150), style="dim")
if req_body and isinstance(req_body, str):
text.append("\n")
text.append(" >> ", style="#3b82f6")
body_lines = req_body.split("\n")[:4]
for i, line in enumerate(body_lines):
if i > 0:
text.append("\n")
text.append(" ", style="dim")
text.append(_truncate(line, MAX_LINE_LENGTH), style="dim")
if len(req_body.split("\n")) > 4:
text.append(" ...", style="dim italic")
if status == "completed" and isinstance(result, dict):
if "error" in result:
text.append(f"\n error: {_sanitize(str(result['error']), 150)}", style="#ef4444")
else:
code = result.get("status_code")
time_ms = result.get("response_time_ms")
text.append("\n")
text.append(" << ", style="#22c55e")
if code:
text.append(f"{code}", style=_status_style(code))
if time_ms:
text.append(f" ({time_ms}ms)", style="dim")
body = result.get("body", "")
if body and isinstance(body, str):
lines = body.split("\n")[:6]
for line in lines:
text.append("\n")
text.append(" << ", style="#22c55e")
text.append(_truncate(line, MAX_LINE_LENGTH - 5), style="dim")
if len(body.split("\n")) > 6:
text.append("\n")
text.append(" ...", style="dim italic")
css_classes = cls.get_css_classes(status)
return Static(text, classes=css_classes)
@register_tool_renderer
class RepeatRequestRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "repeat_request"
strix/skills/tooling/python.md
+6 -2
View File
@@ -21,7 +21,6 @@ from caido_api import (
list_requests,
repeat_request,
scope_rules,
send_request,
view_request,
)
```
@@ -59,10 +58,15 @@ Available helpers:
- `list_requests(httpql_filter=, first=50, after=, sort_by=, sort_order=, scope_id=)` returns a cursor-paginated Caido SDK `Connection`.
- `view_request(request_id, part="request")` returns a Caido SDK request object with raw request/response bytes.
- `send_request(method, url, headers=None, body="")` sends an arbitrary raw request through Caido Replay.
- `repeat_request(request_id, modifications={...})` replays a captured request after modifying `url`, `params`, `headers`, `body`, or `cookies`.
- `scope_rules(action, allowlist=, denylist=, scope_id=, scope_name=)` manages Caido scopes.
For one-off arbitrary requests (e.g. probing a fresh endpoint, hitting an
external API), use `exec_command` with `curl` / `httpx` / `requests`. The
sandbox's `HTTP_PROXY` env routes all such traffic through Caido
automatically, so it shows up in `list_requests` and you can use
`repeat_request` to replay-and-modify any of it.
## Workflow
For iterative exploit work, put code in a file:
strix/tools/proxy/caido_api.py
+51 -34
View File
@@ -13,8 +13,6 @@ from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
from caido_sdk_client import Client, TokenAuthOptions
from caido_sdk_client.types import (
ConnectionInfoInput,
CreateReplaySessionFromRaw,
CreateReplaySessionOptions,
CreateScopeOptions,
ReplaySendOptions,
RequestGetOptions,
@@ -130,10 +128,13 @@ async def get_request_with_client(
*,
part: RequestPart = "request",
) -> Any:
opts = RequestGetOptions(
request_raw=(part == "request"),
response_raw=(part == "response"),
)
# The Caido SDK's generated pydantic model marks Request.raw and
# Response.raw as required strings even though the GraphQL fragment
# makes them conditional via `@include(if: $includeRequestRaw)`.
# Passing False for either causes pydantic validation to fail with
# "Field required" on the missing raw field. Always request both —
# the caller picks which one to surface via ``part``.
opts = RequestGetOptions(request_raw=True, response_raw=True)
return await client.request.get(request_id, opts)
@@ -236,6 +237,15 @@ def apply_modifications(
}
# Hard wall-clock bound on a single replay dispatch. Caido's Replay
# API has no built-in send-side timeout, so a stalled connection
# (unroutable target, slow loopback, etc.) hangs the caller until the
# function_tool wrapper's 120s budget expires — by which point we've
# lost any useful error context. 30s is generous for legitimate HTTP
# and short enough that the model can decide to retry rather than wait.
_REPLAY_SEND_TIMEOUT_SECONDS = 30.0
async def replay_send_raw(
client: CaidoClient,
*,
@@ -243,17 +253,42 @@ async def replay_send_raw(
connection: ConnectionInfoInput,
) -> dict[str, Any]:
started = time.time()
session = await client.replay.sessions.create(
CreateReplaySessionOptions(
request_source=CreateReplaySessionFromRaw(raw=raw, connection=connection),
),
)
result = await client.replay.send(
session.id,
ReplaySendOptions(raw=raw, connection=connection),
)
# Create an empty replay session, then dispatch via ``send()``.
# Passing ``CreateReplaySessionFromRaw`` here would also seed a stored
# entry on the server side, leading the caller to observe two history
# rows per call (one without response from the create-step seed, one
# with response from the actual send). The empty-create + send flow
# produces exactly one dispatched request.
session = await client.replay.sessions.create()
try:
result = await asyncio.wait_for(
client.replay.send(
session.id,
ReplaySendOptions(raw=raw, connection=connection),
),
timeout=_REPLAY_SEND_TIMEOUT_SECONDS,
)
except TimeoutError:
elapsed_ms = int((time.time() - started) * 1000)
return {
"session_id": str(session.id),
"status": "ERROR",
"error": (
f"Caido replay dispatch did not complete within "
f"{_REPLAY_SEND_TIMEOUT_SECONDS:.0f}s. The target may be "
"unroutable from the sandbox, or Caido's outbound HTTP client "
"is stalled. Check the target host/port and retry."
),
"elapsed_ms": elapsed_ms,
"response_raw": None,
}
elapsed_ms = int((time.time() - started) * 1000)
response_raw = result.entry.response_raw if hasattr(result.entry, "response_raw") else None
# ``result.entry.response`` is the parsed Response (with ``.raw`` bytes
# when ``includeResponseRaw`` was True, which is the entries SDK's
# default). The previous ``result.entry.response_raw`` lookup matched
# no attribute on ReplayEntry and silently returned ``None``.
response = getattr(result.entry, "response", None)
response_raw = getattr(response, "raw", None) if response is not None else None
return {
"session_id": str(session.id),
"status": result.status,
@@ -335,23 +370,6 @@ async def view_request(request_id: str, *, part: RequestPart = "request") -> Any
return await get_request_with_client(await get_client(), request_id, part=part)
async def send_request(
method: str,
url: str,
*,
headers: dict[str, str] | None = None,
body: str = "",
) -> dict[str, Any]:
"""Send an arbitrary raw HTTP request through Caido Replay."""
connection, raw = build_raw_request(
method=method,
url=url,
headers=headers or {},
body=body,
)
return await replay_send_raw(await get_client(), raw=raw, connection=connection)
async def repeat_request(
request_id: str,
*,
@@ -432,6 +450,5 @@ __all__ = [
"list_requests",
"repeat_request",
"scope_rules",
"send_request",
"view_request",
]
strix/tools/proxy/tools.py
+22 -55
View File
@@ -1,12 +1,15 @@
"""Caido proxy tools — host-side ``@function_tool`` wrappers.
The five tools delegate to :mod:`strix.tools.proxy.caido_api` for the actual
The four tools delegate to :mod:`strix.tools.proxy.caido_api` for the actual
caido-sdk-client work and add LLM-friendly JSON serialization + error
wrapping on top. The delegated ``caido_api.py`` module is also copied into
the sandbox image as the importable ``caido_api`` Python module.
Tools: ``list_requests``, ``view_request``, ``send_request``,
``repeat_request``, ``scope_rules``.
Tools: ``list_requests``, ``view_request``, ``repeat_request``,
``scope_rules``. Arbitrary one-off requests should be made through
``exec_command`` (e.g. ``curl``) — they're captured automatically via
the sandbox's ``HTTP_PROXY`` env, so wrapping them in a Strix tool only
adds an extra layer of indirection.
"""
from __future__ import annotations
@@ -160,6 +163,21 @@ async def list_requests(
for edge in connection.edges:
req = edge.node.request
resp = edge.node.response
response_payload: dict[str, Any] | None = None
if resp is not None:
response_payload = {
"id": resp.id,
"status_code": resp.status_code,
"length": resp.length,
"created_at": resp.created_at.isoformat(),
}
# Caido populates ``roundtripTime`` for some traffic sources
# and leaves it as ``0`` for others (notably proxy captures
# of upstream env-routed traffic). Surface the value only
# when it's actually measured so the model doesn't waste
# tokens reading a zero field on every entry.
if resp.roundtrip_time:
response_payload["roundtrip_ms"] = resp.roundtrip_time
entries.append(
{
"cursor": edge.cursor,
@@ -173,17 +191,7 @@ async def list_requests(
"is_tls": req.is_tls,
"created_at": req.created_at.isoformat(),
},
"response": (
{
"id": resp.id,
"status_code": resp.status_code,
"length": resp.length,
"roundtrip_ms": resp.roundtrip_time,
"created_at": resp.created_at.isoformat(),
}
if resp is not None
else None
),
"response": response_payload,
},
)
@@ -326,47 +334,6 @@ def _format_text_page(content: str, *, page: int, page_size: int) -> dict[str, A
}
# ----------------------------------------------------------------------
# send_request
# ----------------------------------------------------------------------
@function_tool(timeout=120, strict_mode=False)
async def send_request(
ctx: RunContextWrapper,
method: str,
url: str,
headers: dict[str, str] | None = None,
body: str = "",
timeout: int = 30,
) -> str:
"""Send an arbitrary HTTP request through the Caido proxy.
Use this for one-off probes (test endpoints, reach external APIs).
For modifying-and-replaying a request you've already captured, use
``repeat_request`` instead — it inherits the original headers /
cookies / auth and only patches the fields you specify.
Args:
method: ``"GET"`` / ``"POST"`` / ``"PUT"`` / ``"DELETE"`` / etc.
url: Full URL with protocol.
headers: Optional header dict.
body: Optional request body string.
timeout: Per-request timeout in seconds (default 30).
"""
del timeout # The SDK applies its own timeout via the GraphQL settings.
client = _ctx_client(ctx)
if client is None:
return _no_client()
try:
connection, raw = caido_api.build_raw_request(
method=method, url=url, headers=headers or {}, body=body
)
result = await caido_api.replay_send_raw(client, raw=raw, connection=connection)
return _format_replay_tool_result(result)
except Exception as exc: # noqa: BLE001
return _err("send_request", exc)
# ----------------------------------------------------------------------
# repeat_request
# ----------------------------------------------------------------------