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

refactor: Caido — replace ProxyManager with caido-sdk-client (host-side)

Browse Source 
Drop our 797-LoC manual GraphQL ``ProxyManager`` and the in-container
sandbox dispatch. Caido goes host-side via the official async Python
SDK. The Caido CLI still runs as a sidecar in the container — only the
control-plane moves.

Bootstrap moves host-side:
- New ``strix/sandbox/caido_bootstrap.py``: ``loginAsGuest`` via
  aiohttp (5 retries), then ``client.project.create(temporary=True)``
  + ``client.project.select(...)``, then return the connected
  ``caido_sdk_client.Client``. Drop the equivalent bash from
  ``docker-entrypoint.sh`` (~60 lines of curl + jq).
- ``entry.py`` calls ``bootstrap_caido_client`` after the
  ``wait_for_tcp_ready`` healthcheck, stashes the client in the bundle
  and threads it through ``make_agent_context(caido_client=...)``.
  ``agents_graph.create_agent`` propagates the same client to children.
- ``session_manager.cleanup`` ``await``s ``client.aclose()`` before
  tearing down the container.
- Drop ``CAIDO_PORT`` from the manifest env (only the in-container
  ProxyManager read it) and ``CAIDO_API_TOKEN`` from the entrypoint's
  ``/etc/profile.d/proxy.sh`` + ``/etc/environment`` heredocs.

Tools (``strix/tools/proxy/tools.py``):
- ``list_requests`` → ``client.request.list().filter().first().after()``
  with ascending/descending order. **Pagination changes from
  start_page/end_page (1-indexed) to first/after cursors** matching the
  SDK's native shape; response includes ``page_info.end_cursor`` for
  the model to thread.
- ``view_request`` → ``client.request.get(id, RequestGetOptions(...))``;
  decode raw bytes locally; existing regex-search and line-pagination
  modes preserved.
- ``send_request`` → synthesize raw HTTP bytes, parse URL into
  ``ConnectionInfoInput(host, port, is_tls)``, create a replay session
  via ``client.replay.sessions.create(CreateReplaySessionFromRaw(...))``,
  then ``client.replay.send(session_id, ReplaySendOptions(...))``.
- ``repeat_request`` → ``client.request.get(id, request_raw=True)`` →
  port the existing parse/_apply_modifications/build helpers verbatim →
  send via the same replay flow as ``send_request``.
- ``scope_rules`` → direct mapping to ``client.scope.{list, get, create,
  update, delete}``.
- **Drop ``list_sitemap`` + ``view_sitemap_entry``** — the official SDK
  has no sitemap module. The model uses HTTPQL filters
  (``req.host.eq:"X" AND req.path.cont:"/api/"``) for the same
  drill-down workflow.

Deletions:
- ``strix/tools/proxy/proxy_manager.py`` (797 LoC)
- ``strix/tools/proxy/proxy_actions.py`` (113 LoC)
- The 6-line proxy_actions pre-import in ``python_instance.py``
  (broken once proxy_actions is gone; that file is queued for deletion
  in commit 2 anyway).

Deps:
- Add ``caido-sdk-client>=0.2.0`` and ``aiohttp>=3.10.0`` to runtime
  ``[project] dependencies``.
- Drop ``gql[requests]>=3.5.3`` from ``[project.optional-dependencies]
  sandbox`` — only the in-container ProxyManager used the sync transport
  variant; the SDK pulls in ``gql[aiohttp]`` transitively for us.
- ``[[tool.mypy.overrides]]``: add ``caido_sdk_client.*`` and
  ``aiohttp.*`` to the missing-imports list with
  ``disable_error_code=["import-untyped"]`` (neither ships ``py.typed``).
- ``[tool.ruff.lint.per-file-ignores]``: bump the proxy/tools.py
  ignore to also include ``PLR0911`` (the scope_rules action dispatcher
  has many short-circuit returns).

