Compare commits

...
Author SHA1 Message Date
Alex Schapiro 1e56e1c010 fix(proxy): close replacement Caido client when project.select fails
Addresses Greptile P1: in reconnect_caido (and bootstrap_caido) a successful
connect() followed by a failing project.select()/create() discarded the
connected client without closing it, so a missing/unavailable project could
leak a transport on every retry. Close the client before re-raising.
2026-07-17 02:40:54 +00:00
Alex Schapiro ed6ca662f0 fix(proxy): host-side reconnect, close stale clients, don't retry mutations
Addresses Greptile review on the reconnect logic:

- Host path had no reconnect: a dead shared context client (Caido restart /
  network blip) previously disabled proxy tools for the rest of the scan. Add
  SharedCaidoClient, a serialized reconnect-safe holder stored once per scan in
  the run context and shared across agents. On a dead transport it rebuilds via
  reconnect_caido, which re-selects the SAME Caido project (preserving captured
  traffic) instead of creating a new empty one.
- Don't repeat completed mutations: call_with_client / SharedCaidoClient.call
  take idempotent=. Reads retry once on reconnect; replay + scope
  create/update/delete heal the client but re-raise instead of risking a
  double-apply.
- Don't leak replaced clients: the stale client is aclose()d (best-effort) on
  every reconnect.
- Extend tests to cover close-on-reconnect, non-idempotent re-raise, and the
  SharedCaidoClient holder.
2026-07-17 02:34:39 +00:00
Alex Schapiro 907ea79714 fix(proxy,tooling): serialize+reconnect Caido client, actionable HTTPQL errors, sandbox tool guidance
Addresses the top recurring agent tool-call failures observed in telemetry:

- proxy: the shared Caido client had no locking or reconnect, so concurrent
  agent calls raced ("Transport is already connected") and a dead transport
  poisoned the rest of the run ("Connector is closed"/"Server disconnected").
  Add an asyncio lock + bounded reconnect in caido_api.call_with_client (sandbox
  path) and a scan-wide caido_lock in the run context that host-side proxy tools
  hold around every call. Deterministic errors are not retried.
- proxy: list_requests now returns Caido's exact parser message, echoes the
  offending query, and includes a corrected-syntax hint so agents self-correct
  instead of retrying a broken HTTPQL filter.
- shell/prompt: document that write_stdin requires a process started with
  tty=true; nudge toward writing Python to a file over deeply-nested one-liners;
  note the venv pre-installs common libs.
