| """ |
| Agent streaming fuzzing harness for discovering crashes in streaming endpoints. |
| |
| This module uses Atheris to fuzz agent streaming endpoints: |
| - POST /api/agents/{id}/chat - Agent chat with streaming responses |
| - WebSocket connections for real-time streaming |
| - Server-Sent Events (SSE) endpoints |
| - Timeout handling for long-running streams |
| |
| Target: Streaming endpoint parsing/validation code crashes |
| Uses httpx client (not requests) for async/streaming support and realistic testing |
| """ |
|
|
| import os |
| import sys |
|
|
| import pytest |
|
|
| |
| backend_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) |
| if backend_dir not in sys.path: |
| sys.path.insert(0, backend_dir) |
|
|
| |
| try: |
| import httpx |
| HTTPX_AVAILABLE = True |
| except ImportError: |
| HTTPX_AVAILABLE = False |
| print("Warning: httpx not installed. Streaming fuzzing tests will be skipped.") |
| print("Install with: pip install httpx") |
|
|
| |
| from tests.e2e_ui.fixtures.auth_fixtures import authenticated_user |
|
|
| |
| try: |
| import atheris |
| from atheris import fp |
| ATHERIS_AVAILABLE = True |
| except ImportError: |
| ATHERIS_AVAILABLE = False |
|
|
|
|
| |
| |
| |
|
|
| @pytest.mark.fuzzing |
| @pytest.mark.slow |
| @pytest.mark.timeout(300) |
| def test_agent_chat_streaming_fuzz(authenticated_user): |
| """ |
| Fuzz agent chat streaming endpoint (POST /api/agents/{id}/chat). |
| |
| Target crashes in: |
| - Agent ID parsing/validation |
| - Message parsing |
| - Streaming response handling |
| - Connection timeout handling |
| |
| Edge cases: |
| - Agent ID: None, empty, SQL injection, huge length |
| - Message: Empty strings, huge messages (10000+ chars), null bytes |
| - Parameters: Nested structures, malicious payloads |
| - Connection drops during streaming |
| |
| Uses httpx client (FUZZ-04 requirement) for: |
| - Realistic async HTTP testing |
| - Streaming response support |
| - Timeout configuration (5-10s to prevent hangs) |
| |
| Args: |
| authenticated_user: (user, token) tuple for JWT auth |
| """ |
| if not ATHERIS_AVAILABLE: |
| pytest.skip("Atheris not installed - fuzzing test skipped") |
| if not HTTPX_AVAILABLE: |
| pytest.skip("httpx not installed - streaming fuzzing test skipped") |
|
|
| user, token = authenticated_user |
|
|
| |
| base_url = "http://localhost:8000" |
|
|
| |
| headers = {"Authorization": f"Bearer {token}"} |
|
|
| def fuzz_one_input(data: bytes): |
| """Fuzz agent chat streaming endpoint with mutated input. |
| |
| Args: |
| data: Random bytes from Atheris fuzzer |
| |
| Raises: |
| Exception: Crash discovered (Atheris catches this) |
| """ |
| try: |
| fdp = fp.FuzzedDataProvider(data) |
|
|
| |
| agent_id_option = fdp.ConsumeIntInRange(0, 5) |
| if agent_id_option == 0: |
| agent_id = None |
| elif agent_id_option == 1: |
| agent_id = "" |
| elif agent_id_option == 2: |
| |
| agent_id = "'; DROP TABLE agents; --" |
| elif agent_id_option == 3: |
| |
| agent_id = "<script>alert('xss')</script>" |
| elif agent_id_option == 4: |
| |
| agent_id = "A" * 10000 |
| else: |
| |
| agent_id = fdp.ConsumeRandomLengthString(50) |
|
|
| |
| message_option = fdp.ConsumeIntInRange(0, 4) |
| if message_option == 0: |
| message = None |
| elif message_option == 1: |
| message = "" |
| elif message_option == 2: |
| |
| message = "A" * 10000 |
| elif message_option == 3: |
| |
| message = "test\x00\x00\x00message" |
| else: |
| message = fdp.ConsumeRandomLengthString(1000) |
|
|
| |
| num_params = fdp.ConsumeIntInRange(0, 10) |
| parameters = {} |
|
|
| for i in range(num_params): |
| key = fdp.ConsumeRandomLengthString(20) |
| value_option = fdp.ConsumeIntInRange(0, 3) |
|
|
| if value_option == 0: |
| value = None |
| elif value_option == 1: |
| value = fdp.ConsumeRandomLengthString(100) |
| else: |
| |
| value = { |
| "nested": fdp.ConsumeRandomLengthString(50) |
| } |
|
|
| parameters[key] = value |
|
|
| |
| payload = { |
| "message": message, |
| "parameters": parameters |
| } |
|
|
| |
| |
| with httpx.Client(timeout=5.0) as client: |
| |
| if agent_id is None: |
| url = f"{base_url}/api/agents/None/chat" |
| else: |
| url = f"{base_url}/api/agents/{agent_id}/chat" |
|
|
| try: |
| response = client.post( |
| url, |
| json=payload, |
| headers=headers |
| ) |
|
|
| |
| |
| assert response.status_code in [200, 400, 401, 404, 422], \ |
| f"Unexpected status code {response.status_code}: {response.text}" |
|
|
| except httpx.ConnectError: |
| |
| pass |
| except httpx.TimeoutException: |
| |
| pass |
| except httpx.RemoteProtocolError: |
| |
| pass |
|
|
| except (ValueError, KeyError, IndexError, AttributeError) as e: |
| |
| pass |
| except Exception as e: |
| |
| raise Exception(f"Crash in agent chat streaming fuzzing: {e}") |
|
|
| |
| iterations = int(os.getenv("FUZZ_ITERATIONS", "10000")) |
| atheris.Setup(sys.argv, []) |
| atheris.Fuzz(fuzz_one_input, iterations=iterations) |
|
|
|
|
| |
| |
| |
|
|
| @pytest.mark.fuzzing |
| @pytest.mark.slow |
| @pytest.mark.timeout(300) |
| def test_agent_websocket_fuzz(authenticated_user): |
| """ |
| Fuzz WebSocket connection for agent chat. |
| |
| Target crashes in: |
| - WebSocket connection handling |
| - Message frame parsing |
| - Connection close logic |
| - Invalid message format handling |
| |
| Edge cases: |
| - Agent ID: SQL injection, XSS, null bytes |
| - Initial message: Malformed frames, invalid JSON |
| - Connection drops during handshake |
| - Concurrent WebSocket connections |
| |
| Uses httpx WebSocket client (or websockets library) for realistic testing. |
| |
| Args: |
| authenticated_user: (user, token) tuple for JWT auth |
| """ |
| if not ATHERIS_AVAILABLE: |
| pytest.skip("Atheris not installed - fuzzing test skipped") |
| if not HTTPX_AVAILABLE: |
| pytest.skip("httpx not installed - websocket fuzzing test skipped") |
|
|
| user, token = authenticated_user |
|
|
| |
| ws_base_url = "ws://localhost:8000" |
|
|
| def fuzz_one_input(data: bytes): |
| """Fuzz WebSocket connection with mutated input.""" |
| try: |
| fdp = fp.FuzzedDataProvider(data) |
|
|
| |
| agent_id_option = fdp.ConsumeIntInRange(0, 3) |
| if agent_id_option == 0: |
| agent_id = "'; DROP TABLE agents; --" |
| elif agent_id_option == 1: |
| agent_id = "\x00\x00\x00" |
| else: |
| agent_id = fdp.ConsumeRandomLengthString(50) |
|
|
| |
| message_option = fdp.ConsumeIntInRange(0, 2) |
| if message_option == 0: |
| initial_message = None |
| elif message_option == 1: |
| |
| initial_message = '{"invalid": json}' |
| else: |
| initial_message = fdp.ConsumeRandomLengthString(500) |
|
|
| |
| try: |
| |
| |
| try: |
| import websockets |
|
|
| async def websocket_test(): |
| uri = f"{ws_base_url}/ws/{agent_id}" |
| try: |
| async with websockets.connect(uri, timeout=5, close_timeout=5) as websocket: |
| if initial_message: |
| await websocket.send(initial_message) |
| response = await websocket.recv() |
| except Exception: |
| |
| pass |
|
|
| |
| import asyncio |
| asyncio.run(websocket_test()) |
|
|
| except ImportError: |
| |
| pass |
|
|
| except Exception as e: |
| |
| |
| pass |
|
|
| except (ValueError, KeyError) as e: |
| pass |
| except Exception as e: |
| |
| raise Exception(f"Crash in WebSocket fuzzing: {e}") |
|
|
| iterations = int(os.getenv("FUZZ_ITERATIONS", "10000")) |
| atheris.Setup(sys.argv, []) |
| atheris.Fuzz(fuzz_one_input, iterations=iterations) |
|
|
|
|
| |
| |
| |
|
|
| @pytest.mark.fuzzing |
| @pytest.mark.slow |
| @pytest.mark.timeout(300) |
| def test_streaming_sse_fuzz(authenticated_user): |
| """ |
| Fuzz Server-Sent Events (SSE) endpoint. |
| |
| Target crashes in: |
| - SSE endpoint parsing |
| - Invalid Accept headers |
| - Malformed query parameters |
| - Connection handling |
| |
| Edge cases: |
| - Agent ID: SQL injection, XSS, null bytes |
| - Query parameters: Invalid values, huge strings |
| - Accept headers: Invalid MIME types |
| - Connection drops during SSE stream |
| |
| Args: |
| authenticated_user: (user, token) tuple for JWT auth |
| """ |
| if not ATHERIS_AVAILABLE: |
| pytest.skip("Atheris not installed - fuzzing test skipped") |
| if not HTTPX_AVAILABLE: |
| pytest.skip("httpx not installed - SSE fuzzing test skipped") |
|
|
| user, token = authenticated_user |
|
|
| base_url = "http://localhost:8000" |
| headers = {"Authorization": f"Bearer {token}"} |
|
|
| def fuzz_one_input(data: bytes): |
| """Fuzz SSE endpoint with mutated input.""" |
| try: |
| fdp = fp.FuzzedDataProvider(data) |
|
|
| |
| agent_id_option = fdp.ConsumeIntInRange(0, 3) |
| if agent_id_option == 0: |
| agent_id = "'; DROP TABLE agents; --" |
| elif agent_id_option == 1: |
| agent_id = "<script>alert(1)</script>" |
| else: |
| agent_id = fdp.ConsumeRandomLengthString(50) |
|
|
| |
| num_params = fdp.ConsumeIntInRange(0, 5) |
| params = {} |
|
|
| for i in range(num_params): |
| key = fdp.ConsumeRandomLengthString(20) |
| value = fdp.ConsumeRandomLengthString(100) |
| params[key] = value |
|
|
| |
| accept_option = fdp.ConsumeIntInRange(0, 3) |
| if accept_option == 0: |
| accept_header = "text/event-stream" |
| elif accept_option == 1: |
| |
| accept_header = "invalid/mime-type" |
| else: |
| accept_header = fdp.ConsumeRandomLengthString(50) |
|
|
| headers_with_accept = headers.copy() |
| headers_with_accept["Accept"] = accept_header |
|
|
| |
| with httpx.Client(timeout=5.0) as client: |
| try: |
| response = client.get( |
| f"{base_url}/api/agents/{agent_id}/stream", |
| params=params, |
| headers=headers_with_accept |
| ) |
|
|
| |
| |
| assert response.status_code in [200, 400, 404, 406, 422], \ |
| f"Unexpected status code {response.status_code}" |
|
|
| except httpx.ConnectError: |
| |
| pass |
| except httpx.TimeoutException: |
| |
| pass |
| except httpx.RemoteProtocolError: |
| |
| pass |
|
|
| except (ValueError, KeyError): |
| pass |
| except Exception as e: |
| raise Exception(f"Crash in SSE fuzzing: {e}") |
|
|
| iterations = int(os.getenv("FUZZ_ITERATIONS", "10000")) |
| atheris.Setup(sys.argv, []) |
| atheris.Fuzz(fuzz_one_input, iterations=iterations) |
|
|
|
|
| |
| |
| |
|
|
| @pytest.mark.fuzzing |
| @pytest.mark.slow |
| @pytest.mark.timeout(300) |
| def test_streaming_timeout_fuzz(authenticated_user): |
| """ |
| Fuzz streaming timeout handling. |
| |
| Target crashes in: |
| - Client timeout handling |
| - Server timeout logic |
| - Hung connection cleanup |
| - Graceful shutdown |
| |
| Edge cases: |
| - Agents with infinite loops (simulated) |
| - Very short timeouts (< 1s) |
| - Very long timeouts (> 300s) |
| - Connection drops during timeout |
| |
| Ensures graceful timeout handling (no hung connections, no crashes). |
| |
| Args: |
| authenticated_user: (user, token) tuple for JWT auth |
| """ |
| if not ATHERIS_AVAILABLE: |
| pytest.skip("Atheris not installed - fuzzing test skipped") |
| if not HTTPX_AVAILABLE: |
| pytest.skip("httpx not installed - timeout fuzzing test skipped") |
|
|
| user, token = authenticated_user |
|
|
| base_url = "http://localhost:8000" |
| headers = {"Authorization": f"Bearer {token}"} |
|
|
| def fuzz_one_input(data: bytes): |
| """Fuzz streaming timeout with varied timeout values.""" |
| try: |
| fdp = fp.FuzzedDataProvider(data) |
|
|
| |
| agent_id = fdp.ConsumeRandomLengthString(50) |
|
|
| |
| timeout_option = fdp.ConsumeIntInRange(0, 4) |
| if timeout_option == 0: |
| timeout = 0.1 |
| elif timeout_option == 1: |
| timeout = 5.0 |
| elif timeout_option == 2: |
| timeout = 300.0 |
| else: |
| timeout = fdp.ConsumeFloatInRange(0.1, 300.0) |
|
|
| |
| message = fdp.ConsumeRandomLengthString(1000) |
|
|
| |
| payload = { |
| "message": message, |
| "parameters": {"timeout": timeout} |
| } |
|
|
| |
| |
| with httpx.Client(timeout=timeout) as client: |
| try: |
| response = client.post( |
| f"{base_url}/api/agents/{agent_id}/chat", |
| json=payload, |
| headers=headers |
| ) |
|
|
| |
| |
| assert response.status_code in [200, 400, 401, 404, 422], \ |
| f"Unexpected status code {response.status_code}" |
|
|
| except httpx.TimeoutException: |
| |
| pass |
| except httpx.ConnectError: |
| |
| pass |
|
|
| except (ValueError, KeyError): |
| pass |
| except Exception as e: |
| |
| raise Exception(f"Crash in timeout fuzzing: {e}") |
|
|
| iterations = int(os.getenv("FUZZ_ITERATIONS", "10000")) |
| atheris.Setup(sys.argv, []) |
| atheris.Fuzz(fuzz_one_input, iterations=iterations) |
|
|