diff --git a/containers/Dockerfile b/containers/Dockerfile index 7de7128..6249528 100644 --- a/containers/Dockerfile +++ b/containers/Dockerfile @@ -78,13 +78,6 @@ RUN pipx install arjun && \ pipx inject dirsearch setuptools && \ pipx install wafw00f -# ``python_action`` ships a fresh ``caido-sdk-client`` per call into /tmp. -# Install it system-wide so the driver's ``import caido_sdk_client`` resolves -# without venv activation. The helper *logic* lives host-side in -# ``strix/tools/proxy/_calls.py`` (shipped at runtime) — only the SDK dep -# is image-baked. -RUN pip install --break-system-packages --no-cache-dir caido-sdk-client - ENV NPM_CONFIG_PREFIX=/home/pentester/.npm-global RUN mkdir -p /home/pentester/.npm-global @@ -198,9 +191,13 @@ RUN mkdir -p /workspace && chown -R pentester:pentester /workspace /app USER pentester RUN python3 -m venv /app/.venv && \ + /app/.venv/bin/pip install --no-cache-dir caido-sdk-client && \ /app/.venv/bin/pip install --no-cache-dir -r /home/pentester/tools/jwt_tool/requirements.txt && \ ln -s /home/pentester/tools/jwt_tool/jwt_tool.py /home/pentester/.local/bin/jwt_tool +COPY --chown=pentester:pentester strix/tools/proxy/caido_api.py /opt/strix-python/caido_api.py +ENV PYTHONPATH=/opt/strix-python + RUN echo 'export PATH="/home/pentester/go/bin:/home/pentester/.local/bin:/home/pentester/.npm-global/bin:$PATH"' >> /home/pentester/.bashrc && \ echo 'export PATH="/home/pentester/go/bin:/home/pentester/.local/bin:/home/pentester/.npm-global/bin:$PATH"' >> /home/pentester/.profile diff --git a/docs/tools/proxy.mdx b/docs/tools/proxy.mdx index 39b7be6..ea023c7 100644 --- a/docs/tools/proxy.mdx +++ b/docs/tools/proxy.mdx @@ -30,23 +30,33 @@ The agent can take any captured request and replay it with modifications: ## Python Integration -All proxy functions are automatically available in Python sessions. This enables powerful scripted security testing: +Proxy helpers are available to sandbox Python scripts through the image-baked `caido_api` module. This enables powerful scripted security testing: ```python -# List recent POST requests -post_requests = list_requests( - httpql_filter='req.method.eq:"POST"', - page_size=20 -) +import asyncio -# View a specific request -request_details = view_request("req_123", part="request") +from caido_api import list_requests, repeat_request, view_request -# Replay with modified payload -response = repeat_request("req_123", { - "body": '{"user_id": "admin"}' -}) -print(f"Status: {response['status_code']}") + +async def main(): + # List recent POST requests + post_requests = await list_requests( + httpql_filter='req.method.eq:"POST"', + first=20, + ) + + # View a specific request + request_details = await view_request("req_123", part="request") + + # Replay with modified payload + response = await repeat_request( + "req_123", + modifications={"body": '{"user_id": "admin"}'}, + ) + print(response["status"], request_details is not None, len(post_requests.edges)) + + +asyncio.run(main()) ``` ### Available Functions @@ -58,26 +68,34 @@ print(f"Status: {response['status_code']}") | `repeat_request()` | Replay a request with modifications | | `send_request()` | Send a new HTTP request | | `scope_rules()` | Manage proxy scope (allowlist/denylist) | -| `list_sitemap()` | View discovered endpoints | -| `view_sitemap_entry()` | Get details for a sitemap entry | ### Example: Automated IDOR Testing ```python +import asyncio + # Get all requests to user endpoints -user_requests = list_requests( - httpql_filter='req.path.cont:"/users/"' -) +from caido_api import list_requests, repeat_request -for req in user_requests.get('requests', []): - # Try accessing with different user IDs - for test_id in ['1', '2', 'admin', '../admin']: - response = repeat_request(req['id'], { - 'url': req['path'].replace('/users/1', f'/users/{test_id}') - }) - if response['status_code'] == 200: - print(f"Potential IDOR: {test_id} returned 200") +async def main(): + user_requests = await list_requests(httpql_filter='req.path.cont:"/users/"') + + for edge in user_requests.edges: + req = edge.node.request + scheme = "https" if req.is_tls else "http" + for test_id in ["1", "2", "admin", "../admin"]: + url = f"{scheme}://{req.host}{req.path.replace('/users/1', f'/users/{test_id}')}" + response = await repeat_request( + req.id, + modifications={"url": url}, + ) + print(req.id, test_id, response["status"]) + if response["status"] == "DONE": + print(f"Replay completed for candidate {test_id}") + + +asyncio.run(main()) ``` ## Human-in-the-Loop diff --git a/pyproject.toml b/pyproject.toml index 8afa285..301cc2f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -210,9 +210,6 @@ ignore = [ "strix/tools/thinking/tool.py" = ["TC002"] "strix/tools/web_search/tool.py" = ["TC002"] "strix/tools/proxy/tools.py" = ["TC002", "PLR0911"] -# Driver script writes to /tmp inside the sandbox container — single-user, -# isolated rootfs, so the multi-user race S108 warns about doesn't apply. -"strix/tools/python/tool.py" = ["S108", "TC002"] "strix/tools/agents_graph/tools.py" = ["TC002"] "strix/agents/factory.py" = ["TC002"] # Entry point: ``Path`` is used at runtime by the typing of the diff --git a/strix/agents/factory.py b/strix/agents/factory.py index 83a09b3..218f568 100644 --- a/strix/agents/factory.py +++ b/strix/agents/factory.py @@ -51,7 +51,6 @@ from strix.tools.proxy.tools import ( send_request, view_request, ) -from strix.tools.python.tool import python_action from strix.tools.reporting.tool import create_vulnerability_report from strix.tools.thinking.tool import think from strix.tools.todo.tools import ( @@ -272,8 +271,6 @@ _BASE_TOOLS: tuple[Tool, ...] = ( send_request, repeat_request, scope_rules, - # Stateless Python execution with proxy helpers pre-bound - python_action, # Multi-agent graph tools (the coordinator is in ctx.context) view_agent_graph, send_message_to_agent, diff --git a/strix/agents/prompt.py b/strix/agents/prompt.py index 6b935b5..bccfe78 100644 --- a/strix/agents/prompt.py +++ b/strix/agents/prompt.py @@ -37,8 +37,8 @@ def _resolve_skills( 2. ``scan_modes/`` (always). 3. ``tooling/agent_browser`` (always — every agent has shell + the agent-browser CLI). - 4. ``tooling/python`` (always — every agent has the ``python_action`` - tool with proxy helpers pre-bound). + 4. ``tooling/python`` (always — Python runs through ``exec_command``; + sandbox scripts can import ``caido_api`` for Caido automation). 5. ``coordination/root_agent`` for the root agent only — orchestration guidance for delegating to specialist subagents. 6. Whitebox-specific skills if applicable. diff --git a/strix/agents/prompts/system_prompt.jinja b/strix/agents/prompts/system_prompt.jinja index 3491794..bcc192d 100644 --- a/strix/agents/prompts/system_prompt.jinja +++ b/strix/agents/prompts/system_prompt.jinja @@ -161,14 +161,19 @@ OPERATIONAL PRINCIPLES: EFFICIENCY TACTICS: - Automate with Python scripts for complex workflows and repetitive inputs/tasks - Batch similar operations together -- Use captured traffic from proxy in Python tool to automate analysis +- Use captured traffic from the proxy tools directly, or import `caido_api` + from sandbox Python scripts when proxy automation is easier in code - Download additional tools as needed for specific tasks - Run multiple scans in parallel when possible - Load the most relevant skill before starting a specialized testing workflow if doing so will improve accuracy, speed, or tool usage -- Prefer the python tool for Python code. Do NOT embed Python in terminal commands via heredocs, here-strings, python -c, or interactive REPL driving unless shell-only behavior is specifically required -- The python tool exists to give you persistent interpreter state, structured code execution, cleaner debugging, and easier multi-step automation than terminal-wrapped Python +- Use `exec_command` for Python code: write reusable scripts under + `/workspace/scratch/` and run them with `python3`. For one-off snippets, + `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` - 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 the python or terminal tools +- 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 - Generate/adapt large payload corpora: combine encodings (URL, unicode, base64), comment styles, wrappers, time-based/differential probes. Expand with wordlists/templates - Use the web_search tool to fetch and refresh payload sets (latest bypasses, WAF evasions, DB-specific syntax, browser/JS quirks) and incorporate them into sprays @@ -404,7 +409,8 @@ SPECIALIZED TOOLS: - interactsh-client - OOB interaction testing PROXY & INTERCEPTION: -- Caido CLI - Modern web proxy (already running). Used with proxy tool or with python tool (functions already imported). +- Caido CLI - Modern web proxy (already running). Use the proxy tools + directly, or import `caido_api` from sandbox Python scripts. - NOTE: If you are seeing proxy errors when sending requests, it usually means you are not sending requests to a correct url/host/port. - Ignore Caido proxy-generated 50x HTML error pages; these are proxy issues (might happen when requesting a wrong host or SSL/TLS issues, etc). diff --git a/strix/interface/assets/tui_styles.tcss b/strix/interface/assets/tui_styles.tcss index d1097de..d2984c8 100644 --- a/strix/interface/assets/tui_styles.tcss +++ b/strix/interface/assets/tui_styles.tcss @@ -386,7 +386,6 @@ VulnerabilityDetailScreen { .browser-tool, .terminal-tool, -.python-tool, .agents-graph-tool, .file-edit-tool, .proxy-tool, @@ -411,8 +410,6 @@ VulnerabilityDetailScreen { .browser-tool.status-running, .terminal-tool.status-completed, .terminal-tool.status-running, -.python-tool.status-completed, -.python-tool.status-running, .agents-graph-tool.status-completed, .agents-graph-tool.status-running, .file-edit-tool.status-completed, diff --git a/strix/skills/scan_modes/standard.md b/strix/skills/scan_modes/standard.md index d5f7feb..b7a3739 100644 --- a/strix/skills/scan_modes/standard.md +++ b/strix/skills/scan_modes/standard.md @@ -77,7 +77,7 @@ Test each attack surface methodically. Spawn focused subagents for different are - Demonstrate actual impact, not theoretical risk - Chain vulnerabilities to show maximum severity - Document full attack path from entry to impact -- Use python tool for complex exploit development +- Use Python scripts through `exec_command` for complex exploit development ## Phase 5: Reporting diff --git a/strix/skills/tooling/python.md b/strix/skills/tooling/python.md index a8792e1..c4278ef 100644 --- a/strix/skills/tooling/python.md +++ b/strix/skills/tooling/python.md @@ -1,113 +1,74 @@ --- name: python -description: python_action — execute Python in the sandbox with Caido proxy helpers (list_requests, view_request, send_request, repeat_request, scope_rules) pre-bound as awaitables. Stateless per call; persistence via files. +description: Run Python through exec_command in the SDK sandbox. Use the image-baked caido_api module for Caido proxy automation from Python scripts. --- +# Python In The Sandbox -# python_action — when and how +Use `exec_command` for Python. There is no separate Strix Python executor. -Use ``python_action`` for any Python-side work: payload encoding/decoding, -parsing/transforming captured HTTP traffic, crypto operations, custom -exploit scripts, log/JSON analysis. Use ``exec_command`` for shell tools -(nmap, sqlmap, ffuf, agent-browser, package managers, daemons). +Prefer writing reusable scripts to `/workspace/scratch/.py` and +running them with `python3 /workspace/scratch/.py`. For short +one-off transformations, `python3 -c` or a small here-document is fine. -**Do not** wrap Python in bash heredocs, ``python3 -c`` one-liners, or -``echo | python3`` chains via ``exec_command`` — ``python_action`` exists -so structured output replaces fragile stdout parsing. +## Proxy Automation From Python -## What's pre-bound (no imports needed) - -All proxy helpers are **async** — call them with ``await``: - -- ``list_requests(httpql_filter=, first=50, after=, sort_by=, sort_order=, - scope_id=)`` → cursor-paginated SDK ``Connection``. Iterate - ``connection.edges``; each edge has ``.cursor`` and ``.node.request`` / - ``.node.response``. -- ``view_request(request_id, part="request")`` → SDK request object. - ``.request.raw`` and ``.response.raw`` are bytes. -- ``send_request(method, url, headers=None, body="")`` → dict with - ``status``, ``error``, ``elapsed_ms``, ``response_raw`` (bytes or None), - ``session_id``. -- ``repeat_request(request_id, modifications={...})`` → same shape. - ``modifications`` keys: ``url`` / ``params`` / ``headers`` / ``body`` / - ``cookies``. -- ``scope_rules(action, allowlist=, denylist=, scope_id=, scope_name=)`` - — same actions as the host-side tool (``list``/``get``/``create``/ - ``update``/``delete``). - -Top-level ``await`` works — the body is wrapped in an async function for -you. ``print()`` to emit visible output; the last expression is **not** -auto-shown. - -## Stateless model + how to keep state - -Each ``python_action`` call is a **fresh process**: variables, imports, -and definitions do not survive. To carry state across steps: - -- **Combine into one call** when the workflow is short — write the full - multi-step routine as one ``code`` block. -- **Persist to disk** for longer-lived state. ``/workspace/scratch/`` is - pentester-writable and survives across calls within a scan. -- **Build a script** with ``apply_patch`` to ``/workspace/scratch/.py`` - and run it via ``exec_command python3 ...`` when you need a file the - agent can iterate on. - -## Examples - -### Hunt SQLi candidates by inspecting captured traffic +The sandbox image includes an installed `caido_api` module. Import it +explicitly when Python code needs Caido traffic or replay access: ```python -# All POSTs that look interesting -posts = await list_requests( - httpql_filter='req.method.eq:"POST" AND req.path.cont:"/api/"', - first=50, +from caido_api import ( + list_requests, + repeat_request, + scope_rules, + send_request, + view_request, ) -candidates = [] -for edge in posts.edges: - body = await view_request(edge.node.request.id, part="request") - raw = body.request.raw.decode("utf-8", errors="replace") - if "id=" in raw or "user=" in raw: - candidates.append(edge.node.request.id) - -print(f"{len(candidates)} candidates") -print(candidates[:10]) ``` -### Replay with a SQLi probe and a tampered cookie +All helpers are async. Use them inside `asyncio.run(...)` or an async +function: ```python -result = await repeat_request( - "req_abc123", - modifications={ - "params": {"id": "1' OR '1'='1"}, - "cookies": {"session": "ATTACKER_TOKEN"}, - }, -) -print(result["status"], result["elapsed_ms"], "ms") -if result["response_raw"]: - print(result["response_raw"].decode("utf-8", errors="replace")[:500]) +import asyncio + +from caido_api import list_requests, view_request + + +async def main(): + posts = await list_requests( + httpql_filter='req.method.eq:"POST" AND req.path.cont:"/api/"', + first=50, + ) + candidates = [] + for edge in posts.edges: + request_id = edge.node.request.id + body = await view_request(request_id, part="request") + raw = body.request.raw.decode("utf-8", errors="replace") + if "id=" in raw or "user=" in raw: + candidates.append(request_id) + + print(f"{len(candidates)} candidates") + print(candidates[:10]) + + +asyncio.run(main()) ``` -### Decode/encode payloads +Available helpers: -```python -import base64, urllib.parse, hashlib +- `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. -token = "eyJhbGciOiJIUzI1NiJ9.eyJ1c2VyIjoiYWxpY2UifQ.sig" -header_b64, payload_b64, _ = token.split(".") -print(base64.urlsafe_b64decode(payload_b64 + "==")) -``` +## Workflow -### Iterate an exploit by writing to scratch - -When iterating, prefer writing the script to disk so you can edit-and-rerun -without re-sending the whole code each call: +For iterative exploit work, put code in a file: ```text -# 1. Use apply_patch to create /workspace/scratch/exploit.py -# 2. exec_command: python3 /workspace/scratch/exploit.py -# 3. Edit + re-run; repeat until working +1. Create or edit `/workspace/scratch/exploit.py` with `apply_patch`. +2. Run it with `exec_command`: `python3 /workspace/scratch/exploit.py`. +3. Edit and rerun until the proof-of-concept is reliable. ``` - -For one-shot crypto/encoding work or a single proxy-data analysis, -``python_action`` is the cleaner choice. diff --git a/strix/tools/proxy/_calls.py b/strix/tools/proxy/caido_api.py similarity index 53% rename from strix/tools/proxy/_calls.py rename to strix/tools/proxy/caido_api.py index b6b3295..d7aa9d6 100644 --- a/strix/tools/proxy/_calls.py +++ b/strix/tools/proxy/caido_api.py @@ -1,28 +1,16 @@ -"""Pure caido-sdk-client call sequences shared by ``tools.py`` and ``python_action``. - -Functions here: - -- Take an explicit ``Client`` argument — no module-level state, no - context lookups. The caller decides what client to use. -- Return raw caido-sdk-client objects (or dicts of primitives where the - composition itself is the value-add, like :func:`replay_send_raw`). -- Live without ``@function_tool`` decorators, ``RunContextWrapper``, or - any host-side framework dependency. They run identically inside the - Strix host process (called from ``tools.py``) and inside the sandbox - container (called from the ``python_action`` driver), against the - same Caido instance — host gets there via the host-mapped port, - container gets there via ``localhost:48080``. - -Single source of truth for the Caido SDK call shapes; ``tools.py`` adds -LLM-specific JSON serialization and error wrapping on top. -""" +"""Shared Caido proxy helpers and sandbox-importable ``caido_api`` module.""" from __future__ import annotations +import asyncio +import json +import os import time +import urllib.request from typing import TYPE_CHECKING, Any, Literal from urllib.parse import parse_qs, urlencode, urlparse, urlunparse +from caido_sdk_client import Client, TokenAuthOptions from caido_sdk_client.types import ( ConnectionInfoInput, CreateReplaySessionFromRaw, @@ -35,7 +23,7 @@ from caido_sdk_client.types import ( if TYPE_CHECKING: - from caido_sdk_client import Client + from caido_sdk_client import Client as CaidoClient RequestPart = Literal["request", "response"] @@ -50,8 +38,10 @@ SortBy = Literal[ "source", ] SortOrder = Literal["asc", "desc"] +ScopeAction = Literal["get", "list", "create", "update", "delete"] - +_DEFAULT_CAIDO_URL = "http://127.0.0.1:48080" +_CLIENT_CACHE: dict[str, Client] = {} _REQ_FIELD_MAP: dict[SortBy, tuple[str, str]] = { "timestamp": ("req", "created_at"), "host": ("req", "host"), @@ -64,11 +54,56 @@ _REQ_FIELD_MAP: dict[SortBy, tuple[str, str]] = { } -# ---------------------------------------------------------------------- -# Requests — list / get -# ---------------------------------------------------------------------- -async def list_requests( - client: Client, +def caido_url() -> str: + """Return the in-sandbox Caido endpoint used by ``caido_api``.""" + return os.environ.get("STRIX_CAIDO_URL", _DEFAULT_CAIDO_URL).rstrip("/") + + +def _graphql_url() -> str: + base_url = caido_url() + parsed = urlparse(base_url) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ValueError(f"Invalid Caido URL: {base_url}") + return f"{base_url}/graphql" + + +def _login_as_guest() -> str: + body = json.dumps({"query": "mutation { loginAsGuest { token { accessToken } } }"}).encode( + "utf-8" + ) + req = urllib.request.Request( # noqa: S310 + _graphql_url(), + data=body, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(req, timeout=10) as resp: # noqa: S310 # nosec B310 + payload = json.loads(resp.read()) + return str(payload["data"]["loginAsGuest"]["token"]["accessToken"]) + + +async def get_client() -> Client: + """Return a connected Caido SDK client for the local sandbox sidecar.""" + if client := _CLIENT_CACHE.get("default"): + return client + + token = await asyncio.to_thread(_login_as_guest) + client = Client(caido_url(), auth=TokenAuthOptions(token=token)) + await client.connect() + _CLIENT_CACHE["default"] = client + return client + + +async def close_client() -> None: + """Close the cached sandbox Caido client, if one was opened.""" + client = _CLIENT_CACHE.pop("default", None) + if client is None: + return + await client.aclose() + + +async def list_requests_with_client( + client: CaidoClient, *, httpql_filter: str | None = None, first: int = 50, @@ -89,8 +124,8 @@ async def list_requests( return await builder.execute() -async def get_request( - client: Client, +async def get_request_with_client( + client: CaidoClient, request_id: str, *, part: RequestPart = "request", @@ -102,9 +137,6 @@ async def get_request( return await client.request.get(request_id, opts) -# ---------------------------------------------------------------------- -# Raw HTTP request build / parse / mutate -# ---------------------------------------------------------------------- def build_raw_request( *, method: str, @@ -131,7 +163,6 @@ def build_raw_request( lines = [f"{method.upper()} {path} HTTP/1.1"] lines.extend(f"{k}: {v}" for k, v in final_headers.items()) raw = ("\r\n".join(lines) + "\r\n\r\n" + body).encode("utf-8") - return ConnectionInfoInput(host=host, port=port, is_tls=is_tls), raw @@ -183,13 +214,10 @@ def apply_modifications( existing = {k: v[0] if v else "" for k, v in parse_qs(parsed.query).items()} existing.update(modifications["params"]) final_url = urlunparse(parsed._replace(query=urlencode(existing))) - if "headers" in modifications: headers.update(modifications["headers"]) - if "body" in modifications: body = modifications["body"] - if "cookies" in modifications: cookies: dict[str, str] = {} if headers.get("Cookie"): @@ -208,11 +236,8 @@ def apply_modifications( } -# ---------------------------------------------------------------------- -# Replay — send raw bytes, get a result -# ---------------------------------------------------------------------- async def replay_send_raw( - client: Client, + client: CaidoClient, *, raw: bytes, connection: ConnectionInfoInput, @@ -238,19 +263,16 @@ async def replay_send_raw( } -# ---------------------------------------------------------------------- -# Scope CRUD -# ---------------------------------------------------------------------- -async def scope_list(client: Client) -> Any: +async def scope_list(client: CaidoClient) -> Any: return await client.scope.list() -async def scope_get(client: Client, scope_id: str) -> Any: +async def scope_get(client: CaidoClient, scope_id: str) -> Any: return await client.scope.get(scope_id) async def scope_create( - client: Client, + client: CaidoClient, *, name: str, allowlist: list[str] | None = None, @@ -266,7 +288,7 @@ async def scope_create( async def scope_update( - client: Client, + client: CaidoClient, scope_id: str, *, name: str, @@ -283,5 +305,133 @@ async def scope_update( ) -async def scope_delete(client: Client, scope_id: str) -> None: +async def scope_delete(client: CaidoClient, scope_id: str) -> None: await client.scope.delete(scope_id) + + +async def list_requests( + *, + httpql_filter: str | None = None, + first: int = 50, + after: str | None = None, + sort_by: SortBy = "timestamp", + sort_order: SortOrder = "desc", + scope_id: str | None = None, +) -> Any: + """List captured HTTP requests from sandbox Python.""" + return await list_requests_with_client( + await get_client(), + httpql_filter=httpql_filter, + first=first, + after=after, + sort_by=sort_by, + sort_order=sort_order, + scope_id=scope_id, + ) + + +async def view_request(request_id: str, *, part: RequestPart = "request") -> Any: + """Return one captured request/response from sandbox Python.""" + 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, + *, + modifications: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Replay a captured request after applying request modifications.""" + mods = modifications or {} + result = await get_request_with_client(await get_client(), request_id, part="request") + if result is None or result.request.raw is None: + raise ValueError(f"Request {request_id} not found") + + original = result.request + raw_str = result.request.raw.decode("utf-8", errors="replace") + components = parse_raw_request(raw_str) + full_url = full_url_from_components(original, components, mods) + modified = apply_modifications(components, mods, full_url) + connection, raw = build_raw_request( + method=modified["method"], + url=modified["url"], + headers=modified["headers"], + body=modified["body"], + ) + return await replay_send_raw(await get_client(), raw=raw, connection=connection) + + +async def scope_rules( + action: ScopeAction, + *, + allowlist: list[str] | None = None, + denylist: list[str] | None = None, + scope_id: str | None = None, + scope_name: str | None = None, +) -> Any: + """Manage Caido scope rules from sandbox Python.""" + client = await get_client() + if action == "list": + result = await scope_list(client) + elif action == "get": + if not scope_id: + raise ValueError("scope_id required for get") + result = await scope_get(client, scope_id) + elif action == "create": + if not scope_name: + raise ValueError("scope_name required for create") + result = await scope_create( + client, + name=scope_name, + allowlist=allowlist, + denylist=denylist, + ) + elif action == "update": + if not scope_id or not scope_name: + raise ValueError("scope_id and scope_name required for update") + result = await scope_update( + client, + scope_id, + name=scope_name, + allowlist=allowlist, + denylist=denylist, + ) + elif action == "delete": + if not scope_id: + raise ValueError("scope_id required for delete") + await scope_delete(client, scope_id) + result = {"deleted": scope_id} + else: + raise ValueError(f"Unknown action: {action}") + return result + + +__all__ = [ + "RequestPart", + "ScopeAction", + "SortBy", + "SortOrder", + "close_client", + "get_client", + "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 52a25ef..beb0919 100644 --- a/strix/tools/proxy/tools.py +++ b/strix/tools/proxy/tools.py @@ -1,10 +1,9 @@ -"""Caido proxy tools — host-side ``@function_tool`` wrappers around ``_calls``. +"""Caido proxy tools — host-side ``@function_tool`` wrappers. -The five tools delegate to :mod:`strix.tools.proxy._calls` for the actual +The five 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 shared call layer is also reused by the -``python_action`` tool to expose the same proxy surface inside the -sandbox's Python kernel — single source of truth for the SDK shapes. +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``. @@ -22,7 +21,7 @@ from typing import TYPE_CHECKING, Any, Literal from agents import RunContextWrapper, function_tool -from strix.tools.proxy import _calls +from strix.tools.proxy import caido_api logger = logging.getLogger(__name__) @@ -31,13 +30,13 @@ logger = logging.getLogger(__name__) if TYPE_CHECKING: from caido_sdk_client import Client - from strix.tools.proxy._calls import RequestPart, SortBy, SortOrder + from strix.tools.proxy.caido_api import RequestPart, SortBy, SortOrder else: # Runtime import: ``function_tool`` resolves the annotations via # ``typing.get_type_hints`` so the Literal aliases must be reachable # in module globals at decoration time even though they're "only" # used in annotations. - from strix.tools.proxy._calls import ( # noqa: TC001 + from strix.tools.proxy.caido_api import ( # noqa: TC001 RequestPart, SortBy, SortOrder, @@ -52,8 +51,10 @@ def _ctx_client(ctx: RunContextWrapper) -> Client | None: return inner.get("caido_client") -def _serialize(value: Any) -> Any: - """Recursively convert SDK dataclasses/Pydantic objects to JSON-safe primitives.""" +# Tool-output formatting. Caido SDK returns typed Python objects; function +# tools need compact JSON-safe values for the model and TUI. +def _to_tool_json(value: Any) -> Any: + """Recursively convert SDK dataclasses/Pydantic objects to tool JSON values.""" if value is None or isinstance(value, str | int | float | bool): return value if isinstance(value, bytes): @@ -64,13 +65,13 @@ def _serialize(value: Any) -> Any: if isinstance(value, datetime): return value.isoformat() if is_dataclass(value) and not isinstance(value, type): - return {k: _serialize(v) for k, v in dataclasses.asdict(value).items()} + return {k: _to_tool_json(v) for k, v in dataclasses.asdict(value).items()} if hasattr(value, "model_dump"): - return _serialize(value.model_dump()) + return _to_tool_json(value.model_dump()) if isinstance(value, dict): - return {str(k): _serialize(v) for k, v in value.items()} + return {str(k): _to_tool_json(v) for k, v in value.items()} if isinstance(value, list | tuple | set): - return [_serialize(v) for v in value] + return [_to_tool_json(v) for v in value] return str(value) @@ -145,7 +146,7 @@ async def list_requests( return _no_client() try: - connection = await _calls.list_requests( + connection = await caido_api.list_requests_with_client( client, httpql_filter=httpql_filter, first=first, @@ -246,7 +247,7 @@ async def view_request( return _no_client() try: - result = await _calls.get_request(client, request_id, part=part) + result = await caido_api.get_request_with_client(client, request_id, part=part) if result is None: return json.dumps( {"success": False, "error": f"Request {request_id} not found"}, @@ -271,10 +272,14 @@ async def view_request( content = raw_bytes.decode("utf-8", errors="replace") if search_pattern: - return json.dumps(_regex_hits(content, search_pattern), ensure_ascii=False, default=str) + return json.dumps( + _format_search_hits(content, search_pattern), + ensure_ascii=False, + default=str, + ) return json.dumps( - _paginate_lines(content, page=page, page_size=page_size), + _format_text_page(content, page=page, page_size=page_size), ensure_ascii=False, default=str, ) @@ -282,7 +287,7 @@ async def view_request( return _err("view_request", exc) -def _regex_hits(content: str, pattern: str) -> dict[str, Any]: +def _format_search_hits(content: str, pattern: str) -> dict[str, Any]: try: regex = re.compile(pattern) except re.error as exc: @@ -307,7 +312,7 @@ def _regex_hits(content: str, pattern: str) -> dict[str, Any]: return {"success": True, "hits": hits, "total_hits": len(hits)} -def _paginate_lines(content: str, *, page: int, page_size: int) -> dict[str, Any]: +def _format_text_page(content: str, *, page: int, page_size: int) -> dict[str, Any]: lines = content.splitlines() start = max(0, (page - 1) * page_size) end = start + page_size @@ -353,11 +358,11 @@ async def send_request( return _no_client() try: - connection, raw = _calls.build_raw_request( + connection, raw = caido_api.build_raw_request( method=method, url=url, headers=headers or {}, body=body ) - result = await _calls.replay_send_raw(client, raw=raw, connection=connection) - return _format_replay_result(result) + 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) @@ -402,7 +407,7 @@ async def repeat_request( mods = modifications or {} try: - result = await _calls.get_request(client, request_id, part="request") + result = await caido_api.get_request_with_client(client, request_id, part="request") if result is None or result.request.raw is None: return json.dumps( {"success": False, "error": f"Request {request_id} not found"}, @@ -412,22 +417,22 @@ async def repeat_request( original = result.request raw_str = result.request.raw.decode("utf-8", errors="replace") - components = _calls.parse_raw_request(raw_str) - full_url = _calls.full_url_from_components(original, components, mods) - modified = _calls.apply_modifications(components, mods, full_url) - connection, raw = _calls.build_raw_request( + components = caido_api.parse_raw_request(raw_str) + full_url = caido_api.full_url_from_components(original, components, mods) + modified = caido_api.apply_modifications(components, mods, full_url) + connection, raw = caido_api.build_raw_request( method=modified["method"], url=modified["url"], headers=modified["headers"], body=modified["body"], ) - replay = await _calls.replay_send_raw(client, raw=raw, connection=connection) - return _format_replay_result(replay) + replay = await caido_api.replay_send_raw(client, raw=raw, connection=connection) + return _format_replay_tool_result(replay) except Exception as exc: # noqa: BLE001 return _err("repeat_request", exc) -def _format_replay_result(replay: dict[str, Any]) -> str: +def _format_replay_tool_result(replay: dict[str, Any]) -> str: response_raw = replay.get("response_raw") response: dict[str, Any] | None = None if response_raw is not None: @@ -501,9 +506,9 @@ async def scope_rules( try: if action == "list": - scopes = await _calls.scope_list(client) + scopes = await caido_api.scope_list(client) return json.dumps( - {"success": True, "scopes": [_serialize(s) for s in scopes]}, + {"success": True, "scopes": [_to_tool_json(s) for s in scopes]}, ensure_ascii=False, default=str, ) @@ -514,9 +519,9 @@ async def scope_rules( ensure_ascii=False, default=str, ) - scope = await _calls.scope_get(client, scope_id) + scope = await caido_api.scope_get(client, scope_id) return json.dumps( - {"success": True, "scope": _serialize(scope)}, ensure_ascii=False, default=str + {"success": True, "scope": _to_tool_json(scope)}, ensure_ascii=False, default=str ) if action == "create": if not scope_name: @@ -525,11 +530,11 @@ async def scope_rules( ensure_ascii=False, default=str, ) - scope = await _calls.scope_create( + scope = await caido_api.scope_create( client, name=scope_name, allowlist=allowlist, denylist=denylist ) return json.dumps( - {"success": True, "scope": _serialize(scope)}, ensure_ascii=False, default=str + {"success": True, "scope": _to_tool_json(scope)}, ensure_ascii=False, default=str ) if action == "update": if not scope_id or not scope_name: @@ -541,11 +546,11 @@ async def scope_rules( ensure_ascii=False, default=str, ) - scope = await _calls.scope_update( + scope = await caido_api.scope_update( client, scope_id, name=scope_name, allowlist=allowlist, denylist=denylist ) return json.dumps( - {"success": True, "scope": _serialize(scope)}, ensure_ascii=False, default=str + {"success": True, "scope": _to_tool_json(scope)}, ensure_ascii=False, default=str ) # action == "delete" — exhaustive Literal if not scope_id: @@ -554,7 +559,7 @@ async def scope_rules( ensure_ascii=False, default=str, ) - await _calls.scope_delete(client, scope_id) + await caido_api.scope_delete(client, scope_id) return json.dumps({"success": True, "deleted": scope_id}, ensure_ascii=False, default=str) except Exception as exc: # noqa: BLE001 return _err("scope_rules", exc) diff --git a/strix/tools/python/__init__.py b/strix/tools/python/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/strix/tools/python/tool.py b/strix/tools/python/tool.py deleted file mode 100644 index a65691c..0000000 --- a/strix/tools/python/tool.py +++ /dev/null @@ -1,308 +0,0 @@ -"""``python_action`` — execute Python code in the sandbox with proxy helpers. - -Stateless per-call. Each invocation: - -1. Ships the shared :mod:`strix.tools.proxy._calls` module source into - ``/tmp/`` inside the sandbox (single source of truth: the host-side - ``proxy/tools.py`` and this kernel both import the same file). The - logic is host-managed; updating the proxy helpers does not require - an image rebuild. -2. Writes a per-call driver that connects a fresh ``caido_sdk_client`` - to the in-container Caido at ``localhost:48080``, defines user-facing - wrappers (``list_requests``, ``view_request``, ``send_request``, - ``repeat_request``, ``scope_rules``) bound to that client, then - ``exec``s the user's code wrapped in an ``async def`` so top-level - ``await`` works. -3. Captures ``stdout`` / ``stderr``, parses a sentinel-delimited JSON - payload from the driver's output, and returns it as the tool result. - -State is **not** preserved across calls. For multi-step workflows, write -a single combined script via ``apply_patch`` and run it with -``exec_command``, or pass intermediate results through files in -``/workspace/scratch/``. -""" - -from __future__ import annotations - -import base64 -import importlib.resources -import io -import json -import logging -import uuid -from pathlib import Path -from typing import TYPE_CHECKING - -from agents import RunContextWrapper, function_tool - - -if TYPE_CHECKING: - from agents.sandbox.session.base_sandbox_session import BaseSandboxSession - - -logger = logging.getLogger(__name__) - - -_CALLS_SOURCE = (importlib.resources.files("strix.tools.proxy") / "_calls.py").read_text( - encoding="utf-8" -) - - -_DRIVER_TEMPLATE = '''\ -"""Auto-generated python_action driver. Do not edit.""" - -from __future__ import annotations - -import asyncio -import base64 -import io -import json -import sys -import textwrap -import traceback -import urllib.request - -sys.path.insert(0, "/tmp") -import strix_calls # noqa: E402 shipped alongside this driver - -from caido_sdk_client import Client # noqa: E402 -from caido_sdk_client.types import TokenAuthOptions # noqa: E402 - - -_SENTINEL = "{sentinel}" -_USER_CODE_B64 = "{user_code_b64}" - - -def _login_as_guest() -> str: - body = json.dumps( - {{"query": "mutation {{ loginAsGuest {{ token {{ accessToken }} }} }}"}} - ).encode("utf-8") - req = urllib.request.Request( - "http://127.0.0.1:48080/graphql", - data=body, - headers={{"Content-Type": "application/json"}}, - method="POST", - ) - with urllib.request.urlopen(req, timeout=10) as resp: # noqa: S310 - payload = json.loads(resp.read()) - return str(payload["data"]["loginAsGuest"]["token"]["accessToken"]) - - -async def _strix_main() -> None: - client = Client("http://127.0.0.1:48080", auth=TokenAuthOptions(token=_login_as_guest())) - await client.connect() - - async def list_requests(**kwargs): - return await strix_calls.list_requests(client, **kwargs) - - async def view_request(request_id, *, part="request"): - return await strix_calls.get_request(client, request_id, part=part) - - async def send_request(method, url, *, headers=None, body=""): - connection, raw = strix_calls.build_raw_request( - method=method, url=url, headers=headers or {{}}, body=body - ) - return await strix_calls.replay_send_raw(client, raw=raw, connection=connection) - - async def repeat_request(request_id, *, modifications=None): - mods = modifications or {{}} - result = await strix_calls.get_request(client, request_id, part="request") - if result is None or result.request.raw is None: - raise ValueError(f"Request {{request_id}} not found") - original = result.request - raw_str = result.request.raw.decode("utf-8", errors="replace") - components = strix_calls.parse_raw_request(raw_str) - full_url = strix_calls.full_url_from_components(original, components, mods) - modified = strix_calls.apply_modifications(components, mods, full_url) - connection, raw = strix_calls.build_raw_request( - method=modified["method"], - url=modified["url"], - headers=modified["headers"], - body=modified["body"], - ) - return await strix_calls.replay_send_raw(client, raw=raw, connection=connection) - - async def scope_rules(action, *, allowlist=None, denylist=None, scope_id=None, scope_name=None): - if action == "list": - return await strix_calls.scope_list(client) - if action == "get": - if not scope_id: - raise ValueError("scope_id required for get") - return await strix_calls.scope_get(client, scope_id) - if action == "create": - if not scope_name: - raise ValueError("scope_name required for create") - return await strix_calls.scope_create( - client, name=scope_name, allowlist=allowlist, denylist=denylist - ) - if action == "update": - if not scope_id or not scope_name: - raise ValueError("scope_id and scope_name required for update") - return await strix_calls.scope_update( - client, scope_id, name=scope_name, allowlist=allowlist, denylist=denylist - ) - if action == "delete": - if not scope_id: - raise ValueError("scope_id required for delete") - await strix_calls.scope_delete(client, scope_id) - return {{"deleted": scope_id}} - raise ValueError(f"Unknown action: {{action}}") - - user_code = base64.b64decode(_USER_CODE_B64).decode("utf-8") - wrapped = "async def __strix_user():\\n pass\\n" + textwrap.indent(user_code, " ") - - namespace: dict = {{ - "list_requests": list_requests, - "view_request": view_request, - "send_request": send_request, - "repeat_request": repeat_request, - "scope_rules": scope_rules, - "client": client, - }} - - out_buf = io.StringIO() - err_buf = io.StringIO() - error: str | None = None - - real_stdout = sys.stdout - real_stderr = sys.stderr - sys.stdout = out_buf - sys.stderr = err_buf - try: - try: - exec(compile(wrapped, "", "exec"), namespace) # noqa: S102 - await namespace["__strix_user"]() - except BaseException: - error = traceback.format_exc() - finally: - sys.stdout = real_stdout - sys.stderr = real_stderr - - payload = json.dumps( - {{ - "stdout": out_buf.getvalue(), - "stderr": err_buf.getvalue(), - "error": error, - }}, - ensure_ascii=False, - ) - sys.stdout.write("\\n" + _SENTINEL + payload + "\\n") - - -asyncio.run(_strix_main()) -''' - - -@function_tool(timeout=180) -async def python_action( - ctx: RunContextWrapper, - code: str, - timeout: int = 60, -) -> str: - """Execute Python code in the sandbox with Caido proxy helpers pre-bound. - - Each call is a fresh process — variables, imports, and function - definitions do **not** persist between calls. For multi-step - workflows, combine into a single ``code`` block, or write a script - to ``/workspace/scratch/.py`` via ``apply_patch`` and run with - ``exec_command``. - - **Pre-bound proxy helpers** (no imports needed; all are async — use - ``await``): - - - ``list_requests(httpql_filter=, first=, after=, sort_by=, - sort_order=, scope_id=)`` → cursor-paginated SDK ``Connection``. - Iterate ``connection.edges`` for entries. - - ``view_request(request_id, part="request")`` → SDK request object - with ``.request.raw`` / ``.response.raw`` bytes. - - ``send_request(method, url, headers=None, body="")`` → dict with - ``status``, ``response_raw``, ``elapsed_ms``. - - ``repeat_request(request_id, modifications={"url"|"headers"| - "body"|"params"|"cookies": ...})`` → same shape as - ``send_request``. - - ``scope_rules(action, allowlist=, denylist=, scope_id=, - scope_name=)`` → SDK scope objects. - - Top-level ``await`` is supported — the body is wrapped in an async - function. Use ``print()`` to emit visible output; the last - expression is **not** auto-shown. - - Args: - code: Python source to execute. Multi-line is fine. - timeout: Hard timeout in seconds (default 60). The container - is killed if the driver exceeds this. - - Returns: - JSON dict with ``success``, ``stdout``, ``stderr``, ``error`` - (a formatted traceback if the user code raised). - """ - inner = ctx.context if isinstance(ctx.context, dict) else {} - session: BaseSandboxSession | None = inner.get("sandbox_session") - if session is None: - return json.dumps( - {"success": False, "error": "No sandbox session in run context"}, - ensure_ascii=False, - ) - - sentinel = "__STRIX_PY_RESULT_" + uuid.uuid4().hex + "__" - user_code_b64 = base64.b64encode(code.encode("utf-8")).decode("ascii") - driver = _DRIVER_TEMPLATE.format(sentinel=sentinel, user_code_b64=user_code_b64) - logger.info("python_action: invoking driver (code_len=%d, timeout=%ds)", len(code), timeout) - # /tmp inside the sandbox container is single-user (pentester) and - # disposable per scan; the multi-user race B108/S108 warns about - # doesn't apply. - driver_path = Path(f"/tmp/strix_driver_{uuid.uuid4().hex}.py") # nosec B108 - calls_path = Path("/tmp/strix_calls.py") # nosec B108 - - try: - await session.write(calls_path, io.BytesIO(_CALLS_SOURCE.encode("utf-8"))) - await session.write(driver_path, io.BytesIO(driver.encode("utf-8"))) - result = await session.exec("python3", "-u", str(driver_path), timeout=timeout) - except Exception as exc: # noqa: BLE001 - return json.dumps( - {"success": False, "error": f"python_action launch failed: {exc}"}, - ensure_ascii=False, - ) - finally: - # Best-effort cleanup; ignore failure (the file is in /tmp anyway). - try: - await session.exec("rm", "-f", str(driver_path), timeout=5) - except Exception: # noqa: BLE001 - logger.debug("cleanup failed for %s", driver_path) - - raw_stdout = result.stdout.decode("utf-8", errors="replace") - raw_stderr = result.stderr.decode("utf-8", errors="replace") - - if sentinel in raw_stdout: - head, _, tail = raw_stdout.rpartition(sentinel) - try: - payload = json.loads(tail.strip()) - except json.JSONDecodeError as exc: - return json.dumps( - { - "success": False, - "error": f"Driver output not parseable: {exc}", - "stdout": raw_stdout, - "stderr": raw_stderr, - }, - ensure_ascii=False, - ) - return json.dumps( - { - "success": payload.get("error") is None, - "stdout": (head.rstrip() + payload.get("stdout", "")) or "", - "stderr": payload.get("stderr", "") or raw_stderr, - "error": payload.get("error"), - }, - ensure_ascii=False, - ) - - return json.dumps( - { - "success": False, - "error": "Driver exited without producing a result sentinel", - "stdout": raw_stdout, - "stderr": raw_stderr, - }, - ensure_ascii=False, - )