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

feat(tools): python_action — stateless Python execution with proxy helpers

Browse Source 
Restores the legacy persistent-IPython tool's *ergonomics* (proxy
helpers pre-bound, structured stdout/stderr/error returns) without the
in-container daemon: each call ships ``strix.tools.proxy._calls`` source
into ``/tmp`` alongside a per-call driver, runs ``python3 -u`` against
it, and parses a sentinel-delimited JSON payload back from stdout. The
driver fetches its own guest token from Caido at ``localhost:48080``
and binds ``list_requests`` / ``view_request`` / ``send_request`` /
``repeat_request`` / ``scope_rules`` to that client; user code runs
inside an ``async def`` wrapper so top-level ``await`` works.

The proxy SDK call sequences live in one file —
``strix/tools/proxy/_calls.py`` — and are reused by both the host-side
``@function_tool`` wrappers (which add JSON serialization for the LLM)
and the in-container kernel (which exposes the bare async functions).
No code duplication; the helper logic itself is host-shipped, so
tweaking the proxy helpers does not require an image rebuild.

Image: a single ``pip install caido-sdk-client`` line so the driver's
``import caido_sdk_client`` resolves. Skill ``tooling/python`` is
always-loaded alongside ``tooling/agent_browser``.

Trade-off accepted: state does not persist across calls (no kernel).
For multi-step workflows the agent combines into one ``code`` block or
writes a script to ``/workspace/scratch/`` and runs via
``exec_command``. If a workflow surfaces that genuinely needs
persistence, the same tool surface migrates to a kernel-backed
executor without changing the LLM contract.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
0xallam
2026-04-25 22:58:53 -07:00
parent 767dc83581
commit 9d7f754b59
9 changed files with 805 additions and 260 deletions
Download Patch File Download Diff File Expand all files Collapse all files
strix/agents/factory.py
+3
View File
@@ -51,6 +51,7 @@ 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 (
@@ -99,6 +100,8 @@ _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 bus is in ctx.context)
view_agent_graph,
agent_status,
strix/agents/prompt.py
+5 -2
View File
@@ -37,13 +37,16 @@ def _resolve_skills(
2. ``scan_modes/<mode>`` (always).
3. ``tooling/agent_browser`` (always — every agent has shell + the
agent-browser CLI).
4. ``coordination/root_agent`` for the root agent only — orchestration
4. ``tooling/python`` (always — every agent has the ``python_action``
tool with proxy helpers pre-bound).
5. ``coordination/root_agent`` for the root agent only — orchestration
guidance for delegating to specialist subagents.
5. Whitebox-specific skills if applicable.
6. Whitebox-specific skills if applicable.
"""
ordered: list[str] = list(requested or [])
ordered.append(f"scan_modes/{scan_mode}")
ordered.append("tooling/agent_browser")
ordered.append("tooling/python")
if is_root:
ordered.append("coordination/root_agent")
if is_whitebox:
strix/skills/tooling/python.md
+113
View File
@@ -0,0 +1,113 @@
---
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.
---
# python_action — when and how
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).
**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.
## 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/<name>.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
```python
# All POSTs that look interesting
posts = await list_requests(
httpql_filter='req.method.eq:"POST" AND req.path.cont:"/api/"',
first=50,
)
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
```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])
```
### Decode/encode payloads
```python
import base64, urllib.parse, hashlib
token = "eyJhbGciOiJIUzI1NiJ9.eyJ1c2VyIjoiYWxpY2UifQ.sig"
header_b64, payload_b64, _ = token.split(".")
print(base64.urlsafe_b64decode(payload_b64 + "=="))
```
### 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:
```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
```
For one-shot crypto/encoding work or a single proxy-data analysis,
``python_action`` is the cleaner choice.
strix/tools/proxy/_calls.py
+287
View File
@@ -0,0 +1,287 @@
"""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.
"""
from __future__ import annotations
import time
from typing import TYPE_CHECKING, Any, Literal
from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
from caido_sdk_client.types import (
ConnectionInfoInput,
CreateReplaySessionFromRaw,
CreateReplaySessionOptions,
CreateScopeOptions,
ReplaySendOptions,
RequestGetOptions,
UpdateScopeOptions,
)
if TYPE_CHECKING:
from caido_sdk_client import Client
RequestPart = Literal["request", "response"]
SortBy = Literal[
"timestamp",
"host",
"method",
"path",
"status_code",
"response_time",
"response_size",
"source",
]
SortOrder = Literal["asc", "desc"]
_REQ_FIELD_MAP: dict[SortBy, tuple[str, str]] = {
"timestamp": ("req", "created_at"),
"host": ("req", "host"),
"method": ("req", "method"),
"path": ("req", "path"),
"source": ("req", "source"),
"status_code": ("resp", "code"),
"response_time": ("resp", "roundtrip"),
"response_size": ("resp", "length"),
}
# ----------------------------------------------------------------------
# Requests — list / get
# ----------------------------------------------------------------------
async def list_requests(
client: Client,
*,
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:
builder = client.request.list().first(first)
if httpql_filter:
builder = builder.filter(httpql_filter)
if after:
builder = builder.after(after)
if scope_id:
builder = builder.scope(scope_id)
target, field = _REQ_FIELD_MAP[sort_by]
builder = (builder.descending if sort_order == "desc" else builder.ascending)(target, field)
return await builder.execute()
async def get_request(
client: Client,
request_id: str,
*,
part: RequestPart = "request",
) -> Any:
opts = RequestGetOptions(
request_raw=(part == "request"),
response_raw=(part == "response"),
)
return await client.request.get(request_id, opts)
# ----------------------------------------------------------------------
# Raw HTTP request build / parse / mutate
# ----------------------------------------------------------------------
def build_raw_request(
*,
method: str,
url: str,
headers: dict[str, str],
body: str,
) -> tuple[ConnectionInfoInput, bytes]:
parsed = urlparse(url)
if not parsed.scheme or not parsed.netloc:
raise ValueError(f"Invalid URL: {url}")
is_tls = parsed.scheme.lower() == "https"
host = parsed.hostname or ""
port = parsed.port or (443 if is_tls else 80)
path = parsed.path or "/"
if parsed.query:
path = f"{path}?{parsed.query}"
final_headers = {**headers}
final_headers.setdefault("Host", parsed.netloc)
final_headers.setdefault("User-Agent", "strix")
if body and "Content-Length" not in {k.title() for k in final_headers}:
final_headers["Content-Length"] = str(len(body.encode("utf-8")))
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
def parse_raw_request(raw_content: str) -> dict[str, Any]:
lines = raw_content.split("\n")
request_line = lines[0].strip().split(" ")
if len(request_line) < 2:
raise ValueError("Invalid request line format")
method, url_path = request_line[0], request_line[1]
parsed_headers: dict[str, str] = {}
body_start = 0
for i, line in enumerate(lines[1:], 1):
if line.strip() == "":
body_start = i + 1
break
if ":" in line:
key, value = line.split(":", 1)
parsed_headers[key.strip()] = value.strip()
body = "\n".join(lines[body_start:]).strip() if body_start < len(lines) else ""
return {"method": method, "url_path": url_path, "headers": parsed_headers, "body": body}
def full_url_from_components(
original: Any,
components: dict[str, Any],
modifications: dict[str, Any],
) -> str:
if "url" in modifications:
return str(modifications["url"])
headers = components["headers"]
host_header = headers.get("Host") or original.host
scheme = "https" if original.is_tls else "http"
return f"{scheme}://{host_header}{components['url_path']}"
def apply_modifications(
components: dict[str, Any],
modifications: dict[str, Any],
full_url: str,
) -> dict[str, Any]:
headers = dict(components["headers"])
body = components["body"]
final_url = full_url
if "params" in modifications:
parsed = urlparse(final_url)
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"):
for cookie in headers["Cookie"].split(";"):
if "=" in cookie:
k, v = cookie.split("=", 1)
cookies[k.strip()] = v.strip()
cookies.update(modifications["cookies"])
headers["Cookie"] = "; ".join(f"{k}={v}" for k, v in cookies.items())
return {
"method": components["method"],
"url": final_url,
"headers": headers,
"body": body,
}
# ----------------------------------------------------------------------
# Replay — send raw bytes, get a result
# ----------------------------------------------------------------------
async def replay_send_raw(
client: Client,
*,
raw: bytes,
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),
)
elapsed_ms = int((time.time() - started) * 1000)
response_raw = result.entry.response_raw if hasattr(result.entry, "response_raw") else None
return {
"session_id": str(session.id),
"status": result.status,
"error": result.error,
"elapsed_ms": elapsed_ms,
"response_raw": response_raw,
}
# ----------------------------------------------------------------------
# Scope CRUD
# ----------------------------------------------------------------------
async def scope_list(client: Client) -> Any:
return await client.scope.list()
async def scope_get(client: Client, scope_id: str) -> Any:
return await client.scope.get(scope_id)
async def scope_create(
client: Client,
*,
name: str,
allowlist: list[str] | None = None,
denylist: list[str] | None = None,
) -> Any:
return await client.scope.create(
CreateScopeOptions(
name=name,
allowlist=list(allowlist or []),
denylist=list(denylist or []),
),
)
async def scope_update(
client: Client,
scope_id: str,
*,
name: str,
allowlist: list[str] | None = None,
denylist: list[str] | None = None,
) -> Any:
return await client.scope.update(
scope_id,
UpdateScopeOptions(
name=name,
allowlist=list(allowlist or []),
denylist=list(denylist or []),
),
)
async def scope_delete(client: Client, scope_id: str) -> None:
await client.scope.delete(scope_id)
strix/tools/proxy/tools.py
+80 -258
View File
@@ -1,9 +1,10 @@
"""Caido proxy tools — host-side via ``caido-sdk-client``.
"""Caido proxy tools — host-side ``@function_tool`` wrappers around ``_calls``.
The five tools delegate directly to ``caido_sdk_client.Client`` instances
held in the per-scan agent context. No sandbox round-trip; the SDK
talks GraphQL to the in-container Caido sidecar via the host-mapped
port resolved at session create time.
The five tools delegate to :mod:`strix.tools.proxy._calls` 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.
Tools: ``list_requests``, ``view_request``, ``send_request``,
``repeat_request``, ``scope_rules``.
@@ -14,55 +15,34 @@ from __future__ import annotations
import dataclasses
import json
import re
import time
from dataclasses import is_dataclass
from datetime import datetime
from typing import TYPE_CHECKING, Any, Literal
from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
from agents import RunContextWrapper, function_tool
from caido_sdk_client.types import (
ConnectionInfoInput,
CreateReplaySessionFromRaw,
CreateReplaySessionOptions,
CreateScopeOptions,
ReplaySendOptions,
RequestGetOptions,
UpdateScopeOptions,
)
from strix.tools.proxy import _calls
if TYPE_CHECKING:
from caido_sdk_client import Client
from strix.tools.proxy._calls 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
RequestPart,
SortBy,
SortOrder,
)
RequestPart = Literal["request", "response"]
SortBy = Literal[
"timestamp",
"host",
"method",
"path",
"status_code",
"response_time",
"response_size",
"source",
]
SortOrder = Literal["asc", "desc"]
ScopeAction = Literal["get", "list", "create", "update", "delete"]
_REQ_FIELD_MAP: dict[SortBy, tuple[str, str]] = {
"timestamp": ("req", "created_at"),
"host": ("req", "host"),
"method": ("req", "method"),
"path": ("req", "path"),
"source": ("req", "source"),
"status_code": ("resp", "code"),
"response_time": ("resp", "roundtrip"),
"response_size": ("resp", "length"),
}
def _ctx_client(ctx: RunContextWrapper) -> Client | None:
inner = ctx.context if isinstance(ctx.context, dict) else {}
return inner.get("caido_client")
@@ -92,10 +72,15 @@ def _serialize(value: Any) -> Any:
def _no_client() -> str:
return json.dumps(
{
"success": False,
"error": "Caido client not initialized in context.",
},
{"success": False, "error": "Caido client not available in run context"},
ensure_ascii=False,
default=str,
)
def _err(name: str, exc: Exception) -> str:
return json.dumps(
{"success": False, "error": f"{name} failed: {exc}"},
ensure_ascii=False,
default=str,
)
@@ -155,21 +140,15 @@ async def list_requests(
return _no_client()
try:
builder = client.request.list().first(first)
if httpql_filter:
builder = builder.filter(httpql_filter)
if after:
builder = builder.after(after)
if scope_id:
builder = builder.scope(scope_id)
target, field = _REQ_FIELD_MAP[sort_by]
if sort_order == "asc":
builder = builder.ascending(target, field)
else:
builder = builder.descending(target, field)
connection = await builder.execute()
connection = await _calls.list_requests(
client,
httpql_filter=httpql_filter,
first=first,
after=after,
sort_by=sort_by,
sort_order=sort_order,
scope_id=scope_id,
)
entries = []
for edge in connection.edges:
@@ -217,11 +196,7 @@ async def list_requests(
default=str,
)
except Exception as exc: # noqa: BLE001
return json.dumps(
{"success": False, "error": f"list_requests failed: {exc}"},
ensure_ascii=False,
default=str,
)
return _err("list_requests", exc)
# ----------------------------------------------------------------------
@@ -266,11 +241,7 @@ async def view_request(
return _no_client()
try:
opts = RequestGetOptions(
request_raw=(part == "request"),
response_raw=(part == "response"),
)
result = await client.request.get(request_id, opts)
result = await _calls.get_request(client, request_id, part=part)
if result is None:
return json.dumps(
{"success": False, "error": f"Request {request_id} not found"},
@@ -303,11 +274,7 @@ async def view_request(
default=str,
)
except Exception as exc: # noqa: BLE001
return json.dumps(
{"success": False, "error": f"view_request failed: {exc}"},
ensure_ascii=False,
default=str,
)
return _err("view_request", exc)
def _regex_hits(content: str, pattern: str) -> dict[str, Any]:
@@ -381,16 +348,13 @@ async def send_request(
return _no_client()
try:
connection, raw = _build_raw_request(
connection, raw = _calls.build_raw_request(
method=method, url=url, headers=headers or {}, body=body
)
return await _replay_send(client, raw=raw, connection=connection)
result = await _calls.replay_send_raw(client, raw=raw, connection=connection)
return _format_replay_result(result)
except Exception as exc: # noqa: BLE001
return json.dumps(
{"success": False, "error": f"send_request failed: {exc}"},
ensure_ascii=False,
default=str,
)
return _err("send_request", exc)
# ----------------------------------------------------------------------
@@ -433,7 +397,7 @@ async def repeat_request(
mods = modifications or {}
try:
result = await client.request.get(request_id, RequestGetOptions(request_raw=True))
result = await _calls.get_request(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"},
@@ -443,22 +407,38 @@ async def repeat_request(
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(
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(
method=modified["method"],
url=modified["url"],
headers=modified["headers"],
body=modified["body"],
)
return await _replay_send(client, raw=raw, connection=connection)
replay = await _calls.replay_send_raw(client, raw=raw, connection=connection)
return _format_replay_result(replay)
except Exception as exc: # noqa: BLE001
return json.dumps(
{"success": False, "error": f"repeat_request failed: {exc}"},
ensure_ascii=False,
default=str,
)
return _err("repeat_request", exc)
def _format_replay_result(replay: dict[str, Any]) -> str:
response_raw = replay.get("response_raw")
response: dict[str, Any] | None = None
if response_raw is not None:
response = {"raw": response_raw.decode("utf-8", errors="replace")}
return json.dumps(
{
"success": replay["status"] == "DONE",
"status": replay["status"],
"error": replay["error"],
"session_id": replay["session_id"],
"elapsed_ms": replay["elapsed_ms"],
"response": response,
},
ensure_ascii=False,
default=str,
)
# ----------------------------------------------------------------------
@@ -516,7 +496,7 @@ async def scope_rules(
try:
if action == "list":
scopes = await client.scope.list()
scopes = await _calls.scope_list(client)
return json.dumps(
{"success": True, "scopes": [_serialize(s) for s in scopes]},
ensure_ascii=False,
@@ -529,7 +509,7 @@ async def scope_rules(
ensure_ascii=False,
default=str,
)
scope = await client.scope.get(scope_id)
scope = await _calls.scope_get(client, scope_id)
return json.dumps(
{"success": True, "scope": _serialize(scope)}, ensure_ascii=False, default=str
)
@@ -540,12 +520,8 @@ async def scope_rules(
ensure_ascii=False,
default=str,
)
scope = await client.scope.create(
CreateScopeOptions(
name=scope_name,
allowlist=list(allowlist or []),
denylist=list(denylist or []),
),
scope = await _calls.scope_create(
client, name=scope_name, allowlist=allowlist, denylist=denylist
)
return json.dumps(
{"success": True, "scope": _serialize(scope)}, ensure_ascii=False, default=str
@@ -560,13 +536,8 @@ async def scope_rules(
ensure_ascii=False,
default=str,
)
scope = await client.scope.update(
scope_id,
UpdateScopeOptions(
name=scope_name,
allowlist=list(allowlist or []),
denylist=list(denylist or []),
),
scope = await _calls.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
@@ -578,156 +549,7 @@ async def scope_rules(
ensure_ascii=False,
default=str,
)
await client.scope.delete(scope_id)
await _calls.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 json.dumps(
{"success": False, "error": f"scope_rules failed: {exc}"},
ensure_ascii=False,
default=str,
)
# ----------------------------------------------------------------------
# Helpers — request build / parse / modify
# ----------------------------------------------------------------------
def _build_raw_request(
*,
method: str,
url: str,
headers: dict[str, str],
body: str,
) -> tuple[ConnectionInfoInput, bytes]:
parsed = urlparse(url)
if not parsed.scheme or not parsed.netloc:
raise ValueError(f"Invalid URL: {url}")
is_tls = parsed.scheme.lower() == "https"
host = parsed.hostname or ""
port = parsed.port or (443 if is_tls else 80)
path = parsed.path or "/"
if parsed.query:
path = f"{path}?{parsed.query}"
final_headers = {**headers}
final_headers.setdefault("Host", parsed.netloc)
final_headers.setdefault("User-Agent", "strix")
if body and "Content-Length" not in {k.title() for k in final_headers}:
final_headers["Content-Length"] = str(len(body.encode("utf-8")))
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
def _parse_raw_request(raw_content: str) -> dict[str, Any]:
lines = raw_content.split("\n")
request_line = lines[0].strip().split(" ")
if len(request_line) < 2:
raise ValueError("Invalid request line format")
method, url_path = request_line[0], request_line[1]
parsed_headers: dict[str, str] = {}
body_start = 0
for i, line in enumerate(lines[1:], 1):
if line.strip() == "":
body_start = i + 1
break
if ":" in line:
key, value = line.split(":", 1)
parsed_headers[key.strip()] = value.strip()
body = "\n".join(lines[body_start:]).strip() if body_start < len(lines) else ""
return {"method": method, "url_path": url_path, "headers": parsed_headers, "body": body}
def _full_url_from_components(
original: Any,
components: dict[str, Any],
modifications: dict[str, Any],
) -> str:
if "url" in modifications:
return str(modifications["url"])
headers = components["headers"]
host_header = headers.get("Host") or original.host
scheme = "https" if original.is_tls else "http"
return f"{scheme}://{host_header}{components['url_path']}"
def _apply_modifications(
components: dict[str, Any],
modifications: dict[str, Any],
full_url: str,
) -> dict[str, Any]:
headers = dict(components["headers"])
body = components["body"]
final_url = full_url
if "params" in modifications:
parsed = urlparse(final_url)
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"):
for cookie in headers["Cookie"].split(";"):
if "=" in cookie:
k, v = cookie.split("=", 1)
cookies[k.strip()] = v.strip()
cookies.update(modifications["cookies"])
headers["Cookie"] = "; ".join(f"{k}={v}" for k, v in cookies.items())
return {
"method": components["method"],
"url": final_url,
"headers": headers,
"body": body,
}
async def _replay_send(
client: Client,
*,
raw: bytes,
connection: ConnectionInfoInput,
) -> str:
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),
)
elapsed_ms = int((time.time() - started) * 1000)
response: dict[str, Any] | None = None
response_raw = result.entry.response_raw if hasattr(result.entry, "response_raw") else None
if response_raw is not None:
response = {
"raw": response_raw.decode("utf-8", errors="replace"),
}
return json.dumps(
{
"success": result.status == "DONE",
"status": result.status,
"error": result.error,
"session_id": str(session.id),
"elapsed_ms": elapsed_ms,
"response": response,
},
ensure_ascii=False,
default=str,
)
return _err("scope_rules", exc)
strix/tools/python/__init__.py
View File
strix/tools/python/tool.py
+307
View File
@@ -0,0 +1,307 @@
"""``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, "<python_action>", "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/<name>.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)
# /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,
)