diff --git a/capabilities/web-security/agents/web-security.md b/capabilities/web-security/agents/web-security.md index 0f930bd..8b1d880 100644 --- a/capabilities/web-security/agents/web-security.md +++ b/capabilities/web-security/agents/web-security.md @@ -101,7 +101,7 @@ Any tool that scans, fuzzes, or floods runs on shared local hardware. Cap concur - Use Protoscope when inspecting or crafting protobuf payloads. Prefer the local `protoscope` CLI when it is available on the current `PATH`; use the `protoscope_*` MCP tools as the fallback. - 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). +- 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 `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 3d8b4a5..2e24581 100644 --- a/capabilities/web-security/capability.yaml +++ b/capabilities/web-security/capability.yaml @@ -1,12 +1,13 @@ schema: 1 name: web-security -version: "1.7.0" +version: "1.8.0" description: > Web application penetration testing with 70+ attack technique playbooks covering request smuggling, cache poisoning, SSRF, SSTI, DOM vulnerabilities, authentication bypasses, parser differentials, AEM/Sling exploitation, GraphQL, OAuth, and client-side attacks. - Includes HTTP client tooling with OOB callbacks via interactsh, Caido + Includes HTTP client tooling with OOB callbacks via webhook.site + (API-key aware) and interactsh, Caido integration via MCP, the Python caido-sdk-client, and the caido-mode TypeScript SDK CLI (@caido/sdk-client / caido-ts) for curl-through-Caido testing, match & replace rules, and replay handoffs; Burp proxy diff --git a/capabilities/web-security/tests/test_callback.py b/capabilities/web-security/tests/test_callback.py index 3281331..9b05da0 100644 --- a/capabilities/web-security/tests/test_callback.py +++ b/capabilities/web-security/tests/test_callback.py @@ -41,6 +41,7 @@ _generate_correlation_id, _generate_rsa_keypair, _generate_secret_key, + _resolve_webhook_site_api_key, ) # --------------------------------------------------------------------------- @@ -59,6 +60,7 @@ def client() -> CallbackClient: c._callback_url = None c._provider = None c._token_id = None + c._webhook_api_key = None c._seen_ids = set() c._interactsh_session = None return c @@ -365,6 +367,190 @@ async def test_poll_webhook_site_no_token(self, client: CallbackClient) -> None: result = await client._poll_webhook_site(300) assert "Error" in result +class TestWebhookSiteApiKey: + """webhook.site API-key sourcing, auth headers, and Basic-tier reuse.""" + + def test_resolve_api_key_none(self) -> None: + with patch.dict(os.environ, {}, clear=True): + assert _resolve_webhook_site_api_key() is None + + def test_resolve_api_key_canonical(self) -> None: + with patch.dict(os.environ, {"WEBHOOK_SITE_API_KEY": " key-123 "}, clear=True): + assert _resolve_webhook_site_api_key() == "key-123" + + def test_resolve_api_key_alias_precedence(self) -> None: + env = { + "WEBHOOKSITE_API_KEY": "alias-1", + "WEBHOOK_API_KEY": "alias-2", + } + with patch.dict(os.environ, env, clear=True): + # WEBHOOKSITE_API_KEY has higher priority than WEBHOOK_API_KEY. + assert _resolve_webhook_site_api_key() == "alias-1" + + def test_resolve_api_key_ignores_empty(self) -> None: + env = {"WEBHOOK_SITE_API_KEY": " ", "WEBHOOKSITE_API_KEY": "real"} + with patch.dict(os.environ, env, clear=True): + assert _resolve_webhook_site_api_key() == "real" + + def test_webhook_headers_without_key(self) -> None: + headers = CallbackClient._webhook_headers(None) + assert "Api-Key" not in headers + assert headers["Accept"] == "application/json" + + def test_webhook_headers_with_key(self) -> None: + headers = CallbackClient._webhook_headers("key-abc") + assert headers["Api-Key"] == "key-abc" + + @pytest.mark.asyncio + async def test_register_sends_api_key_header(self, client: CallbackClient) -> None: + mock_resp = _mock_response(201, {"uuid": "auth-uuid"}) + mock_client = AsyncMock() + mock_client.post.return_value = mock_resp + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with ( + patch.dict(os.environ, {"WEBHOOK_SITE_API_KEY": "key-xyz"}, clear=True), + patch("callback.httpx.AsyncClient", return_value=mock_client), + ): + result = await client._register_webhook_site() + + assert result is True + assert client._webhook_api_key == "key-xyz" + assert client._token_id == "auth-uuid" + assert client._callback_url == "https://webhook.site/auth-uuid" + # Verify the Api-Key header was attached to the create call. + _, kwargs = mock_client.post.call_args + assert kwargs["headers"]["Api-Key"] == "key-xyz" + + @pytest.mark.asyncio + async def test_register_reuses_token_when_capped( + self, client: CallbackClient + ) -> None: + """Basic tier caps at 1 token: creation fails, existing token reused.""" + create_resp = _mock_response( + 403, + { + "success": False, + "error": { + "message": "Basic subscriptions can only have 1 URL per account, upgrade to continue" + }, + }, + ) + list_resp = _mock_response( + 200, {"data": [{"uuid": "existing-token"}], "total": 1} + ) + mock_client = AsyncMock() + mock_client.post.return_value = create_resp + mock_client.get.return_value = list_resp + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with ( + patch.dict(os.environ, {"WEBHOOK_SITE_API_KEY": "key-xyz"}, clear=True), + patch("callback.httpx.AsyncClient", return_value=mock_client), + ): + result = await client._register_webhook_site() + + assert result is True + assert client._token_id == "existing-token" + assert client._callback_url == "https://webhook.site/existing-token" + assert client._provider == "webhook_site" + # The token list must have been queried with the Api-Key header. + _, get_kwargs = mock_client.get.call_args + assert get_kwargs["headers"]["Api-Key"] == "key-xyz" + + @pytest.mark.asyncio + async def test_register_no_reuse_without_key(self, client: CallbackClient) -> None: + """Without an API key, a non-201 create must NOT attempt token reuse.""" + create_resp = _mock_response(429, {"error": "rate limited"}) + mock_client = AsyncMock() + mock_client.post.return_value = create_resp + mock_client.get.return_value = _mock_response(200, {"data": []}) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with ( + patch.dict(os.environ, {}, clear=True), + patch("callback.httpx.AsyncClient", return_value=mock_client), + ): + result = await client._register_webhook_site() + + assert result is False + mock_client.get.assert_not_called() + + @pytest.mark.asyncio + async def test_register_capped_no_existing_token( + self, client: CallbackClient + ) -> None: + """API key present, capped, but token list empty -> failure.""" + create_resp = _mock_response(403, {"success": False}) + list_resp = _mock_response(200, {"data": [], "total": 0}) + mock_client = AsyncMock() + mock_client.post.return_value = create_resp + mock_client.get.return_value = list_resp + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with ( + patch.dict(os.environ, {"WEBHOOK_SITE_API_KEY": "key-xyz"}, clear=True), + patch("callback.httpx.AsyncClient", return_value=mock_client), + ): + result = await client._register_webhook_site() + + assert result is False + + @pytest.mark.asyncio + async def test_poll_sends_api_key_header(self, client: CallbackClient) -> None: + client._token_id = "auth-token" + client._provider = "webhook_site" + client._webhook_api_key = "key-poll" + + mock_resp = _mock_response(200, {"data": []}) + mock_client = AsyncMock() + mock_client.get.return_value = mock_resp + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with patch("callback.httpx.AsyncClient", return_value=mock_client): + await client._poll_webhook_site(300) + + _, kwargs = mock_client.get.call_args + assert kwargs["headers"]["Api-Key"] == "key-poll" + + @pytest.mark.asyncio + async def test_get_callback_url_reports_authenticated( + self, client: CallbackClient + ) -> None: + client._callback_url = "https://webhook.site/uuid" + client._provider = "webhook_site" + client._webhook_api_key = "key-abc" + + result = await client.get_callback_url("http") + assert "authenticated" in result + + @pytest.mark.asyncio + async def test_get_callback_url_reports_anonymous( + self, client: CallbackClient + ) -> None: + client._callback_url = "https://webhook.site/uuid" + client._provider = "webhook_site" + client._webhook_api_key = None + + result = await client.get_callback_url("http") + assert "anonymous" in result + + @pytest.mark.asyncio + async def test_reset_clears_api_key(self, client: CallbackClient) -> None: + client._callback_url = "https://webhook.site/uuid" + client._provider = "webhook_site" + client._token_id = "uuid" + client._webhook_api_key = "key-abc" + + await client.reset_callback() + assert client._webhook_api_key is None + + # --------------------------------------------------------------------------- # interactsh API provider tests diff --git a/capabilities/web-security/tools/callback.py b/capabilities/web-security/tools/callback.py index eeeef2d..4123118 100755 --- a/capabilities/web-security/tools/callback.py +++ b/capabilities/web-security/tools/callback.py @@ -3,6 +3,17 @@ Registers callback URLs via webhook.site (primary), interactsh API (secondary), or interactsh-client CLI (fallback) for detecting SSRF, XXE, SSTI, and blind injection vulnerabilities. + +webhook.site now paywalls token creation: anonymous accounts are heavily rate +limited and premium ("Basic") subscriptions cap free-tier accounts at a single +token per account. To keep webhook.site usable this client will pick up an API +key from the environment (``WEBHOOK_SITE_API_KEY``, with ``WEBHOOKSITE_API_KEY`` +and ``WEBHOOK_API_KEY`` accepted as aliases). When a key is present it is sent +as the ``Api-Key`` header on every webhook.site request, and — because paid +Basic accounts are limited to one token — the client transparently reuses the +account's existing token instead of failing when the per-account cap is hit. +The key may be exported in the shell or loaded from a ``.env`` file; it is never +persisted by this toolset. """ from __future__ import annotations @@ -40,6 +51,29 @@ _NONCE_LENGTH = 13 _RSA_KEY_SIZE = 2048 +# webhook.site configuration. +_WEBHOOK_SITE_BASE_URL = "https://webhook.site" +# Environment variables checked (in order) for a webhook.site API key. The +# first non-empty value wins. WEBHOOK_SITE_API_KEY is the canonical name. +_WEBHOOK_SITE_API_KEY_ENVS: tuple[str, ...] = ( + "WEBHOOK_SITE_API_KEY", + "WEBHOOKSITE_API_KEY", + "WEBHOOK_API_KEY", +) + + +def _resolve_webhook_site_api_key() -> str | None: + """Return a webhook.site API key from the environment, if configured. + + Checks the supported env-var names in priority order and returns the first + non-empty, stripped value. Returns None when no key is set. + """ + for env_name in _WEBHOOK_SITE_API_KEY_ENVS: + value = os.environ.get(env_name, "").strip() + if value: + return value + return None + def _generate_correlation_id(length: int = _CORRELATION_ID_LENGTH) -> str: """Generate a random lowercase alphanumeric correlation ID.""" @@ -185,11 +219,18 @@ class CallbackClient(Toolset): Registers with webhook.site (primary), interactsh API (secondary), or interactsh-client CLI (fallback) to provide callback URLs for SSRF, XXE, SSTI, and blind injection testing. + + webhook.site now paywalls token creation. Set ``WEBHOOK_SITE_API_KEY`` (or + the ``WEBHOOKSITE_API_KEY`` / ``WEBHOOK_API_KEY`` aliases) in the environment + or a ``.env`` file to authenticate with a paid account; the key is sent as + the ``Api-Key`` header and, since paid Basic accounts allow only one token, + the existing account token is reused automatically when the cap is reached. """ _callback_url: str | None = PrivateAttr(default=None) _provider: str | None = PrivateAttr(default=None) _token_id: str | None = PrivateAttr(default=None) + _webhook_api_key: str | None = PrivateAttr(default=None) _seen_ids: set[str] = PrivateAttr(default_factory=set) _interactsh_session: _InteractshSession | None = PrivateAttr(default=None) @@ -197,28 +238,84 @@ class CallbackClient(Toolset): # Provider: webhook.site # ------------------------------------------------------------------ + @staticmethod + def _webhook_headers(api_key: str | None) -> dict[str, str]: + """Build request headers for webhook.site, including auth when present.""" + headers = {"Accept": "application/json"} + if api_key: + headers["Api-Key"] = api_key + return headers + async def _register_webhook_site(self) -> bool: - """Register with webhook.site and return True on success.""" + """Register with webhook.site and return True on success. + + When a ``WEBHOOK_SITE_API_KEY`` (or alias) is configured it is sent as + the ``Api-Key`` header so the request is attributed to the paid account. + Paid "Basic" accounts are capped at a single token per account, so if + creation is rejected for exceeding that cap we fall back to reusing the + account's existing token rather than failing outright. + """ + api_key = _resolve_webhook_site_api_key() + self._webhook_api_key = api_key + headers = self._webhook_headers(api_key) try: async with httpx.AsyncClient(timeout=10.0, follow_redirects=True) as client: response = await client.post( - "https://webhook.site/token", + f"{_WEBHOOK_SITE_BASE_URL}/token", json={ "default_content": "OK", "default_status": 200, "default_content_type": "text/plain", }, + headers=headers, ) - if response.status_code != 201: - return False - data = response.json() - token_id = data.get("uuid") - if not token_id: - return False - self._token_id = token_id - self._callback_url = f"https://webhook.site/{token_id}" - self._provider = "webhook_site" - return True + # Successful creation returns 201 with a token uuid. + if response.status_code == 201: + data = response.json() + token_id = data.get("uuid") + if not token_id: + return False + self._set_webhook_token(token_id) + return True + + # With an API key, a Basic subscription caps the account at one + # token. When the cap is hit, reuse the existing token instead. + if api_key and await self._reuse_webhook_site_token(client, headers): + return True + + return False + except Exception: + return False + + def _set_webhook_token(self, token_id: str) -> None: + """Record the active webhook.site token and callback URL.""" + self._token_id = token_id + self._callback_url = f"{_WEBHOOK_SITE_BASE_URL}/{token_id}" + self._provider = "webhook_site" + + async def _reuse_webhook_site_token( + self, client: httpx.AsyncClient, headers: dict[str, str] + ) -> bool: + """Reuse an existing token on an API-key account (Basic tier: 1 token). + + Lists the account's tokens and adopts the most recent one. Returns True + on success, False if none can be found. + """ + try: + resp = await client.get( + f"{_WEBHOOK_SITE_BASE_URL}/tokens", + headers=headers, + ) + if resp.status_code != 200: + return False + body = resp.json() + tokens = body.get("data", []) if isinstance(body, dict) else [] + for entry in tokens: + token_id = entry.get("uuid") + if token_id: + self._set_webhook_token(token_id) + return True + return False except Exception: return False @@ -491,9 +588,14 @@ async def get_callback_url(self, protocol: str = "http") -> str: elif protocol == "dns": url = url.replace("http://", "").replace("https://", "") + provider_note = self._provider + if self._provider == "webhook_site": + auth = "authenticated" if self._webhook_api_key else "anonymous" + provider_note = f"{self._provider} ({auth})" + return ( f"{url}\n\n" - f"Provider: {self._provider}. " + f"Provider: {provider_note}. " f"Inject this URL in payloads, then use check_callbacks to see if the target contacted it." ) @@ -526,10 +628,11 @@ async def _poll_webhook_site(self, since_seconds: int) -> str: return "Error: No webhook.site token." try: - async with httpx.AsyncClient(timeout=10.0) as client: + async with httpx.AsyncClient(timeout=10.0, follow_redirects=True) as client: response = await client.get( - f"https://webhook.site/token/{self._token_id}/requests", + f"{_WEBHOOK_SITE_BASE_URL}/token/{self._token_id}/requests", params={"sorting": "newest"}, + headers=self._webhook_headers(self._webhook_api_key), ) if response.status_code != 200: return f"Error: Poll failed: HTTP {response.status_code}" @@ -619,6 +722,7 @@ async def reset_callback(self) -> str: self._callback_url = None self._provider = None self._token_id = None + self._webhook_api_key = None self._seen_ids.clear() self._interactsh_session = None return "Callback state reset. Next get_callback_url will register a new URL."