diff --git a/docs/tools/proxy.mdx b/docs/tools/proxy.mdx index ea023c7..2c920c3 100644 --- a/docs/tools/proxy.mdx +++ b/docs/tools/proxy.mdx @@ -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 diff --git a/strix/agents/factory.py b/strix/agents/factory.py index 218f568..2e6ff5f 100644 --- a/strix/agents/factory.py +++ b/strix/agents/factory.py @@ -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) diff --git a/strix/agents/prompts/system_prompt.jinja b/strix/agents/prompts/system_prompt.jinja index bcc192d..09dbced 100644 --- a/strix/agents/prompts/system_prompt.jinja +++ b/strix/agents/prompts/system_prompt.jinja @@ -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 diff --git a/strix/interface/tui/renderers/proxy_renderer.py b/strix/interface/tui/renderers/proxy_renderer.py index 3297d2b..5723a57 100644 --- a/strix/interface/tui/renderers/proxy_renderer.py +++ b/strix/interface/tui/renderers/proxy_renderer.py @@ -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" diff --git a/strix/skills/tooling/python.md b/strix/skills/tooling/python.md index c4278ef..b47ff76 100644 --- a/strix/skills/tooling/python.md +++ b/strix/skills/tooling/python.md @@ -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: diff --git a/strix/tools/proxy/caido_api.py b/strix/tools/proxy/caido_api.py index d7aa9d6..fd02f8f 100644 --- a/strix/tools/proxy/caido_api.py +++ b/strix/tools/proxy/caido_api.py @@ -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", ] diff --git a/strix/tools/proxy/tools.py b/strix/tools/proxy/tools.py index beb0919..8b3465a 100644 --- a/strix/tools/proxy/tools.py +++ b/strix/tools/proxy/tools.py @@ -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 # ----------------------------------------------------------------------