diff --git a/docs/tools/proxy.mdx b/docs/tools/proxy.mdx index 2c920c3..3fc027b 100644 --- a/docs/tools/proxy.mdx +++ b/docs/tools/proxy.mdx @@ -66,6 +66,8 @@ asyncio.run(main()) | `list_requests()` | Query captured traffic with HTTPQL filters | | `view_request()` | Get full request/response details | | `repeat_request()` | Replay a request with modifications | +| `list_sitemap()` | Browse the request-tree view of discovered surface | +| `view_sitemap_entry()` | Inspect one sitemap entry + its related requests | | `scope_rules()` | Manage proxy scope (allowlist/denylist) | For one-off arbitrary requests, use shell tooling like `curl` — the diff --git a/strix/agents/factory.py b/strix/agents/factory.py index 2e6ff5f..5203d5d 100644 --- a/strix/agents/factory.py +++ b/strix/agents/factory.py @@ -46,9 +46,11 @@ from strix.tools.notes.tools import ( ) from strix.tools.proxy.tools import ( list_requests, + list_sitemap, repeat_request, scope_rules, view_request, + view_sitemap_entry, ) from strix.tools.reporting.tool import create_vulnerability_report from strix.tools.thinking.tool import think @@ -268,6 +270,8 @@ _BASE_TOOLS: tuple[Tool, ...] = ( list_requests, view_request, repeat_request, + list_sitemap, + view_sitemap_entry, scope_rules, # Multi-agent graph tools (the coordinator is in ctx.context) view_agent_graph, diff --git a/strix/agents/prompts/system_prompt.jinja b/strix/agents/prompts/system_prompt.jinja index 09dbced..3fbfcc7 100644 --- a/strix/agents/prompts/system_prompt.jinja +++ b/strix/agents/prompts/system_prompt.jinja @@ -171,7 +171,7 @@ EFFICIENCY TACTICS: `python3 -c` or a here-document is acceptable. - For Caido proxy automation inside Python, explicitly import from `caido_api`: - `from caido_api import list_requests, view_request, repeat_request, scope_rules` + `from caido_api import list_requests, view_request, repeat_request, list_sitemap, view_sitemap_entry, scope_rules` - Prefer established fuzzers/scanners where applicable: ffuf, sqlmap, zaproxy, nuclei, wapiti, arjun, httpx, katana, semgrep, bandit, trufflehog, nmap. Use scripts mainly to coordinate or validate around them, not to replace them without reason - For trial-heavy vectors (SQLi, XSS, XXE, SSRF, RCE, auth/JWT, deserialization), DO NOT iterate payloads manually in the browser. Always spray payloads via Python scripts through `exec_command` or terminal tools. - When using established fuzzers/scanners, use the proxy for inspection where helpful diff --git a/strix/interface/tui/renderers/proxy_renderer.py b/strix/interface/tui/renderers/proxy_renderer.py index 5723a57..fe48b71 100644 --- a/strix/interface/tui/renderers/proxy_renderer.py +++ b/strix/interface/tui/renderers/proxy_renderer.py @@ -258,30 +258,26 @@ class RepeatRequestRenderer(BaseToolRenderer): text.append(f"\n {_truncate(modifications, 200)}", style="dim italic") if status == "completed" and isinstance(result, dict): - if "error" in result: + if not result.get("success", True) and result.get("error"): text.append(f"\n error: {_sanitize(str(result['error']), 150)}", style="#ef4444") else: - req = result.get("request", {}) - method = req.get("method", "") - url = req.get("url", "") - code = result.get("status_code") - time_ms = result.get("response_time_ms") - - text.append("\n") - text.append(" >> ", style="#3b82f6") - if method: - text.append(f"{method} ", style="#a78bfa") - if url: - text.append(_truncate(url, 180), style="dim") + elapsed_ms = result.get("elapsed_ms") + response = result.get("response") or {} + code = response.get("status_code") if isinstance(response, dict) else None + body = response.get("body", "") if isinstance(response, dict) else "" + body_truncated = ( + bool(response.get("body_truncated")) if isinstance(response, dict) else False + ) text.append("\n") text.append(" << ", style="#22c55e") if code: text.append(f"{code}", style=_status_style(code)) - if time_ms: - text.append(f" ({time_ms}ms)", style="dim") + else: + text.append("(no response)", style="dim") + if elapsed_ms: + text.append(f" ({elapsed_ms}ms)", style="dim") - body = result.get("body", "") if body and isinstance(body, str): lines = body.split("\n")[:5] for line in lines: @@ -289,7 +285,7 @@ class RepeatRequestRenderer(BaseToolRenderer): text.append(" << ", style="#22c55e") text.append(_truncate(line, MAX_LINE_LENGTH - 5), style="dim") - if len(body.split("\n")) > 5: + if body_truncated or len(body.split("\n")) > 5: text.append("\n") text.append(" ...", style="dim italic") diff --git a/strix/skills/tooling/python.md b/strix/skills/tooling/python.md index b47ff76..d3a577e 100644 --- a/strix/skills/tooling/python.md +++ b/strix/skills/tooling/python.md @@ -19,9 +19,11 @@ explicitly when Python code needs Caido traffic or replay access: ```python from caido_api import ( list_requests, + list_sitemap, repeat_request, scope_rules, view_request, + view_sitemap_entry, ) ``` @@ -59,6 +61,8 @@ Available helpers: - `list_requests(httpql_filter=, first=50, after=, sort_by=, sort_order=, scope_id=)` returns a cursor-paginated Caido SDK `Connection`. - `view_request(request_id, part="request")` returns a Caido SDK request object with raw request/response bytes. - `repeat_request(request_id, modifications={...})` replays a captured request after modifying `url`, `params`, `headers`, `body`, or `cookies`. +- `list_sitemap(scope_id=, parent_id=, depth="DIRECT", page=1)` walks Caido's request-tree view of the discovered surface. Omit `parent_id` for root domains; pass an entry id with `depth="DIRECT"` or `"ALL"` to drill in. +- `view_sitemap_entry(entry_id)` returns one entry plus its 30 most recent related requests. - `scope_rules(action, allowlist=, denylist=, scope_id=, scope_name=)` manages Caido scopes. For one-off arbitrary requests (e.g. probing a fresh endpoint, hitting an diff --git a/strix/tools/proxy/caido_api.py b/strix/tools/proxy/caido_api.py index fd02f8f..0ac4ff3 100644 --- a/strix/tools/proxy/caido_api.py +++ b/strix/tools/proxy/caido_api.py @@ -37,6 +37,8 @@ SortBy = Literal[ ] SortOrder = Literal["asc", "desc"] ScopeAction = Literal["get", "list", "create", "update", "delete"] +SitemapDepth = Literal["DIRECT", "ALL"] +_SITEMAP_PAGE_SIZE = 30 _DEFAULT_CAIDO_URL = "http://127.0.0.1:48080" _CLIENT_CACHE: dict[str, Client] = {} @@ -167,6 +169,53 @@ def build_raw_request( return ConnectionInfoInput(host=host, port=port, is_tls=is_tls), raw +# Cap inline response bodies returned through tool results so a single +# large response (HTML pages, JSON dumps) can't blow out the model's +# context. The model can re-fetch the full body via ``view_request`` +# using the captured request id from ``list_requests`` if it needs more. +_RESPONSE_BODY_MAX_CHARS = 8192 + + +def parse_raw_response(raw_bytes: bytes | None) -> dict[str, Any] | None: + """Parse a raw HTTP response into the same shape ``list_requests`` emits. + + Returns ``None`` when ``raw_bytes`` is missing or unparseable. On + success returns ``{status_code, length, headers, body, body_truncated}`` + where ``body`` is decoded as UTF-8 (replacement chars on invalid + bytes) and clipped at :data:`_RESPONSE_BODY_MAX_CHARS`. + """ + if not raw_bytes: + return None + try: + head, _, body_bytes = raw_bytes.partition(b"\r\n\r\n") + lines = head.decode("iso-8859-1", errors="replace").split("\r\n") + if not lines: + return None + status_parts = lines[0].split(" ", 2) + if len(status_parts) < 2 or not status_parts[1].isdigit(): + return None + status_code = int(status_parts[1]) + headers: dict[str, str] = {} + for line in lines[1:]: + if ":" not in line: + continue + k, v = line.split(":", 1) + headers[k.strip()] = v.strip() + body_text = body_bytes.decode("utf-8", errors="replace") + body_truncated = len(body_text) > _RESPONSE_BODY_MAX_CHARS + if body_truncated: + body_text = body_text[:_RESPONSE_BODY_MAX_CHARS] + return { + "status_code": status_code, + "length": len(body_bytes), + "headers": headers, + "body": body_text, + "body_truncated": body_truncated, + } + except Exception: # noqa: BLE001 - tolerate any malformed raw bytes; None signals "unparseable" to the caller. + return None + + def parse_raw_request(raw_content: str) -> dict[str, Any]: lines = raw_content.split("\n") request_line = lines[0].strip().split(" ") @@ -440,15 +489,223 @@ async def scope_rules( return result +_SITEMAP_ROOTS_QUERY = """ +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 } + } +} +""" + +_SITEMAP_DESCENDANTS_QUERY = """ +query GetSitemapDescendants($parentId: ID!, $depth: SitemapDescendantsDepth!) { + sitemapDescendantEntries(parentId: $parentId, depth: $depth) { + edges { node { + id kind label hasDescendants + request { method path response { statusCode } } + } } + count { value } + } +} +""" + +_SITEMAP_ENTRY_QUERY = """ +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 } + } + } +} +""" + + +def _clean_sitemap_metadata(node: dict[str, Any]) -> dict[str, Any]: + cleaned: dict[str, Any] = { + "id": node["id"], + "kind": node["kind"], + "label": node["label"], + "has_descendants": node["hasDescendants"], + } + meta = node.get("metadata") + if isinstance(meta, dict) and (meta.get("isTls") is not None or meta.get("port")): + meta_out: dict[str, Any] = {} + if meta.get("isTls") is not None: + meta_out["is_tls"] = meta["isTls"] + if meta.get("port"): + meta_out["port"] = meta["port"] + cleaned["metadata"] = meta_out + return cleaned + + +def _clean_sitemap_request_summary(req: dict[str, Any] | None) -> dict[str, Any] | None: + """Same field names as ``list_requests`` emits for a request_summary.""" + if not req: + return None + out: dict[str, Any] = {} + if req.get("method"): + out["method"] = req["method"] + if req.get("path"): + out["path"] = req["path"] + resp = req.get("response") or {} + if resp.get("statusCode"): + out["status_code"] = resp["statusCode"] + return out or None + + +def _clean_sitemap_response(resp: dict[str, Any]) -> dict[str, Any]: + """Same field names as ``list_requests`` emits for a response_summary.""" + out: dict[str, Any] = {} + if resp.get("statusCode"): + out["status_code"] = resp["statusCode"] + if resp.get("length"): + out["length"] = resp["length"] + # Suppress 0 the same way list_requests does — Caido leaves it unset + # on a lot of proxy-captured traffic and a misleading "0ms" is worse + # than the field simply being absent. + if resp.get("roundtripTime"): + out["roundtrip_ms"] = resp["roundtripTime"] + return out + + +async def list_sitemap_with_client( + client: CaidoClient, + *, + scope_id: str | None = None, + parent_id: str | None = None, + depth: SitemapDepth = "DIRECT", + page: int = 1, + page_size: int = _SITEMAP_PAGE_SIZE, +) -> dict[str, Any]: + """Browse Caido's discovered sitemap. Mirrors main-branch shape. + + The Caido GraphQL ``sitemap*Entries`` operations don't support native + pagination, so we fetch all edges for the requested level and slice + client-side. That's fine for typical surface sizes; for very large + sitemaps the caller can drill into ``parent_id`` instead of paging + the root list. + """ + if parent_id: + raw = await client.graphql.query( + _SITEMAP_DESCENDANTS_QUERY, + variables={"parentId": parent_id, "depth": depth}, + ) + data = raw.get("sitemapDescendantEntries") or {} + else: + raw = await client.graphql.query( + _SITEMAP_ROOTS_QUERY, + variables={"scopeId": scope_id}, + ) + data = raw.get("sitemapRootEntries") or {} + + edges = data.get("edges") or [] + total = (data.get("count") or {}).get("value", 0) + skip = max(0, (page - 1) * page_size) + sliced = [edge["node"] for edge in edges[skip : skip + page_size]] + + cleaned: list[dict[str, Any]] = [] + for node in sliced: + entry = _clean_sitemap_metadata(node) + summary = _clean_sitemap_request_summary(node.get("request")) + if summary: + entry["request"] = summary + cleaned.append(entry) + + total_pages = (total + page_size - 1) // page_size if total else 0 + return { + "success": True, + "entries": cleaned, + "page": page, + "page_size": page_size, + "total_pages": total_pages, + "total_count": total, + "has_more": page < total_pages, + } + + +async def view_sitemap_entry_with_client( + client: CaidoClient, + entry_id: str, +) -> dict[str, Any]: + """Fetch one sitemap entry plus its recent related requests.""" + raw = await client.graphql.query(_SITEMAP_ENTRY_QUERY, variables={"id": entry_id}) + entry = raw.get("sitemapEntry") + if not entry: + return {"success": False, "error": f"Sitemap entry {entry_id} not found"} + + cleaned = _clean_sitemap_metadata(entry) + primary = entry.get("request") or {} + if primary: + primary_clean: dict[str, Any] = {} + if primary.get("method"): + primary_clean["method"] = primary["method"] + if primary.get("path"): + primary_clean["path"] = primary["path"] + if primary.get("response"): + primary_clean["response"] = _clean_sitemap_response(primary["response"]) + if primary_clean: + cleaned["request"] = primary_clean + + related = entry.get("requests") or {} + related_edges = related.get("edges") or [] + related_nodes = [edge["node"] for edge in related_edges] + related_clean = [ + summary + for summary in (_clean_sitemap_request_summary(n) for n in related_nodes) + if summary is not None + ] + cleaned["related_requests"] = { + "requests": related_clean, + "total_count": (related.get("count") or {}).get("value", 0), + } + return {"success": True, "entry": cleaned} + + +async def list_sitemap( + *, + scope_id: str | None = None, + parent_id: str | None = None, + depth: SitemapDepth = "DIRECT", + page: int = 1, + page_size: int = _SITEMAP_PAGE_SIZE, +) -> dict[str, Any]: + """Sandbox-Python entry point for sitemap browsing.""" + 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, + ) + + +async def view_sitemap_entry(entry_id: str) -> dict[str, Any]: + """Sandbox-Python entry point for sitemap entry detail.""" + return await view_sitemap_entry_with_client(await get_client(), entry_id) + + __all__ = [ "RequestPart", "ScopeAction", + "SitemapDepth", "SortBy", "SortOrder", "close_client", "get_client", "list_requests", + "list_sitemap", "repeat_request", "scope_rules", "view_request", + "view_sitemap_entry", ] diff --git a/strix/tools/proxy/tools.py b/strix/tools/proxy/tools.py index 8b3465a..f6ab5b3 100644 --- a/strix/tools/proxy/tools.py +++ b/strix/tools/proxy/tools.py @@ -33,7 +33,12 @@ logger = logging.getLogger(__name__) if TYPE_CHECKING: from caido_sdk_client import Client - from strix.tools.proxy.caido_api import RequestPart, SortBy, SortOrder + from strix.tools.proxy.caido_api import ( + RequestPart, + SitemapDepth, + SortBy, + SortOrder, + ) else: # Runtime import: ``function_tool`` resolves the annotations via # ``typing.get_type_hints`` so the Literal aliases must be reachable @@ -41,6 +46,7 @@ else: # used in annotations. from strix.tools.proxy.caido_api import ( # noqa: TC001 RequestPart, + SitemapDepth, SortBy, SortOrder, ) @@ -400,22 +406,97 @@ async def repeat_request( def _format_replay_tool_result(replay: dict[str, Any]) -> str: - response_raw = replay.get("response_raw") - response: dict[str, Any] | None = None - if response_raw is not None: - response = {"raw": response_raw.decode("utf-8", errors="replace")} - return json.dumps( - { - "success": replay["status"] == "DONE", - "status": replay["status"], - "error": replay["error"], - "session_id": replay["session_id"], - "elapsed_ms": replay["elapsed_ms"], - "response": response, - }, - ensure_ascii=False, - default=str, - ) + response = caido_api.parse_raw_response(replay.get("response_raw")) + payload: dict[str, Any] = { + "success": replay["status"] == "DONE", + "status": replay["status"], + "session_id": replay["session_id"], + "elapsed_ms": replay["elapsed_ms"], + "response": response, + } + if replay.get("error"): + payload["error"] = replay["error"] + return json.dumps(payload, ensure_ascii=False, default=str) + + +# ---------------------------------------------------------------------- +# list_sitemap +# ---------------------------------------------------------------------- +@function_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, +) -> str: + """Browse Caido's hierarchical sitemap of proxied traffic. + + Caido aggregates every captured request into a tree: + ``DOMAIN`` → ``DIRECTORY`` (path segments) → ``REQUEST`` → + ``REQUEST_BODY`` / ``REQUEST_QUERY`` (variant per body/query shape). + Use this to understand the discovered attack surface, locate + promising directories, and pick endpoints worth deeper testing. + + Workflow: + - Start with no ``parent_id`` to list root domains (scoped by + ``scope_id`` if you only care about in-scope hosts). + - Pick an entry where ``hasDescendants=true`` and pass its ``id`` + as ``parent_id`` to drill in. ``depth="DIRECT"`` returns only + immediate children; ``"ALL"`` flattens the full subtree. + - Hand any ``id`` to ``view_sitemap_entry`` for the full record + and recent matching requests. + + Args: + scope_id: Limit roots to a Caido scope (only used when + ``parent_id`` is omitted). Manage scopes via ``scope_rules``. + parent_id: Entry ID to expand; omit for root domains. + depth: ``"DIRECT"`` (immediate children) or ``"ALL"`` + (recursive subtree). Only meaningful with ``parent_id``. + page: 1-indexed page (30 entries per page). + """ + client = _ctx_client(ctx) + if client 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, + ) + return json.dumps(payload, ensure_ascii=False, default=str) + except Exception as exc: # noqa: BLE001 + return _err("list_sitemap", exc) + + +# ---------------------------------------------------------------------- +# view_sitemap_entry +# ---------------------------------------------------------------------- +@function_tool(timeout=60) +async def view_sitemap_entry( + ctx: RunContextWrapper, + entry_id: str, +) -> str: + """Get full detail for a sitemap entry plus its recent requests. + + Returns the entry's metadata, the primary request shape + (method/path/response if any), and the most recent 30 related + requests that fall under this entry. Pair with ``list_sitemap`` to + pick the ``entry_id``. + + Args: + entry_id: ID from ``list_sitemap`` (or any nested entry). + """ + client = _ctx_client(ctx) + if client is None: + return _no_client() + try: + payload = await 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) # ----------------------------------------------------------------------