From f38cb3d24179118dc76268a8a7432126802d6431 Mon Sep 17 00:00:00 2001 From: GangGreenTemperTatum <104169244+GangGreenTemperTatum@users.noreply.github.com> Date: Thu, 9 Jul 2026 10:32:45 -0400 Subject: [PATCH] feat(web-security): add interactsh API as native callback provider MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a pure-Python interactsh client as a second-priority callback provider, sitting between webhook.site (primary) and interactsh-client CLI (fallback). This enables OOB testing via app.interactsh.com's protocol without requiring the Go binary to be installed. Protocol implementation: - RSA-2048 keypair generation for secure interaction retrieval - POST /register with base64-encoded PEM public key - GET /poll with AES-256-CTR + RSA-OAEP decryption of interactions - POST /deregister for cleanup on reset - Tries all 6 public interactsh servers (oast.pro/live/site/online/fun/me) - Supports INTERACTSH_SERVER env var for custom/self-hosted servers Provider priority: webhook_site → interactsh_api → interactsh_cli Changes: - tools/callback.py: Add _InteractshSession, crypto helpers, register/poll/deregister - tests/test_callback.py: 55 tests covering all 3 providers, crypto round-trip, fallback orchestration, deduplication, and tool method behavior - capability.yaml: Add cryptography>=41.0 dependency Closes CAP-1143 --- capabilities/web-security/capability.yaml | 1 + .../web-security/tests/test_callback.py | 944 ++++++++++++++++++ capabilities/web-security/tools/callback.py | 411 +++++++- 3 files changed, 1345 insertions(+), 11 deletions(-) create mode 100644 capabilities/web-security/tests/test_callback.py diff --git a/capabilities/web-security/capability.yaml b/capabilities/web-security/capability.yaml index b70b42d..a244e28 100644 --- a/capabilities/web-security/capability.yaml +++ b/capabilities/web-security/capability.yaml @@ -107,6 +107,7 @@ dependencies: - "fastmcp>=2.0" - "httpx>=0.28" - "caido-sdk-client" + - "cryptography>=41.0" scripts: - scripts/install_tools.sh diff --git a/capabilities/web-security/tests/test_callback.py b/capabilities/web-security/tests/test_callback.py new file mode 100644 index 0000000..3281331 --- /dev/null +++ b/capabilities/web-security/tests/test_callback.py @@ -0,0 +1,944 @@ +"""Tests for CallbackClient — OOB callback URL toolset. + +Covers webhook.site, interactsh API, and interactsh CLI providers with +mocked HTTP responses. No real network calls are made. +""" + +from __future__ import annotations + +import base64 +import json +import os +import subprocess +import sys +from pathlib import Path +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +# --------------------------------------------------------------------------- +# Path setup & stub installation (must happen before importing the module) +# --------------------------------------------------------------------------- + +_REPO_ROOT = Path(__file__).resolve() +while _REPO_ROOT != _REPO_ROOT.parent: + if (_REPO_ROOT / "capabilities" / "web-security" / "tools").is_dir(): + break + _REPO_ROOT = _REPO_ROOT.parent +sys.path.insert(0, str(_REPO_ROOT / "capabilities" / "web-security" / "tools")) + +# Install the dreadnode stub if not already present +import conftest # noqa: F401 — triggers _install_dreadnode_tools_stub() + +from callback import ( + CallbackClient, + _InteractshSession, + _build_interactsh_url, + _build_registration_payload, + _decrypt_interactsh_message, + _encode_public_key, + _generate_correlation_id, + _generate_rsa_keypair, + _generate_secret_key, +) + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def client() -> CallbackClient: + """Fresh CallbackClient instance with no registered provider. + + Manually initialise Pydantic PrivateAttr defaults because the test-stub + Toolset base class does not run Pydantic's ``__init__``. + """ + c = CallbackClient() + c._callback_url = None + c._provider = None + c._token_id = None + c._seen_ids = set() + c._interactsh_session = None + return c + + +@pytest.fixture +def rsa_keypair() -> tuple[bytes, bytes]: + """Pre-generated RSA keypair (private_der, public_pem).""" + return _generate_rsa_keypair() + + +# --------------------------------------------------------------------------- +# Pure helper tests +# --------------------------------------------------------------------------- + + +class TestHelpers: + def test_generate_correlation_id_default_length(self) -> None: + cid = _generate_correlation_id() + assert len(cid) == 20 + assert cid.isalnum() + assert cid == cid.lower() + + def test_generate_correlation_id_custom_length(self) -> None: + cid = _generate_correlation_id(10) + assert len(cid) == 10 + + def test_generate_secret_key_is_uuid(self) -> None: + import uuid + + sk = _generate_secret_key() + uuid.UUID(sk) # raises ValueError if not a valid UUID + + def test_generate_rsa_keypair(self, rsa_keypair: tuple[bytes, bytes]) -> None: + private_der, public_pem = rsa_keypair + assert isinstance(private_der, bytes) + assert isinstance(public_pem, bytes) + assert b"BEGIN PUBLIC KEY" in public_pem + + def test_encode_public_key(self, rsa_keypair: tuple[bytes, bytes]) -> None: + _, public_pem = rsa_keypair + encoded = _encode_public_key(public_pem) + # Should be valid base64 + decoded = base64.b64decode(encoded) + assert b"BEGIN PUBLIC KEY" in decoded + + def test_build_registration_payload(self, rsa_keypair: tuple[bytes, bytes]) -> None: + _, public_pem = rsa_keypair + payload = _build_registration_payload(public_pem, "secret-123", "corr-id-abc") + assert payload["public-key"] == _encode_public_key(public_pem) + assert payload["secret-key"] == "secret-123" + assert payload["correlation-id"] == "corr-id-abc" + + def test_build_interactsh_url(self) -> None: + url = _build_interactsh_url("abcdefghijklmnopqrst", "oast.fun") + assert url.startswith("abcdefghijklmnopqrst") + assert url.endswith(".oast.fun") + # correlation_id (20) + nonce (13) + "." + "oast.fun" + parts = url.split(".") + assert len(parts[0]) == 33 # 20 + 13 + + def test_interactsh_ts_before_cutoff(self) -> None: + import time + + now = time.time() + old = "2020-01-01T00:00:00Z" + assert CallbackClient._interactsh_ts_before_cutoff(old, now - 1) is True + + future = "2099-01-01T00:00:00Z" + assert CallbackClient._interactsh_ts_before_cutoff(future, now) is False + + def test_interactsh_ts_invalid_returns_false(self) -> None: + assert CallbackClient._interactsh_ts_before_cutoff("not-a-date", 0.0) is False + + def test_parse_interactsh_interaction(self) -> None: + data = { + "protocol": "http", + "timestamp": "2024-01-01T00:00:00Z", + "unique-id": "abc123", + "full-id": "abc123.oast.fun", + "remote-address": "1.2.3.4", + "raw-request": "GET / HTTP/1.1\r\nHost: abc123.oast.fun", + "raw-response": "HTTP/1.1 200 OK", + "q-type": "", + "smtp-from": "", + } + result = CallbackClient._parse_interactsh_interaction(data) + assert result is not None + assert result["protocol"] == "http" + assert result["remote_address"] == "1.2.3.4" + assert result["unique_id"] == "abc123" + + def test_parse_interactsh_interaction_empty_protocol(self) -> None: + assert CallbackClient._parse_interactsh_interaction({}) is None + assert CallbackClient._parse_interactsh_interaction({"protocol": ""}) is None + + +# --------------------------------------------------------------------------- +# Crypto round-trip test +# --------------------------------------------------------------------------- + + +class TestInteractshCrypto: + def test_encrypt_decrypt_round_trip(self, rsa_keypair: tuple[bytes, bytes]) -> None: + """Verify that _decrypt_interactsh_message correctly reverses the + server-side encryption (RSA-OAEP + AES-256-CTR).""" + from cryptography.hazmat.primitives.asymmetric import padding as asym_padding + from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + from cryptography.hazmat.primitives.hashes import SHA256 + from cryptography.hazmat.primitives.serialization import load_pem_public_key + + private_der, public_pem = rsa_keypair + + # Simulate server-side encryption + plaintext = json.dumps( + { + "protocol": "http", + "unique-id": "testid", + "full-id": "testid.oast.fun", + "raw-request": "GET / HTTP/1.1", + "raw-response": "HTTP/1.1 200 OK", + "remote-address": "10.0.0.1", + "timestamp": "2024-06-01T12:00:00Z", + } + ).encode() + + # Generate random AES-256 key and IV + aes_key = os.urandom(32) + iv = os.urandom(16) + + # AES-256-CTR encrypt + cipher = Cipher(algorithms.AES(aes_key), modes.CTR(iv)) + encryptor = cipher.encryptor() + ciphertext = iv + encryptor.update(plaintext) + encryptor.finalize() + ciphertext_b64 = base64.b64encode(ciphertext).decode() + + # RSA-OAEP encrypt the AES key + public_key = load_pem_public_key(public_pem) + encrypted_aes_key = public_key.encrypt( # type: ignore[union-attr] + aes_key, + asym_padding.OAEP( + mgf=asym_padding.MGF1(algorithm=SHA256()), + algorithm=SHA256(), + label=None, + ), + ) + aes_key_b64 = base64.b64encode(encrypted_aes_key).decode() + + # Decrypt using our function + result = _decrypt_interactsh_message(private_der, aes_key_b64, ciphertext_b64) + decoded = json.loads(result) + assert decoded["protocol"] == "http" + assert decoded["unique-id"] == "testid" + assert decoded["remote-address"] == "10.0.0.1" + + +# --------------------------------------------------------------------------- +# webhook.site provider tests +# --------------------------------------------------------------------------- + + +def _mock_response(status_code: int, json_data: Any = None) -> MagicMock: + resp = MagicMock() + resp.status_code = status_code + if json_data is not None: + resp.json.return_value = json_data + return resp + + +class TestWebhookSiteProvider: + @pytest.mark.asyncio + async def test_register_success(self, client: CallbackClient) -> None: + mock_resp = _mock_response(201, {"uuid": "test-uuid-1234"}) + 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("callback.httpx.AsyncClient", return_value=mock_client): + result = await client._register_webhook_site() + + assert result is True + assert client._provider == "webhook_site" + assert client._token_id == "test-uuid-1234" + assert client._callback_url == "https://webhook.site/test-uuid-1234" + + @pytest.mark.asyncio + async def test_register_failure_non_201(self, client: CallbackClient) -> None: + mock_resp = _mock_response(500) + 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("callback.httpx.AsyncClient", return_value=mock_client): + result = await client._register_webhook_site() + + assert result is False + + @pytest.mark.asyncio + async def test_register_failure_no_uuid(self, client: CallbackClient) -> None: + mock_resp = _mock_response(201, {}) + 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("callback.httpx.AsyncClient", return_value=mock_client): + result = await client._register_webhook_site() + + assert result is False + + @pytest.mark.asyncio + async def test_register_failure_exception(self, client: CallbackClient) -> None: + with patch( + "callback.httpx.AsyncClient", side_effect=Exception("network error") + ): + result = await client._register_webhook_site() + + assert result is False + + @pytest.mark.asyncio + async def test_poll_webhook_site_no_interactions( + self, client: CallbackClient + ) -> None: + client._token_id = "test-token" + client._provider = "webhook_site" + + 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): + result = await client._poll_webhook_site(300) + + assert "No callback interactions received yet" in result + + @pytest.mark.asyncio + async def test_poll_webhook_site_with_interactions( + self, client: CallbackClient + ) -> None: + client._token_id = "test-token" + client._provider = "webhook_site" + + mock_resp = _mock_response( + 200, + { + "data": [ + { + "uuid": "req-1", + "method": "GET", + "url": "https://webhook.site/test-token?test=1", + "ip": "10.0.0.1", + "created_at": "2099-01-01T00:00:00Z", + "content": "", + "headers": {"Host": "webhook.site"}, + } + ] + }, + ) + 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): + result = await client._poll_webhook_site(300) + + assert "1 callback interactions" in result + assert "10.0.0.1" in result + + @pytest.mark.asyncio + async def test_poll_webhook_site_deduplicates(self, client: CallbackClient) -> None: + client._token_id = "test-token" + client._provider = "webhook_site" + + interaction = { + "uuid": "req-1", + "method": "GET", + "url": "https://webhook.site/test-token", + "ip": "10.0.0.1", + "created_at": "2099-01-01T00:00:00Z", + "content": "", + "headers": {}, + } + + mock_resp = _mock_response(200, {"data": [interaction]}) + 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): + result1 = await client._poll_webhook_site(300) + result2 = await client._poll_webhook_site(300) + + assert "1 callback interactions" in result1 + assert "No new callback interactions" in result2 + + @pytest.mark.asyncio + async def test_poll_webhook_site_no_token(self, client: CallbackClient) -> None: + result = await client._poll_webhook_site(300) + assert "Error" in result + + +# --------------------------------------------------------------------------- +# interactsh API provider tests +# --------------------------------------------------------------------------- + + +class TestInteractshApiProvider: + @pytest.mark.asyncio + async def test_register_success(self, client: CallbackClient) -> None: + mock_resp = _mock_response(200, {"message": "registration successful"}) + 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("callback.httpx.AsyncClient", return_value=mock_client): + result = await client._register_interactsh_api() + + assert result is True + assert client._provider == "interactsh_api" + assert client._interactsh_session is not None + assert client._callback_url is not None + assert ".oast.pro" in client._callback_url + + @pytest.mark.asyncio + async def test_register_success_custom_server(self, client: CallbackClient) -> None: + mock_resp = _mock_response(200, {"message": "registration successful"}) + 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("callback.httpx.AsyncClient", return_value=mock_client), + patch.dict(os.environ, {"INTERACTSH_SERVER": "my.custom.server"}), + ): + result = await client._register_interactsh_api() + + assert result is True + assert client._interactsh_session is not None + assert client._interactsh_session.server_host == "my.custom.server" + + @pytest.mark.asyncio + async def test_register_failure_all_servers(self, client: CallbackClient) -> None: + mock_resp = _mock_response(500) + 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("callback.httpx.AsyncClient", return_value=mock_client): + result = await client._register_interactsh_api() + + assert result is False + + @pytest.mark.asyncio + async def test_register_failure_wrong_message(self, client: CallbackClient) -> None: + mock_resp = _mock_response(200, {"message": "something else"}) + 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("callback.httpx.AsyncClient", return_value=mock_client): + result = await client._register_interactsh_api() + + assert result is False + + @pytest.mark.asyncio + async def test_register_failure_no_cryptography( + self, client: CallbackClient + ) -> None: + """Gracefully return False when cryptography is not importable.""" + with patch.dict( + sys.modules, {"cryptography.hazmat.primitives.asymmetric.rsa": None} + ): + # Force an ImportError by patching the import + original_import = ( + __builtins__.__import__ + if hasattr(__builtins__, "__import__") + else __import__ + ) # type: ignore[union-attr] + + def mock_import(name: str, *args: Any, **kwargs: Any) -> Any: + if "cryptography" in name: + raise ImportError("no cryptography") + return original_import(name, *args, **kwargs) + + with patch("builtins.__import__", side_effect=mock_import): + result = await client._register_interactsh_api() + + assert result is False + + @pytest.mark.asyncio + async def test_poll_no_session(self, client: CallbackClient) -> None: + result = await client._poll_interactsh_api(300) + assert "Error" in result + + @pytest.mark.asyncio + async def test_poll_empty_response(self, client: CallbackClient) -> None: + session = _InteractshSession( + server_url="https://oast.fun", + server_host="oast.fun", + correlation_id="testcorrelationid01", + secret_key="secret", + private_key_der=b"unused", + ) + client._interactsh_session = session + + mock_resp = _mock_response(200, {"data": [], "aes_key": "", "extra": []}) + 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): + result = await client._poll_interactsh_api(300) + + assert "No new callback interactions" in result + + @pytest.mark.asyncio + async def test_poll_with_extra_unencrypted(self, client: CallbackClient) -> None: + """Test polling with unencrypted 'extra' data (no crypto needed).""" + session = _InteractshSession( + server_url="https://oast.fun", + server_host="oast.fun", + correlation_id="testcorrelationid01", + secret_key="secret", + private_key_der=b"unused", + ) + client._interactsh_session = session + + extra_interaction = json.dumps( + { + "protocol": "dns", + "unique-id": "testcorrelationid01abc", + "full-id": "testcorrelationid01abc.oast.fun", + "remote-address": "8.8.8.8", + "timestamp": "2099-01-01T12:00:00Z", + "raw-request": "DNS A query", + "raw-response": "", + "q-type": "A", + } + ) + + mock_resp = _mock_response( + 200, + {"data": [], "aes_key": "", "extra": [extra_interaction]}, + ) + 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): + result = await client._poll_interactsh_api(300) + + assert "1 callback interactions" in result + assert "DNS" in result + assert "8.8.8.8" in result + + @pytest.mark.asyncio + async def test_poll_with_encrypted_data(self, client: CallbackClient) -> None: + """Full crypto round-trip: simulate server encrypting, client decrypting.""" + private_der, public_pem = _generate_rsa_keypair() + + session = _InteractshSession( + server_url="https://oast.pro", + server_host="oast.pro", + correlation_id="testcorrelationid01", + secret_key="secret-key", + private_key_der=private_der, + ) + client._interactsh_session = session + + # Server-side encryption + from cryptography.hazmat.primitives.asymmetric import padding as asym_padding + from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + from cryptography.hazmat.primitives.hashes import SHA256 + from cryptography.hazmat.primitives.serialization import load_pem_public_key + + interaction_json = json.dumps( + { + "protocol": "http", + "unique-id": "testcorrelationid01nonce", + "full-id": "testcorrelationid01nonce.oast.pro", + "remote-address": "192.168.1.1", + "timestamp": "2099-06-01T12:00:00Z", + "raw-request": "GET /ssrf?target=internal HTTP/1.1\r\nHost: testcorrelationid01nonce.oast.pro", + "raw-response": "HTTP/1.1 200 OK", + "q-type": "", + "smtp-from": "", + } + ).encode() + + aes_key = os.urandom(32) + iv = os.urandom(16) + cipher = Cipher(algorithms.AES(aes_key), modes.CTR(iv)) + encryptor = cipher.encryptor() + ciphertext = iv + encryptor.update(interaction_json) + encryptor.finalize() + ciphertext_b64 = base64.b64encode(ciphertext).decode() + + public_key = load_pem_public_key(public_pem) + encrypted_aes_key = public_key.encrypt( # type: ignore[union-attr] + aes_key, + asym_padding.OAEP( + mgf=asym_padding.MGF1(algorithm=SHA256()), + algorithm=SHA256(), + label=None, + ), + ) + aes_key_b64 = base64.b64encode(encrypted_aes_key).decode() + + mock_resp = _mock_response( + 200, + {"data": [ciphertext_b64], "aes_key": aes_key_b64, "extra": []}, + ) + 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): + result = await client._poll_interactsh_api(300) + + assert "1 callback interactions" in result + assert "HTTP" in result + assert "192.168.1.1" in result + + @pytest.mark.asyncio + async def test_poll_deduplicates(self, client: CallbackClient) -> None: + session = _InteractshSession( + server_url="https://oast.fun", + server_host="oast.fun", + correlation_id="testcorrelationid01", + secret_key="secret", + private_key_der=b"unused", + ) + client._interactsh_session = session + + extra = json.dumps( + { + "protocol": "dns", + "unique-id": "x", + "full-id": "x.oast.fun", + "remote-address": "1.1.1.1", + "timestamp": "2099-01-01T12:00:00Z", + "raw-request": "", + "raw-response": "", + } + ) + + mock_resp = _mock_response(200, {"data": [], "aes_key": "", "extra": [extra]}) + 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): + result1 = await client._poll_interactsh_api(300) + result2 = await client._poll_interactsh_api(300) + + assert "1 callback interactions" in result1 + assert "No new callback interactions" in result2 + + @pytest.mark.asyncio + async def test_deregister_best_effort(self, client: CallbackClient) -> None: + session = _InteractshSession( + server_url="https://oast.fun", + server_host="oast.fun", + correlation_id="test", + secret_key="secret", + private_key_der=b"", + ) + client._interactsh_session = session + + mock_client = AsyncMock() + mock_client.post.return_value = _mock_response(200, {}) + 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._deregister_interactsh_api() + + mock_client.post.assert_called_once() + call_kwargs = mock_client.post.call_args + assert "/deregister" in call_kwargs.args[0] + + @pytest.mark.asyncio + async def test_deregister_no_session(self, client: CallbackClient) -> None: + """Deregister with no active session should be a no-op.""" + await client._deregister_interactsh_api() # Should not raise + + @pytest.mark.asyncio + async def test_deregister_handles_exception(self, client: CallbackClient) -> None: + session = _InteractshSession( + server_url="https://oast.fun", + server_host="oast.fun", + correlation_id="test", + secret_key="secret", + private_key_der=b"", + ) + client._interactsh_session = session + + with patch("callback.httpx.AsyncClient", side_effect=Exception("network")): + await client._deregister_interactsh_api() # Should not raise + + +# --------------------------------------------------------------------------- +# interactsh CLI provider tests +# --------------------------------------------------------------------------- + + +class TestInteractshCliProvider: + def test_register_cli_json_url(self, client: CallbackClient) -> None: + mock_result = MagicMock() + mock_result.stdout = '{"url": "https://abc.oast.fun"}\n' + mock_result.returncode = 0 + + with patch("callback.subprocess.run", return_value=mock_result): + result = client._register_interactsh_cli() + + assert result is True + assert client._provider == "interactsh_cli" + assert client._callback_url == "https://abc.oast.fun" + + def test_register_cli_plain_text(self, client: CallbackClient) -> None: + mock_result = MagicMock() + mock_result.stdout = "abc123.oast.fun\n" + mock_result.returncode = 0 + + with patch("callback.subprocess.run", return_value=mock_result): + result = client._register_interactsh_cli() + + assert result is True + assert client._provider == "interactsh_cli" + assert client._callback_url == "https://abc123.oast.fun" + + def test_register_cli_interact_domain(self, client: CallbackClient) -> None: + mock_result = MagicMock() + mock_result.stdout = "xyz.interact.sh\n" + mock_result.returncode = 0 + + with patch("callback.subprocess.run", return_value=mock_result): + result = client._register_interactsh_cli() + + assert result is True + assert client._callback_url == "https://xyz.interact.sh" + + def test_register_cli_not_installed(self, client: CallbackClient) -> None: + with patch("callback.subprocess.run", side_effect=FileNotFoundError): + result = client._register_interactsh_cli() + + assert result is False + + def test_register_cli_timeout(self, client: CallbackClient) -> None: + with patch( + "callback.subprocess.run", side_effect=subprocess.TimeoutExpired("cmd", 15) + ): + result = client._register_interactsh_cli() + + assert result is False + + def test_register_cli_no_url(self, client: CallbackClient) -> None: + mock_result = MagicMock() + mock_result.stdout = "some random output\n" + mock_result.returncode = 0 + + with patch("callback.subprocess.run", return_value=mock_result): + result = client._register_interactsh_cli() + + assert result is False + + +# --------------------------------------------------------------------------- +# Registration orchestrator tests +# --------------------------------------------------------------------------- + + +class TestEnsureRegistered: + @pytest.mark.asyncio + async def test_already_registered(self, client: CallbackClient) -> None: + client._callback_url = "https://already.registered" + result = await client._ensure_registered() + assert result is True + + @pytest.mark.asyncio + async def test_webhook_site_first(self, client: CallbackClient) -> None: + with patch.object( + client, "_register_webhook_site", return_value=True + ) as mock_ws: + result = await client._ensure_registered() + + assert result is True + mock_ws.assert_called_once() + + @pytest.mark.asyncio + async def test_falls_through_to_interactsh_api( + self, client: CallbackClient + ) -> None: + with ( + patch.object(client, "_register_webhook_site", return_value=False), + patch.object( + client, "_register_interactsh_api", return_value=True + ) as mock_api, + ): + result = await client._ensure_registered() + + assert result is True + mock_api.assert_called_once() + + @pytest.mark.asyncio + async def test_falls_through_to_cli(self, client: CallbackClient) -> None: + with ( + patch.object(client, "_register_webhook_site", return_value=False), + patch.object(client, "_register_interactsh_api", return_value=False), + patch.object( + client, "_register_interactsh_cli", return_value=True + ) as mock_cli, + ): + result = await client._ensure_registered() + + assert result is True + mock_cli.assert_called_once() + + @pytest.mark.asyncio + async def test_all_fail(self, client: CallbackClient) -> None: + with ( + patch.object(client, "_register_webhook_site", return_value=False), + patch.object(client, "_register_interactsh_api", return_value=False), + patch.object(client, "_register_interactsh_cli", return_value=False), + ): + result = await client._ensure_registered() + + assert result is False + + +# --------------------------------------------------------------------------- +# Tool method tests (get_callback_url, check_callbacks, reset_callback) +# --------------------------------------------------------------------------- + + +class TestToolMethods: + @pytest.mark.asyncio + async def test_get_callback_url_http(self, client: CallbackClient) -> None: + client._callback_url = "https://webhook.site/uuid" + client._provider = "webhook_site" + + result = await client.get_callback_url("http") + assert "https://webhook.site/uuid" in result + assert "webhook_site" in result + + @pytest.mark.asyncio + async def test_get_callback_url_https(self, client: CallbackClient) -> None: + client._callback_url = "http://example.com/cb" + client._provider = "test" + + result = await client.get_callback_url("https") + assert "https://example.com/cb" in result + + @pytest.mark.asyncio + async def test_get_callback_url_dns(self, client: CallbackClient) -> None: + client._callback_url = "https://abc.oast.fun" + client._provider = "interactsh_api" + + result = await client.get_callback_url("dns") + assert "abc.oast.fun" in result + assert "https://" not in result.split("\n")[0] + + @pytest.mark.asyncio + async def test_get_callback_url_no_provider(self, client: CallbackClient) -> None: + with ( + patch.object(client, "_register_webhook_site", return_value=False), + patch.object(client, "_register_interactsh_api", return_value=False), + patch.object(client, "_register_interactsh_cli", return_value=False), + ): + result = await client.get_callback_url() + + assert "Error" in result + + @pytest.mark.asyncio + async def test_check_callbacks_no_url(self, client: CallbackClient) -> None: + result = await client.check_callbacks() + assert "Error" in result + + @pytest.mark.asyncio + async def test_check_callbacks_webhook_site(self, client: CallbackClient) -> None: + client._callback_url = "https://webhook.site/uuid" + client._provider = "webhook_site" + client._token_id = "uuid" + + 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): + result = await client.check_callbacks() + + assert "No callback interactions" in result + + @pytest.mark.asyncio + async def test_check_callbacks_interactsh_api(self, client: CallbackClient) -> None: + client._callback_url = "https://abc.oast.fun" + client._provider = "interactsh_api" + client._interactsh_session = _InteractshSession( + server_url="https://oast.fun", + server_host="oast.fun", + correlation_id="test", + secret_key="secret", + private_key_der=b"", + ) + + mock_resp = _mock_response(200, {"data": [], "aes_key": "", "extra": []}) + 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): + result = await client.check_callbacks() + + assert "No new callback interactions" in result + + @pytest.mark.asyncio + async def test_check_callbacks_interactsh_cli(self, client: CallbackClient) -> None: + client._callback_url = "https://abc.oast.fun" + client._provider = "interactsh_cli" + + result = await client.check_callbacks() + assert "interactsh" in result.lower() + assert "bash" in result.lower() + + @pytest.mark.asyncio + async def test_check_callbacks_unknown_provider( + self, client: CallbackClient + ) -> None: + client._callback_url = "https://example.com" + client._provider = "unknown_provider" + + result = await client.check_callbacks() + assert "Error" in result + + @pytest.mark.asyncio + async def test_reset_callback(self, client: CallbackClient) -> None: + client._callback_url = "https://abc.oast.fun" + client._provider = "interactsh_api" + client._interactsh_session = _InteractshSession( + server_url="https://oast.fun", + server_host="oast.fun", + correlation_id="test", + secret_key="secret", + private_key_der=b"", + ) + client._seen_ids.add("seen-1") + + with patch.object(client, "_deregister_interactsh_api", new_callable=AsyncMock): + result = await client.reset_callback() + + assert "reset" in result.lower() + assert client._callback_url is None + assert client._provider is None + assert client._interactsh_session is None + assert len(client._seen_ids) == 0 + + @pytest.mark.asyncio + async def test_reset_callback_webhook_site(self, client: CallbackClient) -> None: + """Reset with webhook_site provider should not call deregister.""" + client._callback_url = "https://webhook.site/uuid" + client._provider = "webhook_site" + client._token_id = "uuid" + + result = await client.reset_callback() + assert "reset" in result.lower() + assert client._callback_url is None + assert client._token_id is None diff --git a/capabilities/web-security/tools/callback.py b/capabilities/web-security/tools/callback.py index 4bcd88b..eeeef2d 100755 --- a/capabilities/web-security/tools/callback.py +++ b/capabilities/web-security/tools/callback.py @@ -1,33 +1,201 @@ """Callback client for out-of-band vulnerability testing. -Registers callback URLs via webhook.site (primary) or interactsh-client CLI -(fallback) for detecting SSRF, XXE, SSTI, and blind injection vulnerabilities. +Registers callback URLs via webhook.site (primary), interactsh API (secondary), +or interactsh-client CLI (fallback) for detecting SSRF, XXE, SSTI, and blind +injection vulnerabilities. """ from __future__ import annotations +import base64 import json +import os +import secrets +import string import subprocess import time -from datetime import datetime, timezone +from datetime import datetime from urllib.parse import urlparse import httpx from dreadnode.agents.tools import Toolset, tool_method from pydantic import PrivateAttr +# --------------------------------------------------------------------------- +# Interactsh protocol helpers (pure functions, no side effects) +# --------------------------------------------------------------------------- + +# Public interactsh servers in priority order. +_INTERACTSH_SERVERS: list[str] = [ + "oast.pro", + "oast.live", + "oast.site", + "oast.online", + "oast.fun", + "oast.me", +] + +# Default lengths matching the upstream Go client. +_CORRELATION_ID_LENGTH = 20 +_NONCE_LENGTH = 13 +_RSA_KEY_SIZE = 2048 + + +def _generate_correlation_id(length: int = _CORRELATION_ID_LENGTH) -> str: + """Generate a random lowercase alphanumeric correlation ID.""" + alphabet = string.ascii_lowercase + string.digits + return "".join(secrets.choice(alphabet) for _ in range(length)) + + +def _generate_secret_key() -> str: + """Generate a UUID-style secret key.""" + import uuid + + return str(uuid.uuid4()) + + +def _generate_rsa_keypair() -> tuple[bytes, bytes]: + """Generate an RSA keypair and return (private_key_der, public_key_pem). + + Returns the private key in PKCS1/DER form (kept in memory only) and the + public key as a PEM-encoded ``RSA PUBLIC KEY`` block. + """ + from cryptography.hazmat.primitives.asymmetric import rsa + from cryptography.hazmat.primitives.serialization import ( + Encoding, + NoEncryption, + PrivateFormat, + PublicFormat, + ) + + private_key = rsa.generate_private_key( + public_exponent=65537, key_size=_RSA_KEY_SIZE + ) + private_der = private_key.private_bytes( + Encoding.DER, PrivateFormat.PKCS8, NoEncryption() + ) + public_pem = private_key.public_key().public_bytes( + Encoding.PEM, PublicFormat.SubjectPublicKeyInfo + ) + return private_der, public_pem + + +def _encode_public_key(public_pem: bytes) -> str: + """Base64-encode a PEM public key for the registration payload.""" + return base64.b64encode(public_pem).decode() + + +def _build_registration_payload( + public_pem: bytes, + secret_key: str, + correlation_id: str, +) -> dict[str, str]: + """Build the JSON registration payload for POST /register.""" + return { + "public-key": _encode_public_key(public_pem), + "secret-key": secret_key, + "correlation-id": correlation_id, + } + + +def _build_interactsh_url(correlation_id: str, server_host: str) -> str: + """Build a unique interactsh callback URL.""" + nonce = _generate_correlation_id(_NONCE_LENGTH) + return f"{correlation_id}{nonce}.{server_host}" + + +def _decrypt_interactsh_message( + private_key_der: bytes, + aes_key_b64: str, + ciphertext_b64: str, +) -> bytes: + """Decrypt an AES-encrypted interaction record. + + The server returns: + - ``aes_key``: RSA-OAEP(SHA-256) encrypted AES-256 key, base64-encoded. + - ``data[]``: AES-256-CTR encrypted interaction JSON, base64-encoded. + The first 16 bytes of the ciphertext are the IV. + """ + from cryptography.hazmat.primitives.asymmetric import padding as asym_padding + from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + from cryptography.hazmat.primitives.hashes import SHA256 + from cryptography.hazmat.primitives.serialization import load_der_private_key + + private_key = load_der_private_key(private_key_der, password=None) + + # Decrypt the AES key with RSA-OAEP + SHA-256 + encrypted_aes_key = base64.b64decode(aes_key_b64) + aes_key = private_key.decrypt( # type: ignore[union-attr] + encrypted_aes_key, + asym_padding.OAEP( + mgf=asym_padding.MGF1(algorithm=SHA256()), + algorithm=SHA256(), + label=None, + ), + ) + + # Decrypt the message with AES-256-CTR + ciphertext = base64.b64decode(ciphertext_b64) + iv = ciphertext[:16] + cipher = Cipher(algorithms.AES(aes_key), modes.CTR(iv)) + decryptor = cipher.decryptor() + plaintext = decryptor.update(ciphertext[16:]) + decryptor.finalize() + return plaintext.rstrip(b" \t\r\n") + + +# --------------------------------------------------------------------------- +# Interactsh session state +# --------------------------------------------------------------------------- + + +class _InteractshSession: + """Holds cryptographic and server state for an active interactsh session.""" + + __slots__ = ( + "server_url", + "server_host", + "correlation_id", + "secret_key", + "private_key_der", + ) + + def __init__( + self, + server_url: str, + server_host: str, + correlation_id: str, + secret_key: str, + private_key_der: bytes, + ) -> None: + self.server_url = server_url + self.server_host = server_host + self.correlation_id = correlation_id + self.secret_key = secret_key + self.private_key_der = private_key_der + + +# --------------------------------------------------------------------------- +# CallbackClient toolset +# --------------------------------------------------------------------------- + class CallbackClient(Toolset): """OOB vulnerability testing via callback URLs. - Registers with webhook.site (primary) or interactsh (fallback) to provide - callback URLs for SSRF, XXE, SSTI, and blind injection testing. + 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. """ _callback_url: str | None = PrivateAttr(default=None) _provider: str | None = PrivateAttr(default=None) _token_id: str | None = PrivateAttr(default=None) _seen_ids: set[str] = PrivateAttr(default_factory=set) + _interactsh_session: _InteractshSession | None = PrivateAttr(default=None) + + # ------------------------------------------------------------------ + # Provider: webhook.site + # ------------------------------------------------------------------ async def _register_webhook_site(self) -> bool: """Register with webhook.site and return True on success.""" @@ -54,7 +222,209 @@ async def _register_webhook_site(self) -> bool: except Exception: return False - def _register_interactsh(self) -> bool: + # ------------------------------------------------------------------ + # Provider: interactsh API (native Python, no CLI binary required) + # ------------------------------------------------------------------ + + async def _register_interactsh_api(self) -> bool: + """Register with an interactsh server via the HTTP API. + + Tries each public interactsh server in order until one succeeds. + Returns True on success, False if all servers fail. + """ + try: + from cryptography.hazmat.primitives.asymmetric import rsa as _rsa # noqa: F401 + except ImportError: + # cryptography not installed — skip this provider + return False + + private_der, public_pem = _generate_rsa_keypair() + correlation_id = _generate_correlation_id() + secret_key = _generate_secret_key() + payload = _build_registration_payload(public_pem, secret_key, correlation_id) + + # Allow overriding the server list via environment variable. + env_servers = os.environ.get("INTERACTSH_SERVER", "") + if env_servers: + servers = [s.strip() for s in env_servers.split(",") if s.strip()] + else: + servers = list(_INTERACTSH_SERVERS) + + for server_host in servers: + for scheme in ("https", "http"): + server_url = f"{scheme}://{server_host}" + try: + async with httpx.AsyncClient(timeout=10.0) as client: + resp = await client.post( + f"{server_url}/register", + json=payload, + ) + if resp.status_code != 200: + continue + body = resp.json() + if body.get("message") != "registration successful": + continue + + # Registration succeeded + session = _InteractshSession( + server_url=server_url, + server_host=server_host, + correlation_id=correlation_id, + secret_key=secret_key, + private_key_der=private_der, + ) + self._interactsh_session = session + url = _build_interactsh_url(correlation_id, server_host) + self._callback_url = f"https://{url}" + self._provider = "interactsh_api" + return True + except Exception: + continue + return False + + async def _deregister_interactsh_api(self) -> None: + """Deregister from the interactsh server (best-effort cleanup).""" + session = self._interactsh_session + if session is None: + return + try: + async with httpx.AsyncClient(timeout=5.0) as client: + await client.post( + f"{session.server_url}/deregister", + json={ + "correlation-id": session.correlation_id, + "secret-key": session.secret_key, + }, + ) + except Exception: + pass + + async def _poll_interactsh_api(self, since_seconds: int) -> str: + """Poll the interactsh server for interactions and decrypt them.""" + session = self._interactsh_session + if session is None: + return "Error: No active interactsh session." + + try: + async with httpx.AsyncClient(timeout=10.0) as client: + resp = await client.get( + f"{session.server_url}/poll", + params={ + "id": session.correlation_id, + "secret": session.secret_key, + }, + ) + if resp.status_code != 200: + return f"Error: Interactsh poll failed: HTTP {resp.status_code}" + + body = resp.json() + encrypted_data: list[str] = body.get("data", []) + aes_key: str = body.get("aes_key", "") + extra: list[str] = body.get("extra", []) + + cutoff = time.time() - since_seconds + interactions: list[dict[str, str]] = [] + + # Decrypt encrypted interactions + if aes_key and encrypted_data: + for entry in encrypted_data: + try: + plaintext = _decrypt_interactsh_message( + session.private_key_der, aes_key, entry + ) + interaction = json.loads(plaintext) + ix = self._parse_interactsh_interaction(interaction) + if ix is None: + continue + # Check timestamp filter + ts_str = interaction.get("timestamp", "") + if ts_str and self._interactsh_ts_before_cutoff( + ts_str, cutoff + ): + continue + # Deduplicate + ix_id = ( + f"{ix['protocol']}:{ix['time']}:{ix['remote_address']}" + ) + if ix_id in self._seen_ids: + continue + self._seen_ids.add(ix_id) + interactions.append(ix) + except Exception: + continue + + # Process unencrypted extra/tlddata interactions + for plaintext_str in extra + body.get("tlddata", []): + if not plaintext_str: + continue + try: + interaction = json.loads(plaintext_str) + ix = self._parse_interactsh_interaction(interaction) + if ix is None: + continue + ts_str = interaction.get("timestamp", "") + if ts_str and self._interactsh_ts_before_cutoff(ts_str, cutoff): + continue + ix_id = f"{ix['protocol']}:{ix['time']}:{ix['remote_address']}" + if ix_id in self._seen_ids: + continue + self._seen_ids.add(ix_id) + interactions.append(ix) + except Exception: + continue + + if not interactions: + return "No new callback interactions since last check." + + lines = [f"Received {len(interactions)} callback interactions:"] + for i, ix in enumerate(interactions[:10], 1): + lines.append( + f" {i}. [{ix['time']}] {ix['protocol'].upper()} from {ix['remote_address']}" + ) + + if interactions: + last = interactions[-1] + raw = last.get("raw_request", "") + if raw: + lines.append(f"\nMost recent request:\n{raw[:1000]}") + + return "\n".join(lines) + + except Exception as e: + return f"Error: Interactsh poll error: {e}" + + @staticmethod + def _parse_interactsh_interaction(data: dict[str, object]) -> dict[str, str] | None: + """Parse a decrypted interactsh interaction dict into a normalised form.""" + protocol = str(data.get("protocol", "")) + if not protocol: + return None + return { + "protocol": protocol, + "time": str(data.get("timestamp", "")), + "unique_id": str(data.get("unique-id", "")), + "full_id": str(data.get("full-id", "")), + "remote_address": str(data.get("remote-address", "")), + "raw_request": str(data.get("raw-request", ""))[:1000], + "raw_response": str(data.get("raw-response", ""))[:500], + "q_type": str(data.get("q-type", "")), + "smtp_from": str(data.get("smtp-from", "")), + } + + @staticmethod + def _interactsh_ts_before_cutoff(ts_str: str, cutoff: float) -> bool: + """Return True if the timestamp is before the cutoff.""" + try: + ts = datetime.fromisoformat(ts_str.replace("Z", "+00:00")) + return ts.timestamp() < cutoff + except (ValueError, AttributeError): + return False + + # ------------------------------------------------------------------ + # Provider: interactsh CLI (legacy fallback) + # ------------------------------------------------------------------ + + def _register_interactsh_cli(self) -> bool: """Register with interactsh-client CLI as fallback.""" try: proc = subprocess.run( @@ -68,7 +438,7 @@ def _register_interactsh(self) -> bool: data = json.loads(line) if "url" in data: self._callback_url = data["url"] - self._provider = "interactsh" + self._provider = "interactsh_cli" return True except json.JSONDecodeError: if ".oast." in line or ".interact." in line: @@ -76,19 +446,29 @@ def _register_interactsh(self) -> bool: if not url.startswith("http"): url = f"https://{url}" self._callback_url = url - self._provider = "interactsh" + self._provider = "interactsh_cli" return True return False except Exception: return False + # ------------------------------------------------------------------ + # Registration orchestrator + # ------------------------------------------------------------------ + async def _ensure_registered(self) -> bool: """Ensure a callback URL is registered, trying providers in order.""" if self._callback_url: return True if await self._register_webhook_site(): return True - return self._register_interactsh() + if await self._register_interactsh_api(): + return True + return self._register_interactsh_cli() + + # ------------------------------------------------------------------ + # Tool methods (public API) + # ------------------------------------------------------------------ @tool_method(name="get_callback_url", catch=True) async def get_callback_url(self, protocol: str = "http") -> str: @@ -104,6 +484,8 @@ async def get_callback_url(self, protocol: str = "http") -> str: return "Error: Could not register with any callback provider." url = self._callback_url + assert url is not None # guarded by _ensure_registered + if protocol == "https" and url.startswith("http://"): url = url.replace("http://", "https://", 1) elif protocol == "dns": @@ -129,9 +511,11 @@ async def check_callbacks(self, since_seconds: int = 300) -> str: if self._provider == "webhook_site": return await self._poll_webhook_site(since_seconds) - if self._provider == "interactsh": + if self._provider == "interactsh_api": + return await self._poll_interactsh_api(since_seconds) + if self._provider == "interactsh_cli": return ( - "For interactsh, run in bash: interactsh-client -json | head -20\n\n" + "For interactsh CLI, run in bash: interactsh-client -json | head -20\n\n" "The CLI will show any interactions with your callback domain." ) return f"Error: Unknown provider: {self._provider}" @@ -228,8 +612,13 @@ async def _poll_webhook_site(self, since_seconds: int) -> str: @tool_method(name="reset_callback", catch=True) async def reset_callback(self) -> str: """Reset callback state. Next get_callback_url will register a new URL.""" + # Best-effort deregistration from interactsh + if self._provider == "interactsh_api": + await self._deregister_interactsh_api() + self._callback_url = None self._provider = None self._token_id = None self._seen_ids.clear() + self._interactsh_session = None return "Callback state reset. Next get_callback_url will register a new URL."