ruff drops from 21 → 12 errors; mypy moves from 82 → 84 (the +2 are in
already-flaky files unrelated to this change). All touched files mypy
clean.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
0xallam
2026-04-25 14:23:56 -07:00
parent 9b31e9fd29
commit 5449af2456
14 changed files with 614 additions and 1173 deletions
Download Patch File Download Diff File Expand all files Collapse all files
containers/docker-entrypoint.sh
+1 -64
View File
@@ -47,68 +47,7 @@ fi
sleep 2
echo "Fetching API token..."
TOKEN=""
for attempt in 1 2 3 4 5; do
RESPONSE=$(curl -sL -X POST \
-H "Content-Type: application/json" \
-d '{"query":"mutation LoginAsGuest { loginAsGuest { token { accessToken } } }"}' \
http://localhost:${CAIDO_PORT}/graphql)
TOKEN=$(echo "$RESPONSE" | jq -r '.data.loginAsGuest.token.accessToken // empty')
if [ -n "$TOKEN" ] && [ "$TOKEN" != "null" ]; then
echo "Successfully obtained API token (attempt $attempt)."
break
fi
echo "Token fetch attempt $attempt failed: $RESPONSE"
sleep $((attempt * 2))
done
if [ -z "$TOKEN" ] || [ "$TOKEN" == "null" ]; then
echo "ERROR: Failed to get API token from Caido after 5 attempts."
echo "=== Caido log ==="
cat "$CAIDO_LOG" 2>/dev/null || echo "(no log available)"
exit 1
fi
export CAIDO_API_TOKEN=$TOKEN
echo "Caido API token has been set."
echo "Creating a new Caido project..."
CREATE_PROJECT_RESPONSE=$(curl -sL -X POST \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $TOKEN" \
-d '{"query":"mutation CreateProject { createProject(input: {name: \"sandbox\", temporary: true}) { project { id } } }"}' \
http://localhost:${CAIDO_PORT}/graphql)
PROJECT_ID=$(echo $CREATE_PROJECT_RESPONSE | jq -r '.data.createProject.project.id')
if [ -z "$PROJECT_ID" ] || [ "$PROJECT_ID" == "null" ]; then
echo "Failed to create Caido project."
echo "Response: $CREATE_PROJECT_RESPONSE"
exit 1
fi
echo "Caido project created with ID: $PROJECT_ID"
echo "Selecting Caido project..."
SELECT_RESPONSE=$(curl -sL -X POST \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $TOKEN" \
-d '{"query":"mutation SelectProject { selectProject(id: \"'$PROJECT_ID'\") { currentProject { project { id } } } }"}' \
http://localhost:${CAIDO_PORT}/graphql)
SELECTED_ID=$(echo $SELECT_RESPONSE | jq -r '.data.selectProject.currentProject.project.id')
if [ "$SELECTED_ID" != "$PROJECT_ID" ]; then
echo "Failed to select Caido project."
echo "Response: $SELECT_RESPONSE"
exit 1
fi
echo "✅ Caido project selected successfully."
echo "Caido is up — host bootstraps the guest token + project via the Python SDK."
echo "Configuring system-wide proxy settings..."
@@ -120,7 +59,6 @@ export HTTPS_PROXY=http://127.0.0.1:${CAIDO_PORT}
export ALL_PROXY=http://127.0.0.1:${CAIDO_PORT}
export REQUESTS_CA_BUNDLE=/etc/ssl/certs/ca-certificates.crt
export SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt
export CAIDO_API_TOKEN=${TOKEN}
EOF
cat << EOF | sudo tee /etc/environment
@@ -129,7 +67,6 @@ https_proxy=http://127.0.0.1:${CAIDO_PORT}
HTTP_PROXY=http://127.0.0.1:${CAIDO_PORT}
HTTPS_PROXY=http://127.0.0.1:${CAIDO_PORT}
ALL_PROXY=http://127.0.0.1:${CAIDO_PORT}
CAIDO_API_TOKEN=${TOKEN}
EOF
cat << EOF | sudo tee /etc/wgetrc
pyproject.toml
+6 -2
View File
@@ -41,6 +41,8 @@ dependencies = [
"requests>=2.32.0",
"cvss>=3.2",
"scrubadub>=2.0.1",
"caido-sdk-client>=0.2.0",
"aiohttp>=3.10.0",
]
[project.scripts]
@@ -53,7 +55,6 @@ sandbox = [
"ipython>=9.3.0",
"openhands-aci>=0.3.0",
"playwright>=1.48.0",
"gql[requests]>=3.5.3",
"libtmux>=0.46.2",
]
@@ -120,8 +121,11 @@ module = [
"cvss.*",
"scrubadub.*",
"docker.*",
"caido_sdk_client.*",
"aiohttp.*",
]
ignore_missing_imports = true
disable_error_code = ["import-untyped"]
# ============================================================================
# Ruff Configuration (Fast Python Linter & Formatter)
@@ -246,7 +250,7 @@ ignore = [
"strix/tools/browser/tool.py" = ["TC002"]
"strix/tools/terminal/tool.py" = ["TC002"]
"strix/tools/python/tool.py" = ["TC002"]
"strix/tools/proxy/tools.py" = ["TC002"]
"strix/tools/proxy/tools.py" = ["TC002", "PLR0911"]
"strix/tools/agents_graph/tools.py" = ["TC002"]
"strix/agents/factory.py" = ["TC002"]
# Entry point: ``Path`` is used at runtime by the typing of the
strix/agents/factory.py
-4
View File
@@ -50,12 +50,10 @@ from strix.tools.notes.tools import (
)
from strix.tools.proxy.tools import (
list_requests,
list_sitemap,
repeat_request,
scope_rules,
send_request,
view_request,
view_sitemap_entry,
)
from strix.tools.python.tool import python_action
from strix.tools.reporting.tool import create_vulnerability_report
@@ -112,8 +110,6 @@ _BASE_TOOLS: tuple[Tool, ...] = (
send_request,
repeat_request,
scope_rules,
list_sitemap,
view_sitemap_entry,
# Multi-agent graph tools (the bus is in ctx.context)
view_agent_graph,
agent_status,
strix/entry.py
+5
View File
@@ -34,6 +34,7 @@ from strix.run_config_factory import (
make_run_config,
)
from strix.sandbox import session_manager
from strix.sandbox.caido_bootstrap import bootstrap_caido_client
from strix.sandbox.healthcheck import wait_for_http_ready, wait_for_tcp_ready
@@ -221,6 +222,9 @@ async def run_strix_scan(
),
)
caido_client = await bootstrap_caido_client(int(bundle["caido_host_port"]))
bundle["caido_client"] = caido_client
try:
scan_mode = str(scan_config.get("scan_mode") or "deep")
is_whitebox = bool(scan_config.get("is_whitebox", False))
@@ -256,6 +260,7 @@ async def run_strix_scan(
sandbox_token=bundle["bearer"],
tool_server_host_port=bundle["tool_server_host_port"],
caido_host_port=bundle["caido_host_port"],
caido_client=caido_client,
agent_id=root_id,
parent_id=None,
tracer=tracer,
strix/run_config_factory.py
+2
View File
@@ -126,6 +126,7 @@ def make_agent_context(
run_id: str | None = None,
sandbox_client: Any | None = None,
agent_factory: Any | None = None,
caido_client: Any | None = None,
) -> dict[str, Any]:
"""Build the per-agent ``context`` dict passed to ``Runner.run(context=...)``.
@@ -145,6 +146,7 @@ def make_agent_context(
"sandbox_token": sandbox_token,
"tool_server_host_port": tool_server_host_port,
"caido_host_port": caido_host_port,
"caido_client": caido_client,
"agent_id": agent_id,
"parent_id": parent_id,
"tracer": tracer,
strix/sandbox/caido_bootstrap.py
+84
View File
@@ -0,0 +1,84 @@
"""Caido client bootstrap.
Caido CLI runs as an in-container sidecar. We connect from the host to
its mapped port, fetch a guest token (the CLI runs with
``--allow-guests``), then create + select a temporary project so the
SDK has a project context to operate on.
"""
from __future__ import annotations
import asyncio
import logging
from typing import Any
import aiohttp
from caido_sdk_client import Client, TokenAuthOptions
from caido_sdk_client.types import CreateProjectOptions
logger = logging.getLogger(__name__)
_LOGIN_AS_GUEST_QUERY = "mutation LoginAsGuest { loginAsGuest { token { accessToken } } }"
async def _login_as_guest(url: str, *, attempts: int = 5) -> str:
"""POST ``loginAsGuest`` mutation; return the access token.
Retries up to ``attempts`` times with exponential-ish backoff, mirroring
what the legacy bash entrypoint did. The Caido sidecar may not be ready
on the first poke even after its TCP port accepts connections.
"""
last_err: Exception | None = None
async with aiohttp.ClientSession() as session:
for i in range(1, attempts + 1):
try:
async with session.post(
f"{url}/graphql",
json={"query": _LOGIN_AS_GUEST_QUERY},
headers={"Content-Type": "application/json"},
timeout=aiohttp.ClientTimeout(total=15),
) as response:
response.raise_for_status()
payload: dict[str, Any] = await response.json()
token = (
payload.get("data", {})
.get("loginAsGuest", {})
.get("token", {})
.get("accessToken")
)
if token:
return str(token)
last_err = RuntimeError(f"loginAsGuest returned no token: {payload}")
except (aiohttp.ClientError, TimeoutError, RuntimeError) as exc:
last_err = exc
logger.debug("loginAsGuest attempt %d/%d failed: %s", i, attempts, exc)
await asyncio.sleep(min(2.0 * i, 8.0))
raise RuntimeError(f"loginAsGuest failed after {attempts} attempts: {last_err}")
async def bootstrap_caido_client(host_port: int) -> Client:
"""Connect to the in-container Caido sidecar and select a fresh project.
Args:
host_port: Resolved host port that maps to the container's Caido
GraphQL listener.
Returns:
A connected :class:`caido_sdk_client.Client` ready to use.
"""
url = f"http://127.0.0.1:{host_port}"
logger.info("Bootstrapping Caido client at %s", url)
access_token = await _login_as_guest(url)
client = Client(url, auth=TokenAuthOptions(token=access_token))
await client.connect()
project = await client.project.create(
CreateProjectOptions(name="sandbox", temporary=True),
)
await client.project.select(project.id)
logger.info("Caido project selected: %s", project.id)
return client
strix/sandbox/session_manager.py
+7 -1
View File
@@ -105,7 +105,6 @@ async def create_or_reuse(
value={
"TOOL_SERVER_TOKEN": bearer,
"TOOL_SERVER_PORT": str(_CONTAINER_TOOL_SERVER_PORT),
"CAIDO_PORT": str(_CONTAINER_CAIDO_PORT),
"STRIX_SANDBOX_EXECUTION_TIMEOUT": str(execution_timeout),
"PYTHONUNBUFFERED": "1",
"HOST_GATEWAY": "host.docker.internal",
@@ -162,6 +161,13 @@ async def cleanup(scan_id: str) -> None:
logger.debug("cleanup(%s): no cached session", scan_id)
return
caido_client = bundle.get("caido_client")
if caido_client is not None:
try:
await caido_client.aclose()
except Exception: # noqa: BLE001
logger.debug("cleanup(%s): caido_client.aclose() raised", scan_id, exc_info=True)
try:
await bundle["client"].delete(bundle["session"])
logger.info("Cleaned up sandbox session for scan %s", scan_id)
strix/tools/agents_graph/tools.py
+1
View File
@@ -406,6 +406,7 @@ async def create_agent(
sandbox_token=inner.get("sandbox_token"),
tool_server_host_port=inner.get("tool_server_host_port"),
caido_host_port=inner.get("caido_host_port"),
caido_client=inner.get("caido_client"),
agent_id=child_id,
parent_id=parent_id,
tracer=inner.get("tracer"),
strix/tools/proxy/__init__.py
+1 -11
View File
@@ -1,20 +1,10 @@
from .proxy_actions import (
list_requests,
list_sitemap,
repeat_request,
scope_rules,
send_request,
view_request,
view_sitemap_entry,
)
from .tools import list_requests, repeat_request, scope_rules, send_request, view_request
__all__ = [
"list_requests",
"list_sitemap",
"repeat_request",
"scope_rules",
"send_request",
"view_request",
"view_sitemap_entry",
]
strix/tools/proxy/proxy_actions.py
-113
View File
@@ -1,113 +0,0 @@
from typing import Any, Literal
from strix.tools.registry import register_tool
RequestPart = Literal["request", "response"]
@register_tool
def list_requests(
httpql_filter: str | None = None,
start_page: int = 1,
end_page: int = 1,
page_size: int = 50,
sort_by: Literal[
"timestamp",
"host",
"method",
"path",
"status_code",
"response_time",
"response_size",
"source",
] = "timestamp",
sort_order: Literal["asc", "desc"] = "desc",
scope_id: str | None = None,
) -> dict[str, Any]:
from .proxy_manager import get_proxy_manager
manager = get_proxy_manager()
return manager.list_requests(
httpql_filter, start_page, end_page, page_size, sort_by, sort_order, scope_id
)
@register_tool
def view_request(
request_id: str,
part: RequestPart = "request",
search_pattern: str | None = None,
page: int = 1,
page_size: int = 50,
) -> dict[str, Any]:
from .proxy_manager import get_proxy_manager
manager = get_proxy_manager()
return manager.view_request(request_id, part, search_pattern, page, page_size)
@register_tool
def send_request(
method: str,
url: str,
headers: dict[str, str] | None = None,
body: str = "",
timeout: int = 30,
) -> dict[str, Any]:
from .proxy_manager import get_proxy_manager
if headers is None:
headers = {}
manager = get_proxy_manager()
return manager.send_simple_request(method, url, headers, body, timeout)
@register_tool
def repeat_request(
request_id: str,
modifications: dict[str, Any] | None = None,
) -> dict[str, Any]:
from .proxy_manager import get_proxy_manager
if modifications is None:
modifications = {}
manager = get_proxy_manager()
return manager.repeat_request(request_id, modifications)
@register_tool
def scope_rules(
action: Literal["get", "list", "create", "update", "delete"],
allowlist: list[str] | None = None,
denylist: list[str] | None = None,
scope_id: str | None = None,
scope_name: str | None = None,
) -> dict[str, Any]:
from .proxy_manager import get_proxy_manager
manager = get_proxy_manager()
return manager.scope_rules(action, allowlist, denylist, scope_id, scope_name)
@register_tool
def list_sitemap(
scope_id: str | None = None,
parent_id: str | None = None,
depth: Literal["DIRECT", "ALL"] = "DIRECT",
page: int = 1,
) -> dict[str, Any]:
from .proxy_manager import get_proxy_manager
manager = get_proxy_manager()
return manager.list_sitemap(scope_id, parent_id, depth, page)
@register_tool
def view_sitemap_entry(
entry_id: str,
) -> dict[str, Any]:
from .proxy_manager import get_proxy_manager
manager = get_proxy_manager()
return manager.view_sitemap_entry(entry_id)
strix/tools/proxy/proxy_manager.py
-797
View File
@@ -1,797 +0,0 @@
import base64
import os
import re
import time
from typing import TYPE_CHECKING, Any
from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
import requests
from gql import Client, gql
from gql.transport.exceptions import TransportQueryError
from gql.transport.requests import RequestsHTTPTransport
from requests.exceptions import ProxyError, RequestException, Timeout
if TYPE_CHECKING:
from collections.abc import Callable
CAIDO_PORT = 48080 # Fixed port inside container
class ProxyManager:
def __init__(self, auth_token: str | None = None):
host = "127.0.0.1"
self.base_url = f"http://{host}:{CAIDO_PORT}/graphql"
self.proxies = {
"http": f"http://{host}:{CAIDO_PORT}",
"https": f"http://{host}:{CAIDO_PORT}",
}
self.auth_token = auth_token or os.getenv("CAIDO_API_TOKEN")
def _get_client(self) -> Client:
transport = RequestsHTTPTransport(
url=self.base_url, headers={"Authorization": f"Bearer {self.auth_token}"}
)
return Client(transport=transport, fetch_schema_from_transport=False)
def list_requests(
self,
httpql_filter: str | None = None,
start_page: int = 1,
end_page: int = 1,
page_size: int = 50,
sort_by: str = "timestamp",
sort_order: str = "desc",
scope_id: str | None = None,
) -> dict[str, Any]:
offset = (start_page - 1) * page_size
limit = (end_page - start_page + 1) * page_size
sort_mapping = {
"timestamp": "CREATED_AT",
"host": "HOST",
"method": "METHOD",
"path": "PATH",
"status_code": "RESP_STATUS_CODE",
"response_time": "RESP_ROUNDTRIP_TIME",
"response_size": "RESP_LENGTH",
"source": "SOURCE",
}
query = gql("""
query GetRequests(
$limit: Int, $offset: Int, $filter: HTTPQL,
$order: RequestResponseOrderInput, $scopeId: ID
) {
requestsByOffset(
limit: $limit, offset: $offset, filter: $filter,
order: $order, scopeId: $scopeId
) {
edges {
node {
id method host path query createdAt length isTls port
source alteration fileExtension
response { id statusCode length roundtripTime createdAt }
}
}
count { value }
}
}
""")
variables = {
"limit": limit,
"offset": offset,
"filter": httpql_filter,
"order": {
"by": sort_mapping.get(sort_by, "CREATED_AT"),
"ordering": sort_order.upper(),
},
"scopeId": scope_id,
}
try:
result = self._get_client().execute(query, variable_values=variables)
data = result.get("requestsByOffset", {})
nodes = [edge["node"] for edge in data.get("edges", [])]
count_data = data.get("count") or {}
return {
"requests": nodes,
"total_count": count_data.get("value", 0),
"start_page": start_page,
"end_page": end_page,
"page_size": page_size,
"offset": offset,
"returned_count": len(nodes),
"sort_by": sort_by,
"sort_order": sort_order,
}
except (TransportQueryError, ValueError, KeyError) as e:
return {"requests": [], "total_count": 0, "error": f"Error fetching requests: {e}"}
def view_request(
self,
request_id: str,
part: str = "request",
search_pattern: str | None = None,
page: int = 1,
page_size: int = 50,
) -> dict[str, Any]:
queries = {
"request": """query GetRequest($id: ID!) {
request(id: $id) {
id method host path query createdAt length isTls port
source alteration edited raw
}
}""",
"response": """query GetRequest($id: ID!) {
request(id: $id) {
id response {
id statusCode length roundtripTime createdAt raw
}
}
}""",
}
if part not in queries:
return {"error": f"Invalid part '{part}'. Use 'request' or 'response'"}
try:
result = self._get_client().execute(
gql(queries[part]), variable_values={"id": request_id}
)
request_data = result.get("request", {})
if not request_data:
return {"error": f"Request {request_id} not found"}
if part == "request":
raw_content = request_data.get("raw")
else:
response_data = request_data.get("response") or {}
raw_content = response_data.get("raw")
if not raw_content:
return {"error": "No content available"}
content = base64.b64decode(raw_content).decode("utf-8", errors="replace")
if part == "response":
request_data["response"]["raw"] = content
else:
request_data["raw"] = content
return (
self._search_content(request_data, content, search_pattern)
if search_pattern
else self._paginate_content(request_data, content, page, page_size)
)
except (TransportQueryError, ValueError, KeyError, UnicodeDecodeError) as e:
return {"error": f"Failed to view request: {e}"}
def _search_content(
self, request_data: dict[str, Any], content: str, pattern: str
) -> dict[str, Any]:
try:
regex = re.compile(pattern, re.IGNORECASE | re.MULTILINE | re.DOTALL)
matches = []
for match in regex.finditer(content):
start, end = match.start(), match.end()
context_size = 120
before = re.sub(r"\s+", " ", content[max(0, start - context_size) : start].strip())[
-100:
]
after = re.sub(r"\s+", " ", content[end : end + context_size].strip())[:100]
matches.append(
{"match": match.group(), "before": before, "after": after, "position": start}
)
if len(matches) >= 20:
break
return {
"id": request_data.get("id"),
"matches": matches,
"total_matches": len(matches),
"search_pattern": pattern,
"truncated": len(matches) >= 20,
}
except re.error as e:
return {"error": f"Invalid regex: {e}"}
def _paginate_content(
self, request_data: dict[str, Any], content: str, page: int, page_size: int
) -> dict[str, Any]:
display_lines = []
for line in content.split("\n"):
if len(line) <= 80:
display_lines.append(line)
else:
display_lines.extend(
[
line[i : i + 80] + (" \\" if i + 80 < len(line) else "")
for i in range(0, len(line), 80)
]
)
total_lines = len(display_lines)
total_pages = (total_lines + page_size - 1) // page_size
page = max(1, min(page, total_pages))
start_line = (page - 1) * page_size
end_line = min(total_lines, start_line + page_size)
return {
"id": request_data.get("id"),
"content": "\n".join(display_lines[start_line:end_line]),
"page": page,
"total_pages": total_pages,
"showing_lines": f"{start_line + 1}-{end_line} of {total_lines}",
"has_more": page < total_pages,
}
def send_simple_request(
self,
method: str,
url: str,
headers: dict[str, str] | None = None,
body: str = "",
timeout: int = 30,
) -> dict[str, Any]:
if headers is None:
headers = {}
try:
start_time = time.time()
response = requests.request(
method=method,
url=url,
headers=headers,
data=body or None,
proxies=self.proxies,
timeout=timeout,
verify=False,
)
response_time = int((time.time() - start_time) * 1000)
body_content = response.text
if len(body_content) > 10000:
body_content = body_content[:10000] + "\n... [truncated]"
return {
"status_code": response.status_code,
"headers": dict(response.headers),
"body": body_content,
"response_time_ms": response_time,
"url": response.url,
"message": (
"Request sent through proxy - check list_requests() for captured traffic"
),
}
except (RequestException, ProxyError, Timeout) as e:
return {"error": f"Request failed: {type(e).__name__}", "details": str(e), "url": url}
def repeat_request(
self, request_id: str, modifications: dict[str, Any] | None = None
) -> dict[str, Any]:
if modifications is None:
modifications = {}
original = self.view_request(request_id, "request")
if "error" in original:
return {"error": f"Could not retrieve original request: {original['error']}"}
raw_content = original.get("content", "")
if not raw_content:
return {"error": "No raw request content found"}
request_components = self._parse_http_request(raw_content)
if "error" in request_components:
return request_components
full_url = self._build_full_url(request_components, modifications)
if "error" in full_url:
return full_url
modified_request = self._apply_modifications(
request_components, modifications, full_url["url"]
)
return self._send_modified_request(modified_request, request_id, modifications)
def _parse_http_request(self, raw_content: str) -> dict[str, Any]:
lines = raw_content.split("\n")
request_line = lines[0].strip().split(" ")
if len(request_line) < 2:
return {"error": "Invalid request line format"}
method, url_path = request_line[0], request_line[1]
headers = {}
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)
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": headers, "body": body}
def _build_full_url(
self, components: dict[str, Any], modifications: dict[str, Any]
) -> dict[str, Any]:
headers = components["headers"]
host = headers.get("Host", "")
if not host:
return {"error": "No Host header found"}
protocol = (
"https" if ":443" in host or "https" in headers.get("Referer", "").lower() else "http"
)
full_url = f"{protocol}://{host}{components['url_path']}"
if "url" in modifications:
full_url = modifications["url"]
return {"url": full_url}
def _apply_modifications(
self, components: dict[str, Any], modifications: dict[str, Any], full_url: str
) -> dict[str, Any]:
headers = components["headers"].copy()
body = components["body"]
final_url = full_url
if "params" in modifications:
parsed = urlparse(final_url)
params = {k: v[0] if v else "" for k, v in parse_qs(parsed.query).items()}
params.update(modifications["params"])
final_url = urlunparse(parsed._replace(query=urlencode(params)))
if "headers" in modifications:
headers.update(modifications["headers"])
if "body" in modifications:
body = modifications["body"]
if "cookies" in modifications:
cookies = {}
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,
}
def _send_modified_request(
self, request_data: dict[str, Any], request_id: str, modifications: dict[str, Any]
) -> dict[str, Any]:
try:
start_time = time.time()
response = requests.request(
method=request_data["method"],
url=request_data["url"],
headers=request_data["headers"],
data=request_data["body"] or None,
proxies=self.proxies,
timeout=30,
verify=False,
)
response_time = int((time.time() - start_time) * 1000)
response_body = response.text
truncated = len(response_body) > 10000
if truncated:
response_body = response_body[:10000] + "\n... [truncated]"
return {
"status_code": response.status_code,
"status_text": response.reason,
"headers": {
k: v
for k, v in response.headers.items()
if k.lower()
in ["content-type", "content-length", "server", "set-cookie", "location"]
},
"body": response_body,
"body_truncated": truncated,
"body_size": len(response.content),
"response_time_ms": response_time,
"url": response.url,
"original_request_id": request_id,
"modifications_applied": modifications,
"request": {
"method": request_data["method"],
"url": request_data["url"],
"headers": request_data["headers"],
"has_body": bool(request_data["body"]),
},
}
except ProxyError as e:
return {
"error": "Proxy connection failed - is Caido running?",
"details": str(e),
"original_request_id": request_id,
}
except (RequestException, Timeout) as e:
return {
"error": f"Failed to repeat request: {type(e).__name__}",
"details": str(e),
"original_request_id": request_id,
}
def _handle_scope_list(self) -> dict[str, Any]:
result = self._get_client().execute(
gql("query { scopes { id name allowlist denylist indexed } }")
)
scopes = result.get("scopes", [])
return {"scopes": scopes, "count": len(scopes)}
def _handle_scope_get(self, scope_id: str | None) -> dict[str, Any]:
if not scope_id:
return self._handle_scope_list()
result = self._get_client().execute(
gql(
"query GetScope($id: ID!) { scope(id: $id) { id name allowlist denylist indexed } }"
),
variable_values={"id": scope_id},
)
scope = result.get("scope")
if not scope:
return {"error": f"Scope {scope_id} not found"}
return {"scope": scope}
def _handle_scope_create(
self, scope_name: str, allowlist: list[str] | None, denylist: list[str] | None
) -> dict[str, Any]:
if not scope_name:
return {"error": "scope_name required for create"}
mutation = gql("""
mutation CreateScope($input: CreateScopeInput!) {
createScope(input: $input) {
scope { id name allowlist denylist indexed }
error {
... on InvalidGlobTermsUserError { code terms }
... on OtherUserError { code }
}
}
}
""")
result = self._get_client().execute(
mutation,
variable_values={
"input": {
"name": scope_name,
"allowlist": allowlist or [],
"denylist": denylist or [],
}
},
)
payload = result.get("createScope", {})
if payload.get("error"):
error = payload["error"]
return {"error": f"Invalid glob patterns: {error.get('terms', error.get('code'))}"}
return {"scope": payload.get("scope"), "message": "Scope created successfully"}
def _handle_scope_update(
self,
scope_id: str,
scope_name: str,
allowlist: list[str] | None,
denylist: list[str] | None,
) -> dict[str, Any]:
if not scope_id or not scope_name:
return {"error": "scope_id and scope_name required"}
mutation = gql("""
mutation UpdateScope($id: ID!, $input: UpdateScopeInput!) {
updateScope(id: $id, input: $input) {
scope { id name allowlist denylist indexed }
error {
... on InvalidGlobTermsUserError { code terms }
... on OtherUserError { code }
}
}
}
""")
result = self._get_client().execute(
mutation,
variable_values={
"id": scope_id,
"input": {
"name": scope_name,
"allowlist": allowlist or [],
"denylist": denylist or [],
},
},
)
payload = result.get("updateScope", {})
if payload.get("error"):
error = payload["error"]
return {"error": f"Invalid glob patterns: {error.get('terms', error.get('code'))}"}
return {"scope": payload.get("scope"), "message": "Scope updated successfully"}
def _handle_scope_delete(self, scope_id: str) -> dict[str, Any]:
if not scope_id:
return {"error": "scope_id required for delete"}
result = self._get_client().execute(
gql("mutation DeleteScope($id: ID!) { deleteScope(id: $id) { deletedId } }"),
variable_values={"id": scope_id},
)
payload = result.get("deleteScope", {})
if not payload.get("deletedId"):
return {"error": f"Failed to delete scope {scope_id}"}
return {"message": f"Scope {scope_id} deleted", "deletedId": payload["deletedId"]}
def scope_rules(
self,
action: str,
allowlist: list[str] | None = None,
denylist: list[str] | None = None,
scope_id: str | None = None,
scope_name: str | None = None,
) -> dict[str, Any]:
handlers: dict[str, Callable[[], dict[str, Any]]] = {
"list": self._handle_scope_list,
"get": lambda: self._handle_scope_get(scope_id),
"create": lambda: (
{"error": "scope_name required for create"}
if not scope_name
else self._handle_scope_create(scope_name, allowlist, denylist)
),
"update": lambda: (
{"error": "scope_id and scope_name required"}
if not scope_id or not scope_name
else self._handle_scope_update(scope_id, scope_name, allowlist, denylist)
),
"delete": lambda: (
{"error": "scope_id required for delete"}
if not scope_id
else self._handle_scope_delete(scope_id)
),
}
handler = handlers.get(action)
if not handler:
return {
"error": f"Unsupported action: {action}. Use 'get', 'list', 'create', "
f"'update', or 'delete'"
}
try:
result = handler()
except (TransportQueryError, ValueError, KeyError) as e:
return {"error": f"Scope operation failed: {e}"}
else:
return result
def list_sitemap(
self,
scope_id: str | None = None,
parent_id: str | None = None,
depth: str = "DIRECT",
page: int = 1,
page_size: int = 30,
) -> dict[str, Any]:
try:
skip_count = (page - 1) * page_size
if parent_id:
query = gql("""
query GetSitemapDescendants($parentId: ID!, $depth: SitemapDescendantsDepth!) {
sitemapDescendantEntries(parentId: $parentId, depth: $depth) {
edges {
node {
id kind label hasDescendants
request { method path response { statusCode } }
}
}
count { value }
}
}
""")
result = self._get_client().execute(
query, variable_values={"parentId": parent_id, "depth": depth}
)
data = result.get("sitemapDescendantEntries", {})
else:
query = gql("""
query GetSitemapRoots($scopeId: ID) {
sitemapRootEntries(scopeId: $scopeId) {
edges { node {
id kind label hasDescendants
metadata { ... on SitemapEntryMetadataDomain { isTls port } }
request { method path response { statusCode } }
} }
count { value }
}
}
""")
result = self._get_client().execute(query, variable_values={"scopeId": scope_id})
data = result.get("sitemapRootEntries", {})
all_nodes = [edge["node"] for edge in data.get("edges", [])]
count_data = data.get("count") or {}
total_count = count_data.get("value", 0)
paginated_nodes = all_nodes[skip_count : skip_count + page_size]
cleaned_nodes = []
for node in paginated_nodes:
cleaned = {
"id": node["id"],
"kind": node["kind"],
"label": node["label"],
"hasDescendants": node["hasDescendants"],
}
if node.get("metadata") and (
node["metadata"].get("isTls") is not None or node["metadata"].get("port")
):
cleaned["metadata"] = node["metadata"]
if node.get("request"):
req = node["request"]
cleaned_req = {}
if req.get("method"):
cleaned_req["method"] = req["method"]
if req.get("path"):
cleaned_req["path"] = req["path"]
response_data = req.get("response") or {}
if response_data.get("statusCode"):
cleaned_req["status"] = response_data["statusCode"]
if cleaned_req:
cleaned["request"] = cleaned_req
cleaned_nodes.append(cleaned)
total_pages = (total_count + page_size - 1) // page_size
return {
"entries": cleaned_nodes,
"page": page,
"page_size": page_size,
"total_pages": total_pages,
"total_count": total_count,
"has_more": page < total_pages,
"showing": (
f"{skip_count + 1}-{min(skip_count + page_size, total_count)} of {total_count}"
),
}
except (TransportQueryError, ValueError, KeyError) as e:
return {"error": f"Failed to fetch sitemap: {e}"}
def _process_sitemap_metadata(self, node: dict[str, Any]) -> dict[str, Any]:
cleaned = {
"id": node["id"],
"kind": node["kind"],
"label": node["label"],
"hasDescendants": node["hasDescendants"],
}
if node.get("metadata") and (
node["metadata"].get("isTls") is not None or node["metadata"].get("port")
):
cleaned["metadata"] = node["metadata"]
return cleaned
def _process_sitemap_request(self, req: dict[str, Any]) -> dict[str, Any] | None:
cleaned_req = {}
if req.get("method"):
cleaned_req["method"] = req["method"]
if req.get("path"):
cleaned_req["path"] = req["path"]
response_data = req.get("response") or {}
if response_data.get("statusCode"):
cleaned_req["status"] = response_data["statusCode"]
return cleaned_req if cleaned_req else None
def _process_sitemap_response(self, resp: dict[str, Any]) -> dict[str, Any]:
cleaned_resp = {}
if resp.get("statusCode"):
cleaned_resp["status"] = resp["statusCode"]
if resp.get("length"):
cleaned_resp["size"] = resp["length"]
if resp.get("roundtripTime"):
cleaned_resp["time_ms"] = resp["roundtripTime"]
return cleaned_resp
def view_sitemap_entry(self, entry_id: str) -> dict[str, Any]:
try:
query = gql("""
query GetSitemapEntry($id: ID!) {
sitemapEntry(id: $id) {
id kind label hasDescendants
metadata { ... on SitemapEntryMetadataDomain { isTls port } }
request { method path response { statusCode length roundtripTime } }
requests(first: 30, order: {by: CREATED_AT, ordering: DESC}) {
edges { node { method path response { statusCode length } } }
count { value }
}
}
}
""")
result = self._get_client().execute(query, variable_values={"id": entry_id})
entry = result.get("sitemapEntry")
if not entry:
return {"error": f"Sitemap entry {entry_id} not found"}
cleaned = self._process_sitemap_metadata(entry)
if entry.get("request"):
req = entry["request"]
cleaned_req = {}
if req.get("method"):
cleaned_req["method"] = req["method"]
if req.get("path"):
cleaned_req["path"] = req["path"]
if req.get("response"):
cleaned_req["response"] = self._process_sitemap_response(req["response"])
if cleaned_req:
cleaned["request"] = cleaned_req
requests_data = entry.get("requests", {})
request_nodes = [edge["node"] for edge in requests_data.get("edges", [])]
cleaned_requests = [
req
for req in (self._process_sitemap_request(node) for node in request_nodes)
if req is not None
]
count_data = requests_data.get("count") or {}
cleaned["related_requests"] = {
"requests": cleaned_requests,
"total_count": count_data.get("value", 0),
"showing": f"Latest {len(cleaned_requests)} requests",
}
return {"entry": cleaned} if cleaned else {"error": "Failed to process sitemap entry"} # noqa: TRY300
except (TransportQueryError, ValueError, KeyError) as e:
return {"error": f"Failed to fetch sitemap entry: {e}"}
def close(self) -> None:
pass
_PROXY_MANAGER: ProxyManager | None = None
def get_proxy_manager() -> ProxyManager:
global _PROXY_MANAGER # noqa: PLW0603
if _PROXY_MANAGER is None:
_PROXY_MANAGER = ProxyManager()
return _PROXY_MANAGER
strix/tools/proxy/tools.py
+472 -143
View File
@@ -1,21 +1,40 @@
"""SDK function-tool wrappers for the seven Caido proxy tools.
"""Caido proxy tools — host-side via ``caido-sdk-client``.
All seven dispatch to the in-container Caido manager via the sandbox
tool server. Same pattern as browser/terminal/python — host wrapper is
pure pass-through, no logic of its own.
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.
Tools: list_requests, view_request, send_request, repeat_request,
scope_rules, list_sitemap, view_sitemap_entry.
Tools: ``list_requests``, ``view_request``, ``send_request``,
``repeat_request``, ``scope_rules``.
"""
from __future__ import annotations
from typing import Any, Literal
import dataclasses
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
from caido_sdk_client.types import (
ConnectionInfoInput,
CreateReplaySessionFromRaw,
CreateReplaySessionOptions,
CreateScopeOptions,
ReplaySendOptions,
RequestGetOptions,
UpdateScopeOptions,
)
from strix.tools._decorator import dump_tool_result, strix_tool
from strix.tools._sandbox_dispatch import post_to_sandbox
if TYPE_CHECKING:
from caido_sdk_client import Client
RequestPart = Literal["request", "response"]
@@ -30,17 +49,66 @@ SortBy = Literal[
"source",
]
SortOrder = Literal["asc", "desc"]
SitemapDepth = Literal["DIRECT", "ALL"]
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")
def _serialize(value: Any) -> Any:
"""Recursively convert SDK dataclasses/Pydantic objects to JSON-safe primitives."""
if value is None or isinstance(value, str | int | float | bool):
return value
if isinstance(value, bytes):
try:
return value.decode("utf-8", errors="replace")
except Exception: # noqa: BLE001
return value.hex()
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()}
if hasattr(value, "model_dump"):
return _serialize(value.model_dump())
if isinstance(value, dict):
return {str(k): _serialize(v) for k, v in value.items()}
if isinstance(value, list | tuple | set):
return [_serialize(v) for v in value]
return str(value)
def _no_client() -> str:
return dump_tool_result(
{
"success": False,
"error": "Caido client not initialized in context.",
},
)
# ----------------------------------------------------------------------
# list_requests
# ----------------------------------------------------------------------
@strix_tool(timeout=120)
async def list_requests(
ctx: RunContextWrapper,
httpql_filter: str | None = None,
start_page: int = 1,
end_page: int = 1,
page_size: int = 50,
first: int = 50,
after: str | None = None,
sort_by: SortBy = "timestamp",
sort_order: SortOrder = "desc",
scope_id: str | None = None,
@@ -64,34 +132,94 @@ async def list_requests(
- **Special**: ``source:intercept`` (only intercepted requests),
``preset:"name"``.
For sitemap-style tree traversal use HTTPQL filters: drill into a
host with ``req.host.eq:"example.com"`` then narrow paths with
``req.path.cont:"/api/"``.
Pagination is cursor-based. Pass the ``end_cursor`` from the
``page_info`` of one call as ``after`` to the next.
Args:
httpql_filter: Caido HTTPQL query.
start_page: Starting page, 1-indexed.
end_page: Ending page (inclusive).
page_size: Entries per page (default 50).
sort_by: ``timestamp`` / ``host`` / ``method`` / ``path`` /
``status_code`` / ``response_time`` / ``response_size`` /
``source``.
httpql_filter: Caido HTTPQL query (optional).
first: Number of entries to return (default 50).
after: Cursor from a previous response's ``page_info.end_cursor``.
sort_by: One of ``timestamp`` / ``host`` / ``method`` / ``path``
/ ``status_code`` / ``response_time`` / ``response_size``
/ ``source``.
sort_order: ``asc`` or ``desc``.
scope_id: Restrict to a scope (managed via ``scope_rules``).
scope_id: Restrict to a Caido scope (managed via ``scope_rules``).
"""
return dump_tool_result(
await post_to_sandbox(
ctx,
"list_requests",
client = _ctx_client(ctx)
if client is None:
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()
entries = []
for edge in connection.edges:
req = edge.node.request
resp = edge.node.response
entries.append(
{
"cursor": edge.cursor,
"request": {
"id": req.id,
"host": req.host,
"port": req.port,
"method": req.method,
"path": req.path,
"query": req.query,
"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
),
},
)
return dump_tool_result(
{
"httpql_filter": httpql_filter,
"start_page": start_page,
"end_page": end_page,
"page_size": page_size,
"sort_by": sort_by,
"sort_order": sort_order,
"scope_id": scope_id,
"success": True,
"entries": entries,
"page_info": {
"has_next_page": connection.page_info.has_next_page,
"has_previous_page": connection.page_info.has_previous_page,
"start_cursor": connection.page_info.start_cursor,
"end_cursor": connection.page_info.end_cursor,
},
},
),
)
)
except Exception as exc: # noqa: BLE001
return dump_tool_result({"success": False, "error": f"list_requests failed: {exc}"})
# ----------------------------------------------------------------------
# view_request
# ----------------------------------------------------------------------
@strix_tool(timeout=60)
async def view_request(
ctx: RunContextWrapper,
@@ -108,8 +236,8 @@ async def view_request(
- **With** ``search_pattern`` (compact regex hits) — returns up to 20
matches with ``before`` / ``after`` context and position. Useful
for hunting reflected input, leaked URLs, hidden parameters.
- **Without** ``search_pattern`` (full content with pagination) —
returns the page of raw content plus ``has_more`` flag.
- **Without** ``search_pattern`` (full content with line pagination)
— returns the page of raw content plus ``has_more`` flag.
Common search patterns:
@@ -126,24 +254,85 @@ async def view_request(
page: 1-indexed page number (only when no ``search_pattern``).
page_size: Lines per page.
"""
return dump_tool_result(
await post_to_sandbox(
ctx,
"view_request",
client = _ctx_client(ctx)
if client is None:
return _no_client()
try:
opts = RequestGetOptions(
request_raw=(part == "request"),
response_raw=(part == "response"),
)
result = await client.request.get(request_id, opts)
if result is None:
return dump_tool_result(
{"success": False, "error": f"Request {request_id} not found"},
)
raw_bytes = (
result.request.raw
if part == "request"
else (result.response.raw if result.response is not None else None)
)
if raw_bytes is None:
return dump_tool_result(
{
"success": False,
"error": f"No raw {part} for {request_id}",
},
)
content = raw_bytes.decode("utf-8", errors="replace")
if search_pattern:
return dump_tool_result(_regex_hits(content, search_pattern))
return dump_tool_result(_paginate_lines(content, page=page, page_size=page_size))
except Exception as exc: # noqa: BLE001
return dump_tool_result({"success": False, "error": f"view_request failed: {exc}"})
def _regex_hits(content: str, pattern: str) -> dict[str, Any]:
try:
regex = re.compile(pattern)
except re.error as exc:
return {"success": False, "error": f"Invalid regex: {exc}"}
hits = []
for match in regex.finditer(content):
start, end = match.span()
before = content[max(0, start - 40) : start]
after = content[end : end + 40]
hits.append(
{
"request_id": request_id,
"part": part,
"search_pattern": search_pattern,
"page": page,
"page_size": page_size,
"match": match.group(0),
"position": start,
"before": before,
"after": after,
},
),
)
)
if len(hits) >= 20:
break
return {"success": True, "hits": hits, "total_hits": len(hits)}
# strict_mode=False because ``headers`` is a free-form dict — the model
# can't enumerate all possible HTTP headers, and the SDK's strict JSON
# schema rejects ``additionalProperties: true``.
def _paginate_lines(content: str, *, page: int, page_size: int) -> dict[str, Any]:
lines = content.splitlines()
start = max(0, (page - 1) * page_size)
end = start + page_size
return {
"success": True,
"content": "\n".join(lines[start:end]),
"page": page,
"page_size": page_size,
"total_lines": len(lines),
"has_more": end < len(lines),
}
# ----------------------------------------------------------------------
# send_request
# ----------------------------------------------------------------------
@strix_tool(timeout=120, strict_mode=False)
async def send_request(
ctx: RunContextWrapper,
@@ -167,24 +356,23 @@ async def send_request(
body: Optional request body string.
timeout: Per-request timeout in seconds (default 30).
"""
return dump_tool_result(
await post_to_sandbox(
ctx,
"send_request",
{
"method": method,
"url": url,
"headers": headers or {},
"body": body,
"timeout": timeout,
},
),
)
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 = _build_raw_request(
method=method, url=url, headers=headers or {}, body=body
)
return await _replay_send(client, raw=raw, connection=connection)
except Exception as exc: # noqa: BLE001
return dump_tool_result({"success": False, "error": f"send_request failed: {exc}"})
# strict_mode=False because ``modifications`` is a free-form patch dict
# (header overrides, body replacements, query-string tweaks) the model
# composes per-call.
# ----------------------------------------------------------------------
# repeat_request
# ----------------------------------------------------------------------
@strix_tool(timeout=120, strict_mode=False)
async def repeat_request(
ctx: RunContextWrapper,
@@ -216,18 +404,37 @@ async def repeat_request(
- ``body`` — replace the body string entirely.
- ``cookies`` — dict of cookies to add/update.
"""
return dump_tool_result(
await post_to_sandbox(
ctx,
"repeat_request",
{
"request_id": request_id,
"modifications": modifications or {},
},
),
)
client = _ctx_client(ctx)
if client is None:
return _no_client()
mods = modifications or {}
try:
result = await client.request.get(request_id, RequestGetOptions(request_raw=True))
if result is None or result.request.raw is None:
return dump_tool_result(
{"success": False, "error": 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(client, raw=raw, connection=connection)
except Exception as exc: # noqa: BLE001
return dump_tool_result({"success": False, "error": f"repeat_request failed: {exc}"})
# ----------------------------------------------------------------------
# scope_rules
# ----------------------------------------------------------------------
@strix_tool(timeout=60)
async def scope_rules(
ctx: RunContextWrapper,
@@ -255,14 +462,13 @@ async def scope_rules(
"*woff*", "*.ttf"]``.
Each scope has a unique id usable as ``scope_id`` in
``list_requests`` / ``list_sitemap`` / ``view_request``.
``list_requests``.
Args:
action:
- ``list`` — return all scopes.
- ``get`` — single scope by ``scope_id`` (or all when
omitted).
- ``get`` — single scope by ``scope_id``.
- ``create`` — needs ``scope_name``, optionally
``allowlist`` / ``denylist``.
- ``update`` — needs ``scope_id`` + ``scope_name``;
@@ -275,79 +481,202 @@ async def scope_rules(
scope_id: Required for ``get`` / ``update`` / ``delete``.
scope_name: Required for ``create`` / ``update``.
"""
return dump_tool_result(
await post_to_sandbox(
ctx,
"scope_rules",
{
"action": action,
"allowlist": allowlist,
"denylist": denylist,
"scope_id": scope_id,
"scope_name": scope_name,
},
),
)
client = _ctx_client(ctx)
if client is None:
return _no_client()
try:
if action == "list":
scopes = await client.scope.list()
return dump_tool_result(
{"success": True, "scopes": [_serialize(s) for s in scopes]},
)
if action == "get":
if not scope_id:
return dump_tool_result(
{"success": False, "error": "scope_id required for get"},
)
scope = await client.scope.get(scope_id)
return dump_tool_result({"success": True, "scope": _serialize(scope)})
if action == "create":
if not scope_name:
return dump_tool_result(
{"success": False, "error": "scope_name required for create"},
)
scope = await client.scope.create(
CreateScopeOptions(
name=scope_name,
allowlist=list(allowlist or []),
denylist=list(denylist or []),
),
)
return dump_tool_result({"success": True, "scope": _serialize(scope)})
if action == "update":
if not scope_id or not scope_name:
return dump_tool_result(
{
"success": False,
"error": "scope_id and scope_name required for update",
},
)
scope = await client.scope.update(
scope_id,
UpdateScopeOptions(
name=scope_name,
allowlist=list(allowlist or []),
denylist=list(denylist or []),
),
)
return dump_tool_result({"success": True, "scope": _serialize(scope)})
# action == "delete" — exhaustive Literal
if not scope_id:
return dump_tool_result(
{"success": False, "error": "scope_id required for delete"},
)
await client.scope.delete(scope_id)
return dump_tool_result({"success": True, "deleted": scope_id})
except Exception as exc: # noqa: BLE001
return dump_tool_result({"success": False, "error": f"scope_rules failed: {exc}"})
@strix_tool(timeout=60)
async def list_sitemap(
ctx: RunContextWrapper,
scope_id: str | None = None,
parent_id: str | None = None,
depth: SitemapDepth = "DIRECT",
page: int = 1,
# ----------------------------------------------------------------------
# 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:
"""View the hierarchical sitemap of discovered attack surface.
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']}"
The sitemap is built from proxied traffic — every URL the target
served gets indexed into a tree of domains → directories → request
leaves. Use it to understand application structure and find
interesting endpoints, hidden directories, parameter variations.
Entry kinds you'll encounter:
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
- ``DOMAIN`` — root host (``example.com``).
- ``DIRECTORY`` — path segment (``/api/``, ``/admin/``).
- ``REQUEST`` — a specific endpoint.
- ``REQUEST_BODY`` — POST/PUT body variations (different payloads
seen at the same URL).
- ``REQUEST_QUERY`` — query-string variations.
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)))
Each entry has ``hasDescendants`` — set ``parent_id`` to that
entry's id to drill in. Pages return 30 entries each.
if "headers" in modifications:
headers.update(modifications["headers"])
Args:
scope_id: Filter to a specific scope.
parent_id: Drill into a subtree. ``None`` returns root domains.
depth: ``"DIRECT"`` (immediate children) or ``"ALL"`` (recursive).
page: 1-indexed page (30 entries/page).
"""
return dump_tool_result(
await post_to_sandbox(
ctx,
"list_sitemap",
{
"scope_id": scope_id,
"parent_id": parent_id,
"depth": depth,
"page": page,
},
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),
),
)
@strix_tool(timeout=60)
async def view_sitemap_entry(ctx: RunContextWrapper, entry_id: str) -> str:
"""Examine one sitemap entry — full metadata + every related request.
Use this after ``list_sitemap`` identifies an interesting directory
or endpoint to see all the requests captured under it (methods,
paths, response codes, timing).
Args:
entry_id: Sitemap entry id from ``list_sitemap``.
"""
return dump_tool_result(
await post_to_sandbox(ctx, "view_sitemap_entry", {"entry_id": entry_id}),
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 dump_tool_result(
{
"success": result.status == "DONE",
"status": result.status,
"error": result.error,
"session_id": str(session.id),
"elapsed_ms": elapsed_ms,
"response": response,
},
)
strix/tools/python/python_instance.py
-21
View File
@@ -25,27 +25,6 @@ class PythonInstance:
self.shell.init_history()
self.shell.init_logger()
self._setup_proxy_functions()
def _setup_proxy_functions(self) -> None:
try:
from strix.tools.proxy import proxy_actions
proxy_functions = [
"list_requests",
"list_sitemap",
"repeat_request",
"scope_rules",
"send_request",
"view_request",
"view_sitemap_entry",
]
proxy_dict = {name: getattr(proxy_actions, name) for name in proxy_functions}
self.shell.user_ns.update(proxy_dict)
except ImportError:
pass
def _validate_session(self) -> dict[str, Any] | None:
if not self.is_running:
return {
uv.lock
Generated
+35 -17
View File
@@ -292,6 +292,32 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/72/76/20fa66124dbe6be5cafeb312ece67de6b61dd91a0247d1ea13db4ebb33c2/cachetools-5.5.2-py3-none-any.whl", hash = "sha256:d26a22bcc62eb95c3beabd9f1ee5e820d3d2704fe2967cbe350e20c8ffcd3f0a", size = 10080, upload-time = "2025-02-20T21:01:16.647Z" },
]
[[package]]
name = "caido-sdk-client"
version = "0.2.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "caido-server-auth" },
{ name = "gql", extra = ["aiohttp", "websockets"] },
{ name = "pydantic" },
]
sdist = { url = "https://files.pythonhosted.org/packages/62/d7/5381d8d94fec799bec7004decbf33a4c5581a8374941fe784e730e01cf80/caido_sdk_client-0.2.0.tar.gz", hash = "sha256:39988fe07b3fa9c69adbd49662db660d7707d60d9245109b1623def97b39bac8", size = 57436, upload-time = "2026-04-12T22:25:12.084Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/30/ae/3530caa6a79bafb8049374ca09515686d98532aca73c4fdbd0f6e06de5c9/caido_sdk_client-0.2.0-py3-none-any.whl", hash = "sha256:bc573651681c093ee9663c7924d38d522a89cea60e2ce00d34ba9b02942b1da1", size = 96207, upload-time = "2026-04-12T22:25:11.168Z" },
]
[[package]]
name = "caido-server-auth"
version = "0.1.2"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "gql", extra = ["aiohttp", "websockets"] },
]
sdist = { url = "https://files.pythonhosted.org/packages/69/be/58cc2eaf97f729124b8939a9ea1c1a664b2d96dce0448788df073fca3ac9/caido_server_auth-0.1.2.tar.gz", hash = "sha256:eb2c25e9de15062760b68112f5d8e9ad63eeb1322518b90c1a0119a69a7524a4", size = 6559, upload-time = "2026-03-14T20:41:55.119Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/76/7b/14d192151bcc3c1624cfb488c59ec03e96c1009d015089d729c1aecd26e9/caido_server_auth-0.1.2-py3-none-any.whl", hash = "sha256:40c6cd3728e24cdff402c4efa5d8f55bf6e6cc73ac0169bdea1ad1e34faff8ff", size = 10197, upload-time = "2026-03-14T20:41:54.091Z" },
]
[[package]]
name = "catalogue"
version = "2.0.10"
@@ -969,9 +995,11 @@ wheels = [
]
[package.optional-dependencies]
requests = [
{ name = "requests" },
{ name = "requests-toolbelt" },
aiohttp = [
{ name = "aiohttp" },
]
websockets = [
{ name = "websockets" },
]
[[package]]
@@ -3128,18 +3156,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/1e/db/4254e3eabe8020b458f1a747140d32277ec7a271daf1d235b70dc0b4e6e3/requests-2.32.5-py3-none-any.whl", hash = "sha256:2462f94637a34fd532264295e186976db0f5d453d1cdd31473c85a6a161affb6", size = 64738, upload-time = "2025-08-18T20:46:00.542Z" },
]
[[package]]
name = "requests-toolbelt"
version = "1.0.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "requests" },
]
sdist = { url = "https://files.pythonhosted.org/packages/f3/61/d7545dafb7ac2230c70d38d31cbfe4cc64f7144dc41f6e4e4b78ecd9f5bb/requests-toolbelt-1.0.0.tar.gz", hash = "sha256:7681a0a3d047012b5bdc0ee37d7f8f07ebe76ab08caeccfc3921ce23c88d5bc6", size = 206888, upload-time = "2023-05-01T04:11:33.229Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/3f/51/d4db610ef29373b879047326cbf6fa98b6c1969d6f6dc423279de2b1be2c/requests_toolbelt-1.0.0-py2.py3-none-any.whl", hash = "sha256:cccfdd665f0a24fcf4726e690f65639d272bb0637b9b92dfd91a5568ccf6bd06", size = 54481, upload-time = "2023-05-01T04:11:28.427Z" },
]
[[package]]
name = "rich"
version = "13.9.4"
@@ -3527,6 +3543,8 @@ name = "strix-agent"
version = "0.8.3"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
{ name = "caido-sdk-client" },
{ name = "cvss" },
{ name = "docker" },
{ name = "openai-agents", extra = ["litellm"] },
@@ -3540,7 +3558,6 @@ dependencies = [
[package.optional-dependencies]
sandbox = [
{ name = "fastapi" },
{ name = "gql", extra = ["requests"] },
{ name = "ipython" },
{ name = "libtmux" },
{ name = "openhands-aci" },
@@ -3560,10 +3577,11 @@ dev = [
[package.metadata]
requires-dist = [
{ name = "aiohttp", specifier = ">=3.10.0" },
{ name = "caido-sdk-client", specifier = ">=0.2.0" },
{ name = "cvss", specifier = ">=3.2" },
{ name = "docker", specifier = ">=7.1.0" },
{ name = "fastapi", marker = "extra == 'sandbox'" },
{ name = "gql", extras = ["requests"], marker = "extra == 'sandbox'", specifier = ">=3.5.3" },
{ name = "ipython", marker = "extra == 'sandbox'", specifier = ">=9.3.0" },
{ name = "libtmux", marker = "extra == 'sandbox'", specifier = ">=0.46.2" },
{ name = "openai-agents", extras = ["litellm"], specifier = "==0.14.6" },