- agent-browser: distinguish daemon/connection failures (run doctor, don't loop)
  from malformed commands; invoke directly (no sh -c wrapper).
- containers: use POSIX '.' instead of the bashism 'source' in generated rc
  files (fixes 'sh: source: not found'); add file + xxd and pre-install
  requests/httpx/beautifulsoup4/lxml/pyjwt/cryptography in the sandbox venv.
- tests: cover proxy serialization/reconnect/no-retry and HTTPQL errors.
2026-07-17 02:14:24 +00:00
Ahmed AllamandGitHub af65796ec0 fix(runtime): close the docker client on session cleanup (#787) 2026-07-16 11:06:18 -07:00
Ahmed AllamandGitHub e2eb39a02e fix(runtime): cap sandbox container logs to prevent host disk exhaustion (#785) 2026-07-16 09:17:46 -07:00
Ahmed AllamandAhmed Allam 3a50a5ab0e docs(python skill): recommend a task-unique PoC filename to avoid inter-agent collisions 2026-07-16 09:15:40 -07:00
Ahmed AllamandAhmed Allam a529d7f73a docs(python skill): use a distinctive PoC filename to avoid clobbering project files 2026-07-16 09:15:40 -07:00
Ahmed AllamandAhmed Allam f6bd617964 docs(prompts,skills): stop hardcoding /workspace/scratch path
The sandbox never creates /workspace/scratch, so guidance pointing agents
there failed on first write. Make the Python/exec_command and recon
output-hygiene guidance path-agnostic (write to a file, relative to the
working dir) instead of naming a directory that may not exist.
2026-07-16 09:15:40 -07:00
devin-ai-integration[bot]andGitHub 6786d24aca docs(tools): guide proportional wait_for_message timeouts (#784) 2026-07-16 07:16:12 -07:00
Ahmed AllamandAhmed Allam 89ee7b9e5e docs(skills): add research-backed katana output-reduction flags
Per projectdiscovery katana docs, add the flags that actually bound
crawl output size and a reduce-then-delete workflow:
- -mdp (max-domain-pages; default is unlimited), -fsu (filter-similar),
  -fs scope, -f url (URL-only), -or/-ob (omit raw/body), -mrs.
- Baseline now includes -mdp 2000 -fsu; new 'Keeping output small'
  section: bound scope/volume, shrink records, distil then delete raw
  crawls.
2026-07-16 04:47:56 -07:00
Ahmed AllamandAhmed Allam 98990bae45 docs(prompts,skills): scope cleanup to own files; dedupe JSONL by URL
Address Greptile review:
- system_prompt: only clean up your own task's files; don't delete
  another agent's files in the shared workspace unless confirmed unused.
- katana.md: extract+dedupe URLs with jq before removing raw .jsonl
  (sort -u on JSONL compares whole records, not URLs).
2026-07-16 04:47:56 -07:00
Ahmed AllamandAhmed Allam 4b619d57a0 docs(prompts,skills): bound recon output for shared-disk hygiene
Add lightweight, always-on disk-hygiene guidance so agents keep recon
artifacts bounded on the shared /workspace instead of writing very large
uncapped crawl output.

- system_prompt.jinja: DISK & SCRATCH HYGIENE note in the shared-workspace
  block; recon PHASE 1 crawl bullet asks to bound each crawl and tidy up.
- skills/tooling/katana.md: bound the baseline/deep examples with -ct,
  add a Keeping-output-manageable note (bound by -ct/-d, reserve -jsl/-kf
  all for narrowed targets, check du -sh, dedupe and remove raw .jsonl).
2026-07-16 04:47:56 -07:00
Devin AIandAhmed Allam 38c2936f69 Revert "fix(runtime): retry transient sandbox startup failures (#768)"
This reverts commit 40f4e67320.
2026-07-16 04:09:08 -07:00
Ahmed AllamandAhmed Allam 16982646df fix(runtime): bound nano_cpus to docker's int64 NanoCPUs range 2026-07-15 18:31:03 -07:00
Ahmed AllamandAhmed Allam 575e10a404 fix(runtime): also suppress OverflowError for non-finite STRIX_SANDBOX_CPUS 2026-07-15 18:31:03 -07:00
Ahmed AllamandAhmed Allam 84185db23b feat(runtime): opt-in resource limits for docker sandbox containers
Apply cgroup caps (mem_limit, shm_size, nano_cpus, pids_limit) to the
sandbox container from STRIX_SANDBOX_* env vars. Unset values keep
docker's unbounded default, so behavior is unchanged unless opted in.
2026-07-15 18:31:03 -07:00
devin-ai-integration[bot]andGitHub 899e07d3a2 fix(core): bound per-agent image memory (proactive budget + inherited-context scrub) (#779) 2026-07-15 18:13:42 -07:00
914207ffb3 feat(runtime): resolve sandbox ports over a shared Docker network (#775)
Co-authored-by: Ahmed Allam <ahmed39652003@gmail.com>
2026-07-15 11:57:42 -07:00
alex sandGitHub 40f4e67320 fix(runtime): retry transient sandbox startup failures (#768)
* fix(runtime): retry transient sandbox startup failures

* fix(runtime): fail closed when sandbox teardown fails
2026-07-14 23:27:06 -04:00
alex sandGitHub d44ca88a18 fix(runtime): stage symlink-safe copies for LocalDir uploads (#766)
The sandbox SDK's LocalDir walker rejects any symlink outright
(LocalDirReadError, reason=symlink_not_supported), so uploading a cloned
repository that commits symlinks (common in JS/TS monorepos) aborts before
the agent starts. Stage such trees into a temp copy first: in-tree links
are dereferenced; out-of-tree, dangling, and cyclic links are dropped and
never followed, preserving the walker's path-escape safety. Symlink-free
trees are uploaded as-is.
2026-07-14 17:40:23 -04:00
24 changed files with 1382 additions and 182 deletions
+3
View File
@@ -24,6 +24,7 @@ RUN apt-get update && \
python3 python3-pip python3-dev python3-venv python3-setuptools \
golang-go \
net-tools dnsutils whois \
file xxd \
jq parallel ripgrep grep \
less man-db procps htop \
iproute2 iputils-ping netcat-traditional \
@@ -192,6 +193,8 @@ RUN mkdir -p /workspace && chown -R pentester:pentester /workspace /app
USER pentester
RUN python3 -m venv /app/.venv && \
/app/.venv/bin/pip install --no-cache-dir caido-sdk-client && \
/app/.venv/bin/pip install --no-cache-dir \
requests httpx beautifulsoup4 lxml pyjwt cryptography && \
/app/.venv/bin/pip install --no-cache-dir -r /home/pentester/tools/jwt_tool/requirements.txt && \
printf '%s\n' \
'#!/bin/bash' \
+6 -3
View File
@@ -91,10 +91,13 @@ http_proxy=http://127.0.0.1:${CAIDO_PORT}
https_proxy=http://127.0.0.1:${CAIDO_PORT}
EOF
echo "source /etc/profile.d/proxy.sh" >> ~/.bashrc
echo "source /etc/profile.d/proxy.sh" >> ~/.zshrc
# Use POSIX `.` (not the bashism `source`) so these lines are safe when the rc
# files are read by a POSIX shell (e.g. `sh -lc`), which otherwise fails with
# "source: not found". `.` is understood by bash, zsh, and dash alike.
echo ". /etc/profile.d/proxy.sh" >> ~/.bashrc
echo ". /etc/profile.d/proxy.sh" >> ~/.zshrc
source /etc/profile.d/proxy.sh
. /etc/profile.d/proxy.sh
echo "✅ System-wide proxy configuration complete"
+25 -4
View File
@@ -168,9 +168,24 @@ EFFICIENCY TACTICS:
- Download additional tools as needed for specific tasks
- Run multiple scans in parallel when possible
- Load the most relevant skill before starting a specialized testing workflow if doing so will improve accuracy, speed, or tool usage
- Use `exec_command` for Python code: write reusable scripts under
`/workspace/scratch/` and run them with `python3`. For one-off snippets,
`python3 -c` or a here-document is acceptable.
- Use `exec_command` for Python code: write reusable scripts to a file and
run them with `python3 script.py`. For one-off snippets, `python3 -c` or a
here-document is acceptable, but avoid deeply nested quotes/parentheses — if
a snippet needs complex quoting or is more than a few lines, write it to a
file first to prevent syntax errors.
- Before importing a third-party Python library, make sure it is installed. The
sandbox's `python3` runs inside a preconfigured virtualenv that ships
`requests`, `httpx`, `beautifulsoup4` (bs4), `lxml`, `pyjwt`, and
`cryptography`; for anything else prefer the stdlib or run `pip install <pkg>`
(it installs into that active venv) before importing, rather than letting the
script fail with `ModuleNotFoundError`.
- `exec_command` runs each command in a fresh non-interactive shell (plain
pipes, no TTY). To drive an interactive or long-running process with
`write_stdin` — REPLs, `ssh`/`nc`/`ftp`, `msfconsole`, or to send Ctrl-C —
you MUST start it with `exec_command(cmd="...", tty=true)` and then
`write_stdin(session_id=<id>, chars="...")`. Calling `write_stdin` on a
default (non-TTY) command or on a process that has already exited fails with
"stdin is not available".
- For Caido proxy automation inside Python, explicitly import from
`caido_api`:
`from caido_api import list_requests, view_request, repeat_request, list_sitemap, view_sitemap_entry, scope_rules`
@@ -240,12 +255,18 @@ AGENT ISOLATION & SANDBOXING:
- All agents share the same /workspace directory and proxy history
- Agents can see each other's files and proxy traffic for better collaboration
DISK & SCRATCH HYGIENE:
- /workspace is a shared, finite disk used by all agents at once — be a considerate tenant
- Prefer bounded recon: scope crawls and scans by depth, duration, and target rather than "collect everything"
- Redirect large tool output to a file, and once you've extracted what you need (e.g. a URL/endpoint list), remove the raw output
- If disk gets tight or a write fails for space, check what's large under /workspace and clean up files from your own task; leave another agent's files unless you've confirmed they're no longer in use
MANDATORY INITIAL PHASES:
BLACK-BOX TESTING - PHASE 1 (RECON & MAPPING):
- COMPLETE full reconnaissance: subdomain enumeration, port scanning, service detection
- MAP entire attack surface: all endpoints, parameters, APIs, forms, inputs
- CRAWL thoroughly: spider all pages (authenticated and unauthenticated), discover hidden paths, analyze JS files
- CRAWL thoroughly: spider all pages (authenticated and unauthenticated), discover hidden paths, analyze JS files — keep each crawl bounded by depth/duration, and tidy up raw output once endpoints are extracted
- ENUMERATE technologies: frameworks, libraries, versions, dependencies
- Reconnaissance should normally happen before targeted vulnerability discovery unless the correct next move is already obvious or the user/system explicitly asks to prioritize a specific area first
- ONLY AFTER comprehensive mapping → proceed to vulnerability testing
+2
View File
@@ -56,6 +56,8 @@ class RuntimeSettings(BaseSettings):
# on large repos). Above this, the user must bind-mount via ``--mount``.
# Set to 0 (or less) to disable the pre-flight check entirely.
max_local_copy_mb: int = Field(default=1024, alias="STRIX_MAX_LOCAL_COPY_MB")
# Max screenshot/image tool outputs kept live per agent context (0 = none).
max_context_images: int = Field(default=3, ge=0, alias="STRIX_MAX_CONTEXT_IMAGES")
class TelemetrySettings(BaseSettings):
+4 -1
View File
@@ -10,6 +10,8 @@ from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal, cast
from strix.core.sessions import session_write_lock
if TYPE_CHECKING:
from agents.items import TResponseInputItem
@@ -137,7 +139,8 @@ class AgentCoordinator:
)
return False
try:
await session.add_items([self._message_to_session_item(message)])
async with session_write_lock(session):
await session.add_items([self._message_to_session_item(message)])
except Exception:
logger.exception(
"agent.send failed to append to SDK session target=%s",
+12 -1
View File
@@ -17,7 +17,11 @@ from openai import APIError
from strix.core.hooks import BudgetExceededError
from strix.core.inputs import child_initial_input
from strix.core.sessions import open_agent_session, strip_all_images_from_session
from strix.core.sessions import (
enforce_image_budget,
open_agent_session,
strip_all_images_from_session,
)
if TYPE_CHECKING:
@@ -349,6 +353,13 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
while True:
try:
await coordinator.mark_running(agent_id)
if session is not None:
max_images = context.get("max_context_images")
if isinstance(max_images, int):
try:
await enforce_image_budget(session, max_images)
except Exception:
logger.exception("image-budget enforcement failed for %s", agent_id)
stream = Runner.run_streamed(
agent,
input=input_data,
+6 -1
View File
@@ -13,6 +13,7 @@ from strix.config.models import (
is_known_openai_bare_model,
model_supports_reasoning,
)
from strix.core.sessions import scrub_images_from_items
if TYPE_CHECKING:
@@ -161,7 +162,11 @@ def child_initial_input(
"""
parts: list[str] = []
if parent_history:
rendered = json.dumps(parent_history, ensure_ascii=False, default=str)
rendered = json.dumps(
scrub_images_from_items(parent_history),
ensure_ascii=False,
default=str,
)
parts.append(
"== Inherited context from parent (background only) ==\n"
f"{rendered}\n"
+5
View File
@@ -282,11 +282,16 @@ async def run_strix_scan(
context: dict[str, Any] = {
"coordinator": coordinator,
"sandbox_session": bundle["session"],
# One ``SharedCaidoClient`` is reused by every agent in the scan
# (child contexts are shallow copies via ``dict(parent_ctx)``). It
# serializes access to the non-concurrency-safe GraphQL transport
# and rebuilds it if it dies mid-scan.
"caido_client": bundle["caido_client"],
"agent_id": root_id,
"parent_id": None,
"interactive": interactive,
"spawn_child_agent": spawn_child_agent,
"max_context_images": settings.runtime.max_context_images,
}
root_session = open_agent_session(root_id, agents_db)
+121 -36
View File
@@ -2,64 +2,149 @@
from __future__ import annotations
import contextlib
import asyncio
import logging
from typing import TYPE_CHECKING, Any, cast
from weakref import WeakKeyDictionary
from agents.memory import SQLiteSession
if TYPE_CHECKING:
from collections.abc import Callable
from pathlib import Path
from agents.items import TResponseInputItem
from agents.memory import Session
logger = logging.getLogger(__name__)
def open_agent_session(agent_id: str, path: Path) -> SQLiteSession:
path.parent.mkdir(parents=True, exist_ok=True)
return SQLiteSession(session_id=agent_id, db_path=path)
_IMAGE_REJECTED_TEXT = "[image rejected by the model]"
_IMAGE_ELIDED_TEXT = "[older screenshot elided to bound context memory]"
_INHERITED_IMAGE_TEXT = "[screenshot omitted from inherited context]"
def _output_has_image(item_dict: dict[str, Any]) -> bool:
return (
item_dict.get("type") == "function_call_output"
and isinstance(item_dict.get("output"), list)
and any(isinstance(b, dict) and b.get("type") == "input_image" for b in item_dict["output"])
)
def _elided_output(item_dict: dict[str, Any], text: str) -> dict[str, Any]:
# Replace only image blocks; sibling text blocks are preserved.
output = item_dict.get("output")
blocks = output if isinstance(output, list) else []
return {
"type": "function_call_output",
"call_id": item_dict.get("call_id"),
"output": [
{"type": "input_text", "text": text}
if isinstance(block, dict) and block.get("type") == "input_image"
else block
for block in blocks
],
}
_session_write_locks: WeakKeyDictionary[Session, asyncio.Lock] = WeakKeyDictionary()
def session_write_lock(session: Session) -> asyncio.Lock:
"""Lock serialising all out-of-band writes to ``session``."""
lock = _session_write_locks.get(session)
if lock is None:
lock = asyncio.Lock()
_session_write_locks[session] = lock
return lock
async def _rewrite_session(
session: Session,
transform: Callable[[list[Any]], tuple[list[Any], bool]],
) -> bool:
"""Read-modify-write a session under its write lock, restoring on failure."""
async with session_write_lock(session):
items = await session.get_items()
if not items:
return False
rebuilt, changed = transform(list(items))
if not changed:
return False
rebuilt_items = cast("list[TResponseInputItem]", rebuilt)
original_items = cast("list[TResponseInputItem]", list(items))
await session.clear_session()
try:
await session.add_items(rebuilt_items)
except Exception:
logger.exception("session rewrite failed; restoring original items")
await session.clear_session()
await session.add_items(original_items)
raise
return True
async def strip_all_images_from_session(session: Session) -> bool:
items = await session.get_items()
if not items:
"""Replace every image tool output with a text placeholder (rejection recovery)."""
def _transform(items: list[Any]) -> tuple[list[Any], bool]:
rebuilt: list[Any] = []
changed = False
for item in items:
item_dict = cast("dict[str, Any]", item) if isinstance(item, dict) else None
if item_dict is not None and _output_has_image(item_dict):
rebuilt.append(_elided_output(item_dict, _IMAGE_REJECTED_TEXT))
changed = True
else:
rebuilt.append(item)
return rebuilt, changed
return await _rewrite_session(session, _transform)
async def enforce_image_budget(session: Session, max_images: int) -> bool:
"""Keep only the most recent ``max_images`` image outputs; elide older ones."""
if max_images < 0:
return False
rebuilt: list[Any] = []
changed = False
for item in items:
item_dict = cast("dict[str, Any]", item) if isinstance(item, dict) else None
if (
item_dict is not None
and item_dict.get("type") == "function_call_output"
and isinstance(item_dict.get("output"), list)
and any(
isinstance(b, dict) and b.get("type") == "input_image" for b in item_dict["output"]
)
):
rebuilt.append(
{
"type": "function_call_output",
"call_id": item_dict.get("call_id"),
"output": [{"type": "input_text", "text": _IMAGE_REJECTED_TEXT}],
},
)
changed = True
else:
rebuilt.append(item)
def _transform(items: list[Any]) -> tuple[list[Any], bool]:
image_indices = [
i
for i, item in enumerate(items)
if isinstance(item, dict) and _output_has_image(cast("dict[str, Any]", item))
]
if len(image_indices) <= max_images:
return items, False
to_elide = set(image_indices[: len(image_indices) - max_images])
rebuilt = [
_elided_output(cast("dict[str, Any]", item), _IMAGE_ELIDED_TEXT)
if i in to_elide
else item
for i, item in enumerate(items)
]
return rebuilt, True
if not changed:
return False
return await _rewrite_session(session, _transform)
rebuilt_items = cast("list[TResponseInputItem]", rebuilt)
await session.clear_session()
try:
await session.add_items(rebuilt_items)
except Exception:
with contextlib.suppress(Exception):
await session.add_items(rebuilt_items)
raise
return True
def scrub_images_from_items(items: list[Any]) -> list[Any]:
"""Return a copy of ``items`` with every image block replaced by text."""
def _scrub(obj: Any) -> Any:
if isinstance(obj, dict):
if obj.get("type") == "input_image":
return {"type": "input_text", "text": _INHERITED_IMAGE_TEXT}
return {k: _scrub(v) for k, v in obj.items()}
if isinstance(obj, list):
return [_scrub(v) for v in obj]
return obj
return [_scrub(item) for item in items]
+59 -9
View File
@@ -10,6 +10,7 @@ exposed-port URL for all subsequent SDK calls.
from __future__ import annotations
import asyncio
import contextlib
import json
import logging
from typing import TYPE_CHECKING
@@ -79,23 +80,72 @@ async def _login_as_guest(
raise RuntimeError(f"loginAsGuest failed after {attempts} attempts: {last_err}")
async def bootstrap_caido(
async def _aclose_quietly(client: Client) -> None:
"""Best-effort close of a client whose setup failed; never raises."""
with contextlib.suppress(Exception):
await client.aclose()
async def _connect_client(
session: BaseSandboxSession,
*,
host_url: str,
container_url: str,
) -> Client:
"""Connect to the in-container Caido sidecar and select a fresh project."""
logger.info("Bootstrapping Caido client (host=%s, container=%s)", host_url, container_url)
access_token = await _login_as_guest(session, container_url=container_url)
client = Client(host_url, auth=TokenAuthOptions(token=access_token))
await client.connect()
return client
project = await client.project.create(
CreateProjectOptions(name="sandbox", temporary=True),
)
await client.project.select(project.id)
async def bootstrap_caido(
session: BaseSandboxSession,
*,
host_url: str,
container_url: str,
) -> tuple[Client, str]:
"""Connect to the in-container Caido sidecar and select a fresh project.
Returns the connected client and the id of the temporary project it
selected. The project id lets :func:`reconnect_caido` rebuild a dead
transport while staying on the *same* project (and its captured traffic)
instead of creating a new empty one.
"""
logger.info("Bootstrapping Caido client (host=%s, container=%s)", host_url, container_url)
client = await _connect_client(session, host_url=host_url, container_url=container_url)
try:
project = await client.project.create(
CreateProjectOptions(name="sandbox", temporary=True),
)
await client.project.select(project.id)
except BaseException:
# Don't leak the connected transport if project setup fails.
await _aclose_quietly(client)
raise
logger.info("Caido project selected: %s", project.id)
return client, str(project.id)
async def reconnect_caido(
session: BaseSandboxSession,
*,
host_url: str,
container_url: str,
project_id: str,
) -> Client:
"""Rebuild a Caido client after its transport died, keeping the project.
Re-authenticates, reconnects, and re-selects the existing project so the
caller keeps access to the traffic captured before the disconnect.
"""
logger.info("Reconnecting Caido client (host=%s, project=%s)", host_url, project_id)
client = await _connect_client(session, host_url=host_url, container_url=container_url)
try:
await client.project.select(project_id)
except BaseException:
# A missing/unavailable project must not leave the freshly-connected
# transport dangling — otherwise every retry leaks another one.
await _aclose_quietly(client)
raise
return client
+116 -1
View File
@@ -24,20 +24,25 @@ from __future__ import annotations
import contextlib
import logging
import os
import uuid
from typing import Any
from typing import Any, cast
from agents.sandbox.errors import ExposedPortUnavailableError
from agents.sandbox.manifest import Manifest
from agents.sandbox.sandboxes.docker import (
DockerSandboxClient,
DockerSandboxSession,
_build_docker_volume_mounts,
_docker_port_key,
_manifest_requires_fuse,
_manifest_requires_sys_admin,
)
from agents.sandbox.session.sandbox_session import SandboxSession
from agents.sandbox.types import ExposedPortEndpoint
from docker import errors as docker_errors # type: ignore[import-untyped, unused-ignore]
from docker.models.containers import Container # type: ignore[import-untyped, unused-ignore]
from docker.types import LogConfig # type: ignore[import-untyped, unused-ignore]
from docker.types import Mount as DockerSDKMount # type: ignore[import-untyped, unused-ignore]
from docker.utils import parse_repository_tag # type: ignore[import-untyped, unused-ignore]
from requests.exceptions import RequestException
@@ -46,6 +51,103 @@ from requests.exceptions import RequestException
logger = logging.getLogger(__name__)
_SANDBOX_NETWORK_ENV = "STRIX_DOCKER_SANDBOX_NETWORK"
def _sandbox_network() -> str | None:
value = os.environ.get(_SANDBOX_NETWORK_ENV, "").strip()
return value or None
def _apply_sandbox_network(create_kwargs: dict[str, Any]) -> None:
network = _sandbox_network()
if network:
create_kwargs["network"] = network
create_kwargs.pop("ports", None)
def _apply_resource_limits(create_kwargs: dict[str, Any]) -> None:
"""Apply optional cgroup resource caps from the environment. Unset/blank
values leave docker's default (unbounded), so this is opt-in per host."""
mem_limit = os.environ.get("STRIX_SANDBOX_MEM_LIMIT", "").strip()
if mem_limit:
create_kwargs["mem_limit"] = mem_limit
shm_size = os.environ.get("STRIX_SANDBOX_SHM_SIZE", "").strip()
if shm_size:
create_kwargs["shm_size"] = shm_size
cpus = os.environ.get("STRIX_SANDBOX_CPUS", "").strip()
if cpus:
with contextlib.suppress(ValueError, OverflowError):
nano_cpus = int(float(cpus) * 1_000_000_000)
if 0 < nano_cpus <= 2**63 - 1:
create_kwargs["nano_cpus"] = nano_cpus
pids_limit = os.environ.get("STRIX_SANDBOX_PIDS_LIMIT", "").strip()
if pids_limit:
with contextlib.suppress(ValueError):
create_kwargs["pids_limit"] = int(pids_limit)
def _apply_log_limits(create_kwargs: dict[str, Any]) -> None:
"""Bound the container's json-file log so a runaway process in the sandbox
(e.g. a tool that busy-loops writing to stdout) cannot fill the host disk
and take the Docker daemon down with it.
Unlike the cgroup caps above, this defaults **on** — docker's own default
is an unbounded json-file, which is unsafe for an autonomous agent that
executes arbitrary commands. ``max-file`` rotation means the on-disk cap is
``max-size * max-file``. Set ``STRIX_SANDBOX_LOG_MAX_SIZE`` to ``0``/``off``
to opt back out to docker's default."""
max_size = os.environ.get("STRIX_SANDBOX_LOG_MAX_SIZE", "50m").strip()
if max_size.lower() in ("0", "off", "none", "unlimited"):
return
max_file = os.environ.get("STRIX_SANDBOX_LOG_MAX_FILE", "3").strip() or "3"
create_kwargs["log_config"] = LogConfig(
type=LogConfig.types.JSON,
config={"max-size": max_size, "max-file": max_file},
)
class StrixDockerSandboxSession(DockerSandboxSession):
sandbox_network: str = ""
async def _resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
try:
self._container.reload()
except docker_errors.APIError as e:
raise ExposedPortUnavailableError(
port=port,
exposed_ports=self.state.exposed_ports,
reason="backend_unavailable",
context={
"backend": "docker",
"detail": "container_reload_failed",
"network": self.sandbox_network,
},
cause=e,
) from e
attrs = getattr(self._container, "attrs", {}) or {}
networks = attrs.get("NetworkSettings", {}).get("Networks", {})
endpoint = networks.get(self.sandbox_network) or {}
ip = endpoint.get("IPAddress") or endpoint.get("GlobalIPv6Address")
if not isinstance(ip, str) or not ip:
raise ExposedPortUnavailableError(
port=port,
exposed_ports=self.state.exposed_ports,
reason="backend_unavailable",
context={
"backend": "docker",
"detail": "container_not_on_network",
"network": self.sandbox_network,
},
)
host = f"[{ip}]" if ":" in ip else ip
return ExposedPortEndpoint(host=host, port=port, tls=False)
class StrixDockerSandboxClient(DockerSandboxClient):
# Host directories to bind-mount into the container, set by the docker
# backend before ``create()``. Each item is ``{source, target, read_only}``.
@@ -117,6 +219,10 @@ class StrixDockerSandboxClient(DockerSandboxClient):
extra_hosts = create_kwargs.setdefault("extra_hosts", {})
extra_hosts["host.docker.internal"] = "host-gateway"
_apply_sandbox_network(create_kwargs)
_apply_resource_limits(create_kwargs)
_apply_log_limits(create_kwargs)
# Strix injection: host bind mounts (e.g. large repos passed via --mount)
# that bypass the SDK's file-by-file LocalDir copy.
bind_mounts = getattr(self, "strix_bind_mounts", ())
@@ -146,6 +252,15 @@ class StrixDockerSandboxClient(DockerSandboxClient):
)
return container
async def create(self, **kwargs: Any) -> SandboxSession:
session = await super().create(**kwargs)
network = _sandbox_network()
inner = session._inner
if network and isinstance(inner, DockerSandboxSession):
inner.__class__ = StrixDockerSandboxSession
cast("StrixDockerSandboxSession", inner).sandbox_network = network
return session
async def delete(self, session: SandboxSession) -> SandboxSession:
container_id = getattr(getattr(session._inner, "state", None), "container_id", None)
if container_id:
+120
View File
@@ -0,0 +1,120 @@
"""Symlink-safe staging for ``LocalDir`` manifest uploads.
The sandbox SDK's ``LocalDir`` walker refuses to copy symlinks at all — it
raises ``LocalDirReadError(reason="symlink_not_supported")`` on the first one
as a path-escape / TOCTOU safeguard. Real source trees (especially JS/TS
monorepos with workspace or shared-config links) routinely commit symlinks, so
handing such a tree straight to ``LocalDir`` aborts the upload before the agent
even starts.
:func:`stage_symlink_safe_dir` returns a path that is always safe to hand to
``LocalDir``:
* a tree with no symlinks is used as-is (no copy);
* otherwise the tree is copied into a temp directory with symlinks resolved:
- a link whose target stays inside the tree is *dereferenced* (its target
content is materialized in place), so the agent still sees the file;
- a link that escapes the tree, dangles, or forms a cycle is *dropped* and
never followed. Refusing to follow out-of-tree links preserves the walker's
path-escape safety and keeps host/out-of-tree content from leaking into the
(hostile) sandbox.
Regular files are hard-linked when possible (falling back to a copy across
devices), so the staged tree adds negligible disk for the non-symlink bulk.
"""
from __future__ import annotations
import logging
import os
import shutil
import tempfile
from pathlib import Path
logger = logging.getLogger(__name__)
_STAGING_PREFIX = "strix-localdir-"
def _is_within(target: Path, root: Path) -> bool:
"""Return whether ``target`` is ``root`` itself or nested under it."""
if target == root:
return True
try:
target.relative_to(root)
except ValueError:
return False
return True
def tree_has_symlink(root: Path) -> bool:
"""Return whether ``root`` contains any symlink (file or directory)."""
for dirpath, dirnames, filenames in os.walk(root, followlinks=False):
base = Path(dirpath)
for name in (*dirnames, *filenames):
if (base / name).is_symlink():
return True
return False
def _link_or_copy(src: Path, dst: Path) -> None:
"""Hard-link ``src`` to ``dst``, falling back to a content copy."""
try:
os.link(src, dst)
except OSError:
shutil.copy2(src, dst, follow_symlinks=True)
def _stage_dir(src: Path, dst: Path, root: Path, seen: frozenset[Path]) -> None:
dst.mkdir(parents=True, exist_ok=True)
for entry in os.scandir(src):
entry_path = Path(entry.path)
dest_path = dst / entry.name
if entry.is_symlink():
target = Path(os.path.realpath(entry_path))
if not _is_within(target, root):
logger.warning("staging: dropping out-of-tree symlink %s -> %s", entry_path, target)
continue
if not target.exists():
logger.warning("staging: dropping dangling symlink %s", entry_path)
continue
if target in seen:
logger.warning("staging: dropping cyclic symlink %s -> %s", entry_path, target)
continue
if target.is_dir():
_stage_dir(target, dest_path, root, seen | {target})
else:
_link_or_copy(target, dest_path)
elif entry.is_dir(follow_symlinks=False):
_stage_dir(entry_path, dest_path, root, seen)
elif entry.is_file(follow_symlinks=False):
_link_or_copy(entry_path, dest_path)
else:
# Sockets, FIFOs, devices — not part of a source tree; skip.
logger.debug("staging: skipping non-regular entry %s", entry_path)
def stage_symlink_safe_dir(src_root: Path) -> tuple[Path, Path | None]:
"""Return ``(upload_path, staged_temp)`` for uploading ``src_root``.
``upload_path`` is safe to hand to ``LocalDir``. When the tree contains no
symlinks it is ``src_root`` itself and ``staged_temp`` is ``None``.
Otherwise a symlink-safe copy is materialized in a temp directory and both
returned values point at it; the caller owns removing ``staged_temp`` once
the upload completes.
"""
root = src_root.resolve()
if not tree_has_symlink(root):
return root, None
staged = Path(tempfile.mkdtemp(prefix=_STAGING_PREFIX))
try:
_stage_dir(root, staged, root, frozenset({root}))
except OSError:
shutil.rmtree(staged, ignore_errors=True)
raise
logger.info("staging: materialized symlink-safe copy of %s at %s", root, staged)
return staged, staged
+50 -16
View File
@@ -3,15 +3,22 @@
from __future__ import annotations
import logging
import shutil
from pathlib import Path
from typing import Any
from typing import TYPE_CHECKING, Any
from agents.sandbox.entries import BaseEntry, LocalDir
from agents.sandbox.manifest import Environment, Manifest
from strix.config import load_settings
from strix.runtime.backends import get_backend
from strix.runtime.caido_bootstrap import bootstrap_caido
from strix.runtime.caido_bootstrap import bootstrap_caido, reconnect_caido
from strix.runtime.local_dir_staging import stage_symlink_safe_dir
from strix.tools.proxy.caido_api import SharedCaidoClient
if TYPE_CHECKING:
from caido_sdk_client import Client as CaidoClient
logger = logging.getLogger(__name__)
@@ -29,16 +36,20 @@ _WORKSPACE_ROOT = "/workspace"
def build_session_entries(
local_sources: list[dict[str, Any]],
) -> tuple[dict[str | Path, BaseEntry], list[dict[str, Any]]]:
) -> tuple[dict[str | Path, BaseEntry], list[dict[str, Any]], list[Path]]:
"""Split local sources into copied manifest entries and host bind mounts.
Sources flagged ``mount`` are bind-mounted read-only at
``/workspace/<workspace_subdir>`` (not added to the manifest, so the SDK
does not stream them in file-by-file). Every other source becomes a
``LocalDir`` entry copied into the container as before.
``LocalDir`` entry copied into the container as before. Trees containing
symlinks (which the SDK's ``LocalDir`` walker refuses outright) are first
staged into a symlink-safe temp copy; those temp dirs are returned so the
caller can remove them once the upload completes.
"""
entries: dict[str | Path, BaseEntry] = {}
bind_mounts: list[dict[str, Any]] = []
staged_dirs: list[Path] = []
for src in local_sources:
ws_subdir = src.get("workspace_subdir") or ""
host_path = src.get("source_path") or ""
@@ -54,8 +65,11 @@ def build_session_entries(
}
)
else:
entries[ws_subdir] = LocalDir(src=resolved)
return entries, bind_mounts
upload_path, staged = stage_symlink_safe_dir(resolved)
if staged is not None:
staged_dirs.append(staged)
entries[ws_subdir] = LocalDir(src=upload_path)
return entries, bind_mounts, staged_dirs
async def create_or_reuse(
@@ -75,7 +89,7 @@ async def create_or_reuse(
logger.info("Reusing existing sandbox session for scan %s", scan_id)
return cached
entries, bind_mounts = build_session_entries(local_sources)
entries, bind_mounts, staged_dirs = build_session_entries(local_sources)
# Caido runs as an in-container sidecar; HTTP(S) traffic from any
# process started via ``session.exec`` (the SDK's Shell tool, etc.)
@@ -106,28 +120,40 @@ async def create_or_reuse(
backend_name,
image,
)
client, session = await backend(
image=image,
manifest=manifest,
exposed_ports=(_CONTAINER_CAIDO_PORT,),
bind_mounts=bind_mounts,
)
try:
client, session = await backend(
image=image,
manifest=manifest,
exposed_ports=(_CONTAINER_CAIDO_PORT,),
bind_mounts=bind_mounts,
)
finally:
for staged in staged_dirs:
shutil.rmtree(staged, ignore_errors=True)
caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT)
scheme = "https" if caido_endpoint.tls else "http"
host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}"
logger.debug("Caido host endpoint resolved: %s", host_caido_url)
caido_client = await bootstrap_caido(
caido_client, caido_project_id = await bootstrap_caido(
session,
host_url=host_caido_url,
container_url=container_caido_url,
)
async def _reconnect_caido() -> CaidoClient:
return await reconnect_caido(
session,
host_url=host_caido_url,
container_url=container_caido_url,
project_id=caido_project_id,
)
bundle = {
"client": client,
"session": session,
"caido_client": caido_client,
"caido_client": SharedCaidoClient(caido_client, _reconnect_caido),
}
_SESSION_CACHE[scan_id] = bundle
logger.info("Sandbox session for scan %s ready and cached", scan_id)
@@ -154,11 +180,19 @@ async def cleanup(scan_id: str) -> None:
except Exception: # noqa: BLE001
logger.debug("cleanup(%s): caido_client.aclose() raised", scan_id, exc_info=True)
client = bundle["client"]
try:
await bundle["client"].delete(bundle["session"])
await client.delete(bundle["session"])
logger.info("Cleaned up sandbox session for scan %s", scan_id)
except Exception:
logger.exception(
"cleanup(%s): client.delete raised; container may need manual reaping",
scan_id,
)
docker_client = getattr(client, "docker_client", None)
if docker_client is not None:
try:
docker_client.close()
except Exception: # noqa: BLE001
logger.debug("cleanup(%s): docker_client.close() raised", scan_id, exc_info=True)
+17
View File
@@ -365,6 +365,23 @@ agent-browser dialog accept "text" # accept with prompt input
agent-browser dialog dismiss # cancel
```
## Readiness & recovery
The first `agent-browser open` in a session launches the headless-Chrome
daemon; later commands reuse it. Distinguish the two failure modes and react
differently — do **not** blindly re-run the same failing command in a loop:
- **Daemon / connection failure** (`Failed to connect`, `connection refused`,
socket missing, `browser not running`): the daemon isn't up or has died. Run
`agent-browser doctor` (add `--fix` if it reports repairable problems), then
re-open the page. Retrying the original command unchanged will keep failing.
- **Malformed command** (`Unknown command`, `Ref not found`, bad flag): fix the
command itself — re-snapshot for fresh refs, or correct the syntax.
Invoke `agent-browser` directly through `exec_command`; there is no need to wrap
it in an extra `sh -c "..."` / `bash -lc "..."` layer, which only adds shell
quoting and startup-file pitfalls.
## Diagnosing install issues
If a command fails unexpectedly (`Unknown command`, `Failed to connect`,
+18 -3
View File
@@ -24,7 +24,15 @@ High-signal flags:
- `-p, -parallelism <n>` concurrent input targets
- `-rl, -rate-limit <n>` request rate limit
- `-timeout <seconds>` request timeout
- `-ct, -crawl-duration <s|m|h|d>` maximum time to crawl the target
- `-retry <n>` retry count
- `-mdp, -max-domain-pages <n>` cap pages crawled per domain (default: unlimited)
- `-fsu, -filter-similar` collapse similar URLs (e.g. /users/123 and /users/456)
- `-fs, -field-scope <dn|rdn|fqdn|regex>` crawl scope (default `rdn` = root domain + ALL subdomains)
- `-f, -field <url|path|...>` emit only one field (e.g. `-f url` for a plain URL list)
- `-or, -omit-raw` omit raw request/response from JSONL output
- `-ob, -omit-body` omit response body from JSONL output
- `-mrs, -max-response-size <bytes>` cap per-response bytes read (default 4194304)
- `-ef, -extension-filter <list>` extension exclusions
- `-tlsi, -tls-impersonate` experimental JA3/TLS impersonation
- `-hl, -headless` enable hybrid headless crawling
@@ -37,13 +45,13 @@ High-signal flags:
- `-silent`, `-j, -jsonl`, `-o <file>` output controls
Agent-safe baseline for automation:
`mkdir -p crawl && katana -u https://target.tld -d 3 -jc -kf robotstxt -c 10 -p 10 -rl 50 -timeout 10 -retry 1 -ef png,jpg,jpeg,gif,svg,css,woff,woff2,ttf,eot,map -silent -j -o crawl/katana.jsonl`
`mkdir -p crawl && katana -u https://target.tld -d 3 -ct 10m -mdp 2000 -fsu -jc -kf robotstxt -c 10 -p 10 -rl 50 -timeout 10 -retry 1 -ef png,jpg,jpeg,gif,svg,css,woff,woff2,ttf,eot,map -silent -j -o crawl/katana.jsonl`
Common patterns:
- Fast crawl baseline:
`katana -u https://target.tld -d 3 -jc -silent`
- Deeper JS-aware crawl:
`katana -u https://target.tld -d 5 -jc -jsl -kf all -c 10 -p 10 -rl 50 -o katana_urls.txt`
- Deeper JS-aware crawl (narrowed target; keep it time-bounded):
`katana -u https://target.tld -d 5 -ct 15m -jc -jsl -kf all -c 10 -p 10 -rl 50 -o katana_urls.txt`
- Multi-target run with JSONL output:
`katana -list urls.txt -d 3 -jc -silent -j -o katana.jsonl`
- Headless crawl with local Chrome:
@@ -59,6 +67,13 @@ Critical correctness rules:
- For `-kf`, keep depth at least `-d 3` so known files are fully covered.
- If writing to a file, ensure parent directory exists before `-o`.
Keeping output small (katana has NO default page cap, so plan for volume):
- Bound scope and volume: `-fs fqdn` (or `-cs`/`-cos` regex) so the crawl doesn't wander across every subdomain, `-mdp <n>` to cap pages per domain, `-fsu` to collapse near-identical URLs, and `-ct`/`-d` to bound time and depth.
- Shrink each record: default JSONL is verbose. If you only need endpoints, emit a plain URL list with `-f url` instead of `-j`. If you need JSONL, drop the heavy parts with `-or` (omit raw) and `-ob` (omit body), and lower `-mrs` to cap per-response bytes.
- Reserve `-jsl` / `-kf all` / higher `-d` for a specific narrowed target — they multiply output fast on large sites.
- Reduce, then delete: once the crawl finishes, extract just what you need (e.g. `katana ... -f url -o urls.txt` or `sort -u` a URL list, or a short note of interesting paths) and remove the raw crawl file/dir. Don't keep large raw crawls around after you've distilled them.
- Sanity-check size (`du -sh <out>`); if it's outsized for the scope, tighten `-fs`/`-mdp`/`-fsu`/`-d`/`-ct` and re-run rather than keeping it.
Usage rules:
- Keep `-d`, `-c`, `-p`, and `-rl` explicit for reproducible runs.
- Use `-ef` early to reduce static-file noise before fuzzing.
+17 -8
View File
@@ -7,9 +7,9 @@ description: Run Python through exec_command in the SDK sandbox. Use the image-b
Use `exec_command` for Python. There is no separate Strix Python executor.
Prefer writing reusable scripts to `/workspace/scratch/<name>.py` and
running them with `python3 /workspace/scratch/<name>.py`. For short
one-off transformations, `python3 -c` or a small here-document is fine.
Prefer writing reusable scripts to a `.py` file and running them with
`python3 <name>.py`. For short one-off transformations, `python3 -c` or a
small here-document is fine.
The `shell` parameter on `exec_command` is for swapping POSIX shells
(`bash`/`zsh`/`sh`), not for picking interpreters. Put the interpreter
@@ -84,17 +84,26 @@ automatically, so it shows up in `list_requests` and you can use
For iterative exploit work, put code in a file:
```text
1. Create or edit `/workspace/scratch/exploit.py` with `apply_patch`.
2. Run it with `exec_command`: `python3 /workspace/scratch/exploit.py`.
1. Create or edit a task-unique script (e.g. `poc_<task-id>.py`, so it can't
clobber a project file or another agent's script) with `apply_patch`.
2. Run it with `exec_command`: `python3 poc_<task-id>.py`.
3. Edit and rerun until the proof-of-concept is reliable.
```
## Installing extra packages
The sandbox's Python lives in `/app/.venv`. To add a one-off dependency
for an exploit script, use `uv` (already in the image and much faster
than pip):
The sandbox's Python lives in `/app/.venv`, and it is the active virtualenv
(`python3` / `pip` already resolve to it). The following common libraries are
**pre-installed** — import them directly, no install step needed:
`requests`, `httpx`, `beautifulsoup4` (`bs4`), `lxml`, `pyjwt` (`jwt`),
`cryptography`.
To add a one-off dependency for an exploit script, use `uv` (already in the
image and much faster than pip):
```bash
uv pip install --python /app/.venv/bin/python <package>
```
Plain `pip install <package>` also works because the venv is active. Install
before you import, so scripts don't fail with `ModuleNotFoundError`.
+15 -4
View File
@@ -229,7 +229,8 @@ async def wait_for_message( # noqa: PLR0911
Use when you have nothing useful to do until a child/peer responds
— typically after spawning subagents and you want to wait for
their completion reports. The agent automatically resumes when any
message arrives.
message arrives, so pick a ``timeout_seconds`` proportional to the
work you're awaiting.
**Critical caveats:**
@@ -246,9 +247,19 @@ async def wait_for_message( # noqa: PLR0911
reason: One-line note shown in graph snapshots while you're
waiting (helps a human or sibling agent debug who's stuck
on what).
timeout_seconds: Hard cap (default 600s). On timeout the tool
returns and you decide whether to keep working or wait
again.
timeout_seconds: Max seconds to wait (default 600). This is only
a cap — the tool returns the INSTANT a message arrives, so a
larger value never makes you wait longer when the reply does
come. Right-size it to what you're waiting on: a short wait
(e.g. 10-60s) for a quick ack or a small/fast subtask, and a
longer one (e.g. ~100-200s) only for genuinely long-running
work (deep recon, exploitation, a full sub-scan). The cap only
bites when the expected message never arrives — so an oversized
timeout on a trivial wait just strands you idle until it
elapses. On timeout the tool returns and you decide whether to
keep working or wait again. (Applies to autonomous multi-agent
runs; in interactive/chat sessions the agent instead parks until
a message arrives and this cap is not enforced.)
"""
inner = _ctx(ctx)
coordinator = coordinator_from_context(inner)
+192 -39
View File
@@ -3,7 +3,9 @@
from __future__ import annotations
import asyncio
import contextlib
import json
import logging
import os
import time
import urllib.request
@@ -21,9 +23,14 @@ from caido_sdk_client.types import (
if TYPE_CHECKING:
from collections.abc import Awaitable, Callable
from caido_sdk_client import Client as CaidoClient
logger = logging.getLogger(__name__)
RequestPart = Literal["request", "response"]
SortBy = Literal[
"timestamp",
@@ -42,6 +49,19 @@ _SITEMAP_PAGE_SIZE = 30
_DEFAULT_CAIDO_URL = "http://127.0.0.1:48080"
_CLIENT_CACHE: dict[str, Client] = {}
_CLIENT_LOCK = asyncio.Lock()
# Substrings that mean the shared client's transport has died or is being used
# concurrently — recoverable by rebuilding the client and retrying once.
_CONNECTION_ERROR_MARKERS = (
"transport is already connected",
"connector is closed",
"server disconnected",
"session is closed",
"cannot write to closing transport",
"connection reset",
"connection closed",
)
_REQ_FIELD_MAP: dict[SortBy, tuple[str, str]] = {
"timestamp": ("req", "created_at"),
"host": ("req", "host"),
@@ -81,19 +101,116 @@ def _login_as_guest() -> str:
return str(payload["data"]["loginAsGuest"]["token"]["accessToken"])
async def get_client() -> Client:
if client := _CLIENT_CACHE.get("default"):
return client
async def _new_client() -> Client:
token = await asyncio.to_thread(_login_as_guest)
client = Client(caido_url(), auth=TokenAuthOptions(token=token))
await client.connect()
_CLIENT_CACHE["default"] = client
return client
async def _safe_aclose(client: Client | None) -> None:
"""Close a (possibly dead) client without letting teardown errors escape."""
if client is None:
return
with contextlib.suppress(Exception):
await client.aclose()
def _is_connection_error(exc: BaseException) -> bool:
message = str(exc).lower()
if any(marker in message for marker in _CONNECTION_ERROR_MARKERS):
return True
cause = exc.__cause__ or exc.__context__
return cause is not None and cause is not exc and _is_connection_error(cause)
async def get_client() -> Client:
"""Return the shared Caido client, creating it under a lock if needed.
The lock prevents two concurrent callers from each building a client and
racing ``connect()`` on the same transport ("Transport is already
connected").
"""
async with _CLIENT_LOCK:
client = _CLIENT_CACHE.get("default")
if client is None:
client = await _new_client()
_CLIENT_CACHE["default"] = client
return client
async def call_with_client[T](
fn: Callable[[Client], Awaitable[T]], *, idempotent: bool = True
) -> T:
"""Run ``fn`` against the shared client, serialized and reconnect-safe.
The Caido GraphQL transport is not safe for concurrent use: two in-flight
requests race and raise "Transport is already connected". All proxy calls
are therefore serialized through ``_CLIENT_LOCK``. If the cached client's
transport has since died ("Connector is closed" / "Server disconnected"),
the stale client is closed and rebuilt so subsequent calls stop failing
against a dead client.
``fn`` is only re-run automatically when ``idempotent`` is true. For
mutations (replay, scope create/update/delete) a connection error may
arrive *after* Caido applied the change, so we heal the client for future
calls but re-raise instead of risking a double-apply.
"""
async with _CLIENT_LOCK:
client = _CLIENT_CACHE.get("default")
if client is None:
client = await _new_client()
_CLIENT_CACHE["default"] = client
try:
return await fn(client)
except Exception as exc:
if not _is_connection_error(exc):
raise
new_client = await _new_client()
_CLIENT_CACHE["default"] = new_client
await _safe_aclose(client)
if not idempotent:
raise
return await fn(new_client)
class SharedCaidoClient:
"""Serialized, reconnect-safe wrapper around one host-side Caido client.
Every agent in a scan shares a single instance (propagated through the
shallow-copied run context). ``call`` serializes access — the SDK transport
is not concurrency-safe — and, when the transport dies, rebuilds the client
via ``reconnect`` (which preserves the Caido project) and closes the dead
one, so a transient Caido restart no longer disables proxy tools for the
rest of the scan.
"""
def __init__(self, client: Client, reconnect: Callable[[], Awaitable[Client]]) -> None:
self._client = client
self._reconnect = reconnect
self._lock = asyncio.Lock()
async def call[T](self, fn: Callable[[Client], Awaitable[T]], *, idempotent: bool = True) -> T:
async with self._lock:
try:
return await fn(self._client)
except Exception as exc:
if not _is_connection_error(exc):
raise
dead, self._client = self._client, await self._reconnect()
await _safe_aclose(dead)
if not idempotent:
raise
return await fn(self._client)
async def aclose(self) -> None:
async with self._lock:
await _safe_aclose(self._client)
async def close_client() -> None:
client = _CLIENT_CACHE.pop("default", None)
async with _CLIENT_LOCK:
client = _CLIENT_CACHE.pop("default", None)
if client is None:
return
await client.aclose()
@@ -385,19 +502,23 @@ async def list_requests(
sort_order: SortOrder = "desc",
scope_id: str | None = None,
) -> Any:
return await list_requests_with_client(
await get_client(),
httpql_filter=httpql_filter,
first=first,
after=after,
sort_by=sort_by,
sort_order=sort_order,
scope_id=scope_id,
return await call_with_client(
lambda client: list_requests_with_client(
client,
httpql_filter=httpql_filter,
first=first,
after=after,
sort_by=sort_by,
sort_order=sort_order,
scope_id=scope_id,
)
)
async def view_request(request_id: str, *, part: RequestPart = "request") -> Any:
return await get_request_with_client(await get_client(), request_id, part=part)
return await call_with_client(
lambda client: get_request_with_client(client, request_id, part=part)
)
async def repeat_request(
@@ -406,22 +527,28 @@ async def repeat_request(
modifications: dict[str, Any] | None = None,
) -> dict[str, Any]:
mods = modifications or {}
result = await get_request_with_client(await get_client(), request_id, part="request")
if result is None or result.request.raw is None:
raise ValueError(f"Request {request_id} not found")
original = result.request
raw_str = result.request.raw.decode("utf-8", errors="replace")
components = parse_raw_request(raw_str)
full_url = full_url_from_components(original, components, mods)
modified = apply_modifications(components, mods, full_url)
connection, raw = build_raw_request(
method=modified["method"],
url=modified["url"],
headers=modified["headers"],
body=modified["body"],
)
return await replay_send_raw(await get_client(), raw=raw, connection=connection)
async def _run(client: CaidoClient) -> dict[str, Any]:
result = await get_request_with_client(client, request_id, part="request")
if result is None or result.request.raw is None:
raise ValueError(f"Request {request_id} not found")
original = result.request
raw_str = result.request.raw.decode("utf-8", errors="replace")
components = parse_raw_request(raw_str)
full_url = full_url_from_components(original, components, mods)
modified = apply_modifications(components, mods, full_url)
connection, raw = build_raw_request(
method=modified["method"],
url=modified["url"],
headers=modified["headers"],
body=modified["body"],
)
return await replay_send_raw(client, raw=raw, connection=connection)
# A replay mutates server state; don't auto-retry if the transport dies
# mid-send (the request may already have been sent).
return await call_with_client(_run, idempotent=False)
async def scope_rules(
@@ -432,7 +559,29 @@ async def scope_rules(
scope_id: str | None = None,
scope_name: str | None = None,
) -> Any:
client = await get_client()
async def _run(client: CaidoClient) -> Any:
return await _scope_rules_with_client(
client,
action,
allowlist=allowlist,
denylist=denylist,
scope_id=scope_id,
scope_name=scope_name,
)
# get/list are read-only and safe to retry; create/update/delete mutate.
return await call_with_client(_run, idempotent=action in {"get", "list"})
async def _scope_rules_with_client(
client: CaidoClient,
action: ScopeAction,
*,
allowlist: list[str] | None = None,
denylist: list[str] | None = None,
scope_id: str | None = None,
scope_name: str | None = None,
) -> Any:
if action == "list":
result = await scope_list(client)
elif action == "get":
@@ -651,26 +800,30 @@ async def list_sitemap(
page: int = 1,
page_size: int = _SITEMAP_PAGE_SIZE,
) -> dict[str, Any]:
return await list_sitemap_with_client(
await get_client(),
scope_id=scope_id,
parent_id=parent_id,
depth=depth,
page=page,
page_size=page_size,
return await call_with_client(
lambda client: list_sitemap_with_client(
client,
scope_id=scope_id,
parent_id=parent_id,
depth=depth,
page=page,
page_size=page_size,
)
)
async def view_sitemap_entry(entry_id: str) -> dict[str, Any]:
return await view_sitemap_entry_with_client(await get_client(), entry_id)
return await call_with_client(lambda client: view_sitemap_entry_with_client(client, entry_id))
__all__ = [
"RequestPart",
"ScopeAction",
"SharedCaidoClient",
"SitemapDepth",
"SortBy",
"SortOrder",
"call_with_client",
"close_client",
"get_client",
"list_requests",
+120 -49
View File
@@ -13,6 +13,7 @@ from typing import TYPE_CHECKING, Any, Literal
from agents import RunContextWrapper, function_tool
from strix.tools.proxy import caido_api
from strix.tools.proxy.caido_api import SharedCaidoClient
logger = logging.getLogger(__name__)
@@ -28,7 +29,7 @@ if TYPE_CHECKING:
SortOrder,
)
else:
from strix.tools.proxy.caido_api import ( # noqa: TC001
from strix.tools.proxy.caido_api import (
RequestPart,
SitemapDepth,
SortBy,
@@ -39,9 +40,18 @@ else:
ScopeAction = Literal["get", "list", "create", "update", "delete"]
def _ctx_client(ctx: RunContextWrapper) -> Client | None:
def _ctx_proxy(ctx: RunContextWrapper) -> SharedCaidoClient | None:
"""Return the scan-wide serialized, reconnect-safe Caido client holder.
All agents in a scan share one :class:`SharedCaidoClient` whose GraphQL
transport is not concurrency-safe (parallel calls raise "Transport is
already connected"). ``SharedCaidoClient.call`` serializes access and
rebuilds the transport if it dies mid-scan. Returns ``None`` when no holder
is present (e.g. standalone tool invocation outside a scan run).
"""
inner = ctx.context if isinstance(ctx.context, dict) else {}
return inner.get("caido_client")
proxy = inner.get("caido_client")
return proxy if isinstance(proxy, SharedCaidoClient) else None
def _to_tool_json(value: Any) -> Any:
@@ -83,6 +93,39 @@ def _err(name: str, exc: Exception) -> str:
)
_HTTPQL_HINT = (
"HTTPQL syntax: quote string values and leave integers unquoted; combine "
"terms with AND / OR (there is no NOT). Numeric fields (resp.code, req.port, "
"id, roundtrip) use eq/ne/gt/gte/lt/lte; text/byte fields (req.host, req.path, "
"req.method, req.raw, resp.raw) use cont/ncont/eq/ne/like/nlike/regex/nregex. "
"Example: 'resp.code.gte:200 AND resp.code.lt:300 AND req.host.cont:\"api\"'."
)
def _is_httpql_error(exc: Exception) -> bool:
message = str(exc).lower()
return "httpql" in message or ("filter" in message and "pars" in message)
def _httpql_error(exc: Exception, httpql_filter: str | None) -> str:
"""Return an actionable error for a rejected HTTPQL filter.
Preserves Caido's exact parser message and echoes the offending query so
the agent can self-correct instead of retrying the same broken filter.
"""
logger.info("list_requests rejected HTTPQL filter %r: %s", httpql_filter, exc)
return json.dumps(
{
"success": False,
"error": f"Invalid HTTPQL filter: {exc}",
"httpql_filter": httpql_filter,
"hint": _HTTPQL_HINT,
},
ensure_ascii=False,
default=str,
)
@function_tool(timeout=120)
async def list_requests(
ctx: RunContextWrapper,
@@ -141,19 +184,21 @@ async def list_requests(
sort_order: ``asc`` or ``desc``.
scope_id: Restrict to a Caido scope (managed via ``scope_rules``).
"""
client = _ctx_client(ctx)
if client is None:
proxy = _ctx_proxy(ctx)
if proxy is None:
return _no_client()
try:
connection = await caido_api.list_requests_with_client(
client,
httpql_filter=httpql_filter,
first=first,
after=after,
sort_by=sort_by,
sort_order=sort_order,
scope_id=scope_id,
connection = await proxy.call(
lambda client: caido_api.list_requests_with_client(
client,
httpql_filter=httpql_filter,
first=first,
after=after,
sort_by=sort_by,
sort_order=sort_order,
scope_id=scope_id,
)
)
entries = []
@@ -207,6 +252,8 @@ async def list_requests(
default=str,
)
except Exception as exc: # noqa: BLE001
if httpql_filter and _is_httpql_error(exc):
return _httpql_error(exc, httpql_filter)
return _err("list_requests", exc)
@@ -244,12 +291,14 @@ async def view_request(
page: 1-indexed page number (only when no ``search_pattern``).
page_size: Lines per page.
"""
client = _ctx_client(ctx)
if client is None:
proxy = _ctx_proxy(ctx)
if proxy is None:
return _no_client()
try:
result = await caido_api.get_request_with_client(client, request_id, part=part)
result = await proxy.call(
lambda client: caido_api.get_request_with_client(client, request_id, part=part)
)
if result is None:
return json.dumps(
{"success": False, "error": f"Request {request_id} not found"},
@@ -359,20 +408,15 @@ async def repeat_request(
- ``body`` replace the body string entirely.
- ``cookies`` dict of cookies to add/update.
"""
client = _ctx_client(ctx)
if client is None:
proxy = _ctx_proxy(ctx)
if proxy is None:
return _no_client()
mods = modifications or {}
try:
async def _do(client: Client) -> dict[str, Any] | None:
result = await caido_api.get_request_with_client(client, request_id, part="request")
if result is None or result.request.raw is None:
return json.dumps(
{"success": False, "error": f"Request {request_id} not found"},
ensure_ascii=False,
default=str,
)
return None
original = result.request
raw_str = result.request.raw.decode("utf-8", errors="replace")
components = caido_api.parse_raw_request(raw_str)
@@ -384,7 +428,18 @@ async def repeat_request(
headers=modified["headers"],
body=modified["body"],
)
replay = await caido_api.replay_send_raw(client, raw=raw, connection=connection)
return await caido_api.replay_send_raw(client, raw=raw, connection=connection)
try:
# A replay mutates target state, so don't auto-retry on a mid-send
# transport failure (the request may already have been sent).
replay = await proxy.call(_do, idempotent=False)
if replay is None:
return json.dumps(
{"success": False, "error": f"Request {request_id} not found"},
ensure_ascii=False,
default=str,
)
return _format_replay_tool_result(replay)
except Exception as exc: # noqa: BLE001
return _err("repeat_request", exc)
@@ -437,16 +492,18 @@ async def list_sitemap(
(recursive subtree). Only meaningful with ``parent_id``.
page: 1-indexed page (30 entries per page).
"""
client = _ctx_client(ctx)
if client is None:
proxy = _ctx_proxy(ctx)
if proxy is None:
return _no_client()
try:
payload = await caido_api.list_sitemap_with_client(
client,
scope_id=scope_id,
parent_id=parent_id,
depth=depth,
page=page,
payload = await proxy.call(
lambda client: caido_api.list_sitemap_with_client(
client,
scope_id=scope_id,
parent_id=parent_id,
depth=depth,
page=page,
)
)
return json.dumps(payload, ensure_ascii=False, default=str)
except Exception as exc: # noqa: BLE001
@@ -468,11 +525,13 @@ async def view_sitemap_entry(
Args:
entry_id: ID from ``list_sitemap`` (or any nested entry).
"""
client = _ctx_client(ctx)
if client is None:
proxy = _ctx_proxy(ctx)
if proxy is None:
return _no_client()
try:
payload = await caido_api.view_sitemap_entry_with_client(client, entry_id)
payload = await proxy.call(
lambda client: caido_api.view_sitemap_entry_with_client(client, entry_id)
)
return json.dumps(payload, ensure_ascii=False, default=str)
except Exception as exc: # noqa: BLE001
return _err("view_sitemap_entry", exc)
@@ -524,13 +583,13 @@ async def scope_rules(
scope_id: Required for ``get`` / ``update`` / ``delete``.
scope_name: Required for ``create`` / ``update``.
"""
client = _ctx_client(ctx)
if client is None:
proxy = _ctx_proxy(ctx)
if proxy is None:
return _no_client()
try:
if action == "list":
scopes = await caido_api.scope_list(client)
scopes = await proxy.call(caido_api.scope_list)
return json.dumps(
{"success": True, "scopes": [_to_tool_json(s) for s in scopes]},
ensure_ascii=False,
@@ -543,9 +602,11 @@ async def scope_rules(
ensure_ascii=False,
default=str,
)
scope = await caido_api.scope_get(client, scope_id)
scope = await proxy.call(lambda client: caido_api.scope_get(client, scope_id))
return json.dumps(
{"success": True, "scope": _to_tool_json(scope)}, ensure_ascii=False, default=str
{"success": True, "scope": _to_tool_json(scope)},
ensure_ascii=False,
default=str,
)
if action == "create":
if not scope_name:
@@ -554,11 +615,16 @@ async def scope_rules(
ensure_ascii=False,
default=str,
)
scope = await caido_api.scope_create(
client, name=scope_name, allowlist=allowlist, denylist=denylist
scope = await proxy.call(
lambda client: caido_api.scope_create(
client, name=scope_name, allowlist=allowlist, denylist=denylist
),
idempotent=False,
)
return json.dumps(
{"success": True, "scope": _to_tool_json(scope)}, ensure_ascii=False, default=str
{"success": True, "scope": _to_tool_json(scope)},
ensure_ascii=False,
default=str,
)
if action == "update":
if not scope_id or not scope_name:
@@ -570,11 +636,16 @@ async def scope_rules(
ensure_ascii=False,
default=str,
)
scope = await caido_api.scope_update(
client, scope_id, name=scope_name, allowlist=allowlist, denylist=denylist
scope = await proxy.call(
lambda client: caido_api.scope_update(
client, scope_id, name=scope_name, allowlist=allowlist, denylist=denylist
),
idempotent=False,
)
return json.dumps(
{"success": True, "scope": _to_tool_json(scope)}, ensure_ascii=False, default=str
{"success": True, "scope": _to_tool_json(scope)},
ensure_ascii=False,
default=str,
)
if not scope_id:
return json.dumps(
@@ -582,7 +653,7 @@ async def scope_rules(
ensure_ascii=False,
default=str,
)
await caido_api.scope_delete(client, scope_id)
await proxy.call(lambda client: caido_api.scope_delete(client, scope_id), idempotent=False)
return json.dumps(
{
"success": True,
+17
View File
@@ -5,6 +5,23 @@ invocation the agent makes (nmap, ffuf, agent-browser, python3, …) goes
through `exec_command`. `write_stdin` streams input to a still-running
process started by an earlier `exec_command` (for interactive prompts).
## `write_stdin` requires a TTY-backed process
`exec_command` runs each command in a fresh **non-interactive** shell (plain
pipes, no TTY) by default. `write_stdin` only works against a process that is
still running **and** was started with a PTY. The canonical sequence is:
```text
exec_command(cmd="python3", tty=true) # start a PTY-backed process
write_stdin(session_id=<id>, chars="print(1)\n")
```
Calling `write_stdin` on a command started with the default `tty=false`, or on
a process that has already exited, fails with
`stdin is not available for this process. Start the command with 'tty=true' in
'exec_command' before using 'write_stdin'.` Use `tty=true` for REPLs,
`ssh`/`nc`/`ftp`, `msfconsole`, or to deliver a Ctrl-C to a long-running job.
- **Implementation:** `agents.sandbox.capabilities.tools.shell_tool.ShellTool`
(in the upstream `agents` SDK)
- **Wired in:** `strix/agents/factory.py` — added per-run via the SDK
+108
View File
@@ -0,0 +1,108 @@
"""Tests for symlink-safe LocalDir staging."""
from __future__ import annotations
from typing import TYPE_CHECKING
from strix.runtime.local_dir_staging import stage_symlink_safe_dir, tree_has_symlink
if TYPE_CHECKING:
from pathlib import Path
def _make_repo(tmp_path: Path) -> Path:
repo = tmp_path / "repo"
(repo / "pkg").mkdir(parents=True)
(repo / "pkg" / "mod.py").write_text("x = 1\n")
(repo / "README.md").write_text("readme\n")
return repo
def test_tree_without_symlinks_used_as_is(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
upload_path, staged = stage_symlink_safe_dir(repo)
assert staged is None
assert upload_path == repo.resolve()
assert not tree_has_symlink(repo)
def test_in_tree_file_symlink_is_dereferenced(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
(repo / "link.py").symlink_to(repo / "pkg" / "mod.py")
upload_path, staged = stage_symlink_safe_dir(repo)
assert staged is not None
assert upload_path == staged
assert not (staged / "link.py").is_symlink()
assert (staged / "link.py").read_text() == "x = 1\n"
assert (staged / "pkg" / "mod.py").read_text() == "x = 1\n"
assert not tree_has_symlink(staged)
def test_in_tree_relative_dir_symlink_is_dereferenced(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
(repo / "pkg_alias").symlink_to("pkg")
_upload, staged = stage_symlink_safe_dir(repo)
assert staged is not None
assert (staged / "pkg_alias" / "mod.py").read_text() == "x = 1\n"
assert not tree_has_symlink(staged)
def test_out_of_tree_symlink_is_dropped(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
outside = tmp_path / "outside.txt"
outside.write_text("secret\n")
(repo / "escape.txt").symlink_to(outside)
(repo / "abs_escape").symlink_to("/etc")
_upload, staged = stage_symlink_safe_dir(repo)
assert staged is not None
assert not (staged / "escape.txt").exists()
assert not (staged / "abs_escape").exists()
assert (staged / "README.md").exists()
def test_dangling_symlink_is_dropped(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
(repo / "dangling").symlink_to(repo / "does-not-exist")
_upload, staged = stage_symlink_safe_dir(repo)
assert staged is not None
assert not (staged / "dangling").exists()
assert not (staged / "dangling").is_symlink()
def test_cyclic_symlink_terminates(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
(repo / "self").symlink_to(repo)
(repo / "pkg" / "up").symlink_to("..")
_upload, staged = stage_symlink_safe_dir(repo)
assert staged is not None
assert (staged / "README.md").exists()
assert not tree_has_symlink(staged)
def test_nested_symlinks_inside_linked_dir(tmp_path: Path) -> None:
repo = _make_repo(tmp_path)
shared = repo / "shared"
shared.mkdir()
(shared / "conf.json").write_text("{}\n")
(shared / "escape").symlink_to("/etc/passwd")
(repo / "pkg" / "shared_link").symlink_to(shared)
_upload, staged = stage_symlink_safe_dir(repo)
assert staged is not None
assert (staged / "pkg" / "shared_link" / "conf.json").read_text() == "{}\n"
assert not (staged / "pkg" / "shared_link" / "escape").exists()
assert not (staged / "shared" / "escape").exists()
+321
View File
@@ -0,0 +1,321 @@
"""Tests for the shared Caido client lifecycle and proxy error handling.
Covers the concurrency/reconnect guarantees of ``caido_api.call_with_client``
(the sandbox-imported path) and ``caido_api.SharedCaidoClient`` (the host-side
holder), plus the actionable HTTPQL errors in ``proxy.tools``.
"""
from __future__ import annotations
import asyncio
import json
from typing import TYPE_CHECKING, Any, cast
import pytest
from strix.tools.proxy import caido_api, tools
from strix.tools.proxy.caido_api import SharedCaidoClient
if TYPE_CHECKING:
from collections.abc import Iterator
class _FakeClient:
def __init__(self, name: str) -> None:
self.name = name
self.closed = False
async def aclose(self) -> None:
self.closed = True
@pytest.fixture(autouse=True)
def _clear_cache() -> Iterator[None]:
caido_api._CLIENT_CACHE.clear()
yield
caido_api._CLIENT_CACHE.clear()
async def test_call_with_client_reuses_cached_client(monkeypatch: pytest.MonkeyPatch) -> None:
cached = _FakeClient("cached")
caido_api._CLIENT_CACHE["default"] = cast("Any", cached)
async def _new() -> Any:
raise AssertionError("_new_client must not run when a client is cached")
monkeypatch.setattr(caido_api, "_new_client", _new)
seen: dict[str, Any] = {}
async def fn(client: Any) -> str:
seen["client"] = client
return "ok"
assert await caido_api.call_with_client(fn) == "ok"
assert seen["client"] is cached
async def test_call_with_client_creates_and_caches_when_empty(
monkeypatch: pytest.MonkeyPatch,
) -> None:
created = _FakeClient("fresh")
async def _new() -> Any:
return created
monkeypatch.setattr(caido_api, "_new_client", _new)
seen: dict[str, Any] = {}
async def fn(client: Any) -> str:
seen["client"] = client
return "ok"
assert await caido_api.call_with_client(fn) == "ok"
assert seen["client"] is created
assert caido_api._CLIENT_CACHE["default"] is created
async def test_failed_init_does_not_poison_cache(monkeypatch: pytest.MonkeyPatch) -> None:
async def _new() -> Any:
raise ConnectionRefusedError("caido not up yet")
monkeypatch.setattr(caido_api, "_new_client", _new)
async def fn(_client: Any) -> str:
return "unreachable"
with pytest.raises(ConnectionRefusedError):
await caido_api.call_with_client(fn)
assert "default" not in caido_api._CLIENT_CACHE
async def test_call_with_client_reconnects_and_closes_dead_transport(
monkeypatch: pytest.MonkeyPatch,
) -> None:
dead = _FakeClient("dead")
fresh = _FakeClient("fresh")
caido_api._CLIENT_CACHE["default"] = cast("Any", dead)
new_calls = {"n": 0}
async def _new() -> Any:
new_calls["n"] += 1
return fresh
monkeypatch.setattr(caido_api, "_new_client", _new)
attempts: list[Any] = []
async def fn(client: Any) -> str:
attempts.append(client)
if len(attempts) == 1:
raise RuntimeError("Transport is already connected")
return "ok"
assert await caido_api.call_with_client(fn) == "ok"
assert attempts == [dead, fresh]
assert new_calls["n"] == 1
assert caido_api._CLIENT_CACHE["default"] is fresh
assert dead.closed is True # stale transport is not leaked
async def test_call_with_client_non_idempotent_rebuilds_but_reraises(
monkeypatch: pytest.MonkeyPatch,
) -> None:
dead = _FakeClient("dead")
fresh = _FakeClient("fresh")
caido_api._CLIENT_CACHE["default"] = cast("Any", dead)
async def _new() -> Any:
return fresh
monkeypatch.setattr(caido_api, "_new_client", _new)
calls = {"n": 0}
async def fn(_client: Any) -> str:
calls["n"] += 1
raise RuntimeError("Server disconnected")
# A mutation must not be auto-retried (it may already have applied), but the
# dead client is still healed so later calls succeed.
with pytest.raises(RuntimeError, match="Server disconnected"):
await caido_api.call_with_client(fn, idempotent=False)
assert calls["n"] == 1
assert caido_api._CLIENT_CACHE["default"] is fresh
assert dead.closed is True
async def test_call_with_client_does_not_retry_application_errors(
monkeypatch: pytest.MonkeyPatch,
) -> None:
cached = _FakeClient("cached")
caido_api._CLIENT_CACHE["default"] = cast("Any", cached)
async def _new() -> Any:
raise AssertionError("deterministic errors must not trigger a reconnect")
monkeypatch.setattr(caido_api, "_new_client", _new)
calls = {"n": 0}
async def fn(_client: Any) -> str:
calls["n"] += 1
raise ValueError("Invalid HTTPQL filter")
with pytest.raises(ValueError, match="Invalid HTTPQL"):
await caido_api.call_with_client(fn)
assert calls["n"] == 1
assert caido_api._CLIENT_CACHE["default"] is cached
async def test_call_with_client_serializes_concurrent_calls(
monkeypatch: pytest.MonkeyPatch,
) -> None:
caido_api._CLIENT_CACHE["default"] = cast("Any", _FakeClient("shared"))
async def _new() -> Any:
raise AssertionError("no reconnect expected")
monkeypatch.setattr(caido_api, "_new_client", _new)
state = {"active": 0, "max": 0}
async def fn(_client: Any) -> str:
state["active"] += 1
state["max"] = max(state["max"], state["active"])
await asyncio.sleep(0.01)
state["active"] -= 1
return "ok"
await asyncio.gather(*(caido_api.call_with_client(fn) for _ in range(6)))
assert state["max"] == 1
async def test_shared_client_reconnects_and_closes_dead_transport() -> None:
dead = _FakeClient("dead")
fresh = _FakeClient("fresh")
async def _reconnect() -> Any:
return fresh
holder = SharedCaidoClient(cast("Any", dead), _reconnect)
attempts: list[Any] = []
async def fn(client: Any) -> str:
attempts.append(client)
if len(attempts) == 1:
raise RuntimeError("Connector is closed")
return "ok"
assert await holder.call(fn) == "ok"
assert attempts == [dead, fresh]
assert dead.closed is True
async def test_shared_client_non_idempotent_rebuilds_but_reraises() -> None:
dead = _FakeClient("dead")
fresh = _FakeClient("fresh")
async def _reconnect() -> Any:
return fresh
holder = SharedCaidoClient(cast("Any", dead), _reconnect)
calls = {"n": 0}
async def fn(_client: Any) -> str:
calls["n"] += 1
raise RuntimeError("Server disconnected")
with pytest.raises(RuntimeError, match="Server disconnected"):
await holder.call(fn, idempotent=False)
assert calls["n"] == 1
assert dead.closed is True
# The healthy client remains for the next call.
assert await holder.call(lambda _c: _ok()) == "ok"
async def _ok() -> str:
return "ok"
async def test_shared_client_serializes_concurrent_calls() -> None:
async def _reconnect() -> Any:
raise AssertionError("no reconnect expected")
holder = SharedCaidoClient(cast("Any", _FakeClient("shared")), _reconnect)
state = {"active": 0, "max": 0}
async def fn(_client: Any) -> str:
state["active"] += 1
state["max"] = max(state["max"], state["active"])
await asyncio.sleep(0.01)
state["active"] -= 1
return "ok"
await asyncio.gather(*(holder.call(fn) for _ in range(6)))
assert state["max"] == 1
async def test_shared_client_passes_through_application_errors() -> None:
async def _reconnect() -> Any:
raise AssertionError("deterministic errors must not trigger a reconnect")
holder = SharedCaidoClient(cast("Any", _FakeClient("c")), _reconnect)
async def fn(_client: Any) -> str:
raise ValueError("Invalid HTTPQL filter")
with pytest.raises(ValueError, match="Invalid HTTPQL"):
await holder.call(fn)
def test_is_connection_error_matches_markers_and_causes() -> None:
assert caido_api._is_connection_error(RuntimeError("Transport is already connected"))
assert caido_api._is_connection_error(RuntimeError("Connector is closed"))
assert caido_api._is_connection_error(RuntimeError("Server disconnected"))
assert not caido_api._is_connection_error(ValueError("Invalid HTTPQL filter"))
nested = RuntimeError("wrapper")
nested.__cause__ = RuntimeError("connection reset by peer")
assert caido_api._is_connection_error(nested)
class _Ctx:
def __init__(self, context: Any) -> None:
self.context = context
def test_ctx_proxy_returns_holder_when_present() -> None:
async def _reconnect() -> Any:
raise AssertionError("unused")
holder = SharedCaidoClient(cast("Any", _FakeClient("c")), _reconnect)
got = tools._ctx_proxy(cast("Any", _Ctx({"caido_client": holder})))
assert got is holder
def test_ctx_proxy_returns_none_without_holder() -> None:
assert tools._ctx_proxy(cast("Any", _Ctx({}))) is None
assert tools._ctx_proxy(cast("Any", _Ctx(None))) is None
assert tools._ctx_proxy(cast("Any", _Ctx({"caido_client": object()}))) is None
def test_is_httpql_error_detection() -> None:
assert tools._is_httpql_error(RuntimeError("HTTPQL parse error at column 4"))
assert tools._is_httpql_error(RuntimeError("failed to parse filter"))
assert not tools._is_httpql_error(RuntimeError("Transport is already connected"))
def test_httpql_error_preserves_message_and_query() -> None:
exc = RuntimeError("HTTPQL parse error: unexpected token at column 12")
payload = json.loads(tools._httpql_error(exc, 'resp.code.eq:"200"'))
assert payload["success"] is False
assert "unexpected token at column 12" in payload["error"]
assert payload["httpql_filter"] == 'resp.code.eq:"200"'
assert "AND / OR" in payload["hint"]
+4 -3
View File
@@ -37,7 +37,8 @@ async def test_persistent_rate_limit_stops_gracefully(
model="openai/gpt-4o",
reasoning_effort="high",
force_required_tool_choice=False,
)
),
runtime=types.SimpleNamespace(max_context_images=3),
)
monkeypatch.setattr(runner, "load_settings", lambda: settings)
monkeypatch.setattr(runner, "configure_sdk_model_defaults", lambda _settings: None)
@@ -54,8 +55,8 @@ async def test_persistent_rate_limit_stops_gracefully(
async def _cleanup(*_args: Any, **_kwargs: Any) -> None:
return None
monkeypatch.setattr(runner.session_manager, "create_or_reuse", _create_or_reuse)
monkeypatch.setattr(runner.session_manager, "cleanup", _cleanup)
monkeypatch.setattr(runner.session_manager, "create_or_reuse", _create_or_reuse) # type: ignore[attr-defined]
monkeypatch.setattr(runner.session_manager, "cleanup", _cleanup) # type: ignore[attr-defined]
monkeypatch.setattr(runner, "build_root_task", lambda _scan_config: "task")
monkeypatch.setattr(runner, "build_scope_context", lambda _scan_config: "")
+24 -4
View File
@@ -18,15 +18,18 @@ def _source(subdir: str, path: str, *, mount: bool = False) -> dict[str, Any]:
def test_copied_source_becomes_localdir_entry(tmp_path: Path) -> None:
entries, bind_mounts = build_session_entries([_source("repo", str(tmp_path))])
entries, bind_mounts, staged_dirs = build_session_entries([_source("repo", str(tmp_path))])
assert bind_mounts == []
assert staged_dirs == []
assert isinstance(entries["repo"], LocalDir)
assert entries["repo"].src == tmp_path.resolve()
def test_mounted_source_becomes_bind_mount(tmp_path: Path) -> None:
entries, bind_mounts = build_session_entries([_source("repo", str(tmp_path), mount=True)])
entries, bind_mounts, _staged = build_session_entries(
[_source("repo", str(tmp_path), mount=True)]
)
assert entries == {}
assert bind_mounts == [
@@ -44,7 +47,7 @@ def test_mixed_sources_split_correctly(tmp_path: Path) -> None:
copied.mkdir()
mounted.mkdir()
entries, bind_mounts = build_session_entries(
entries, bind_mounts, _staged = build_session_entries(
[
_source("copied", str(copied)),
_source("mounted", str(mounted), mount=True),
@@ -57,7 +60,7 @@ def test_mixed_sources_split_correctly(tmp_path: Path) -> None:
def test_incomplete_sources_are_skipped() -> None:
entries, bind_mounts = build_session_entries(
entries, bind_mounts, staged_dirs = build_session_entries(
[
{"source_path": "", "workspace_subdir": "x"},
{"source_path": "/p", "workspace_subdir": ""},
@@ -65,3 +68,20 @@ def test_incomplete_sources_are_skipped() -> None:
)
assert entries == {}
assert bind_mounts == []
assert staged_dirs == []
def test_symlink_tree_is_staged(tmp_path: Path) -> None:
repo = tmp_path / "repo"
repo.mkdir()
(repo / "real.txt").write_text("content")
(repo / "link.txt").symlink_to(repo / "real.txt")
entries, _mounts, staged_dirs = build_session_entries([_source("repo", str(repo))])
assert len(staged_dirs) == 1
entry = entries["repo"]
assert isinstance(entry, LocalDir)
assert entry.src == staged_dirs[0]
assert not (staged_dirs[0] / "link.txt").is_symlink()
assert (staged_dirs[0] / "link.txt").read_text() == "content"