"""Exploit-reproduction + no-breaking-change tests for the custom-LLM baseUrl SSRF guard (STRIDE I15 / I16) and the agent CORS scoping (I17). These tests are written so that they FAIL on the pre-patch code (the exploit succeeds) and PASS on the patched code (the exploit is blocked). The reproduction is demonstrated out-of-band by running this module against the pre-patch tree via `git stash` (see the remediation notes), but the assertions themselves are self-contained. The SSRF reproduction uses a recording fake for the LLM client so we can assert on WHERE the request would have gone: pre-patch the metadata URL reaches the client constructor and `ainvoke` fires (SSRF); post-patch the guard rejects before the client is ever built. Run with: python -m unittest tests.test_ssrf_exploit_reproduction -v """ import sys import unittest from contextlib import asynccontextmanager from pathlib import Path from unittest.mock import patch _AGENTIC_DIR = Path(__file__).resolve().parent.parent sys.path.insert(0, str(_AGENTIC_DIR)) # A real internal address an attacker would target. Nothing legitimate runs # here, but the agent container can reach it -> the SSRF payoff. METADATA_URL = "http://169.254.169.254/latest/meta-data/iam/security-credentials/" REDIS_SSRF_URL = "http://127.0.0.1:6379/" # internal-service probe variant # The operator's REAL self-hosted endpoint, copied from internal/TODO.INTERNAL.md. # This MUST keep working (Docker host gateway, RFC1918) — the no-breaking case. REAL_OLLAMA_URL = "http://172.25.0.1:11435/v1" class _Recorder: """Records every LLM client constructed and every ainvoke, so a test can prove whether the attacker URL reached the network layer.""" constructed_base_urls = [] ainvoke_called_for = [] @classmethod def reset(cls): cls.constructed_base_urls = [] cls.ainvoke_called_for = [] class _FakeChatOpenAI: """Stand-in for langchain_openai.ChatOpenAI that records the base_url it was built with and whether ainvoke (the actual outbound request) fired.""" def __init__(self, **kwargs): self._base_url = kwargs.get("base_url") _Recorder.constructed_base_urls.append(self._base_url) async def ainvoke(self, *_args, **_kwargs): _Recorder.ainvoke_called_for.append(self._base_url) class _Resp: content = "hello" return _Resp() class _AppBase(unittest.TestCase): @classmethod def setUpClass(cls): @asynccontextmanager async def fake_lifespan(_app): yield with patch("api.lifespan", fake_lifespan): import api as api_module cls.api_module = api_module from fastapi.testclient import TestClient cls.client = TestClient(api_module.app) def setUp(self): _Recorder.reset() def _post(self, body): # Patch ChatOpenAI where it is actually looked up. `llm_setup` does a # module-level `from langchain_openai import ChatOpenAI`, so patching # `langchain_openai.ChatOpenAI` would NOT rebind that reference and the # real client would attempt a live connection. Patch both the source and # the bound name so the recording fake is always used. with patch("orchestrator_helpers.llm_setup.ChatOpenAI", _FakeChatOpenAI), \ patch("langchain_openai.ChatOpenAI", _FakeChatOpenAI): return self.client.post("/llm-provider/test", json=body) class SsrfExploitBlocked(_AppBase): """The core exploit: an (unauthenticated) caller makes the key-holding agent issue a request to an internal/metadata address. Post-patch it must never reach the network layer.""" def test_metadata_ssrf_never_reaches_client(self): resp = self._post({ "providerType": "openai_compatible", "baseUrl": METADATA_URL, "modelIdentifier": "x", "apiKey": "anything", }) # Guard rejects with 400 BEFORE building the client. self.assertEqual(resp.status_code, 400) self.assertFalse(resp.json()["success"]) # The SSRF payoff: prove no client was ever pointed at the metadata host # and no outbound request fired. On pre-patch code these lists contain # the metadata URL — i.e. the exploit succeeded. self.assertEqual(_Recorder.constructed_base_urls, []) self.assertEqual(_Recorder.ainvoke_called_for, []) def test_internal_redis_probe_variant(self): # Variant: localhost is allowed (Ollama lives there), so the URL guard # alone does NOT block 127.0.0.1:6379. This asserts the documented # behavior: the request is ATTEMPTED (reaches the client) rather than # guard-blocked — the localhost-internal-service case is mitigated by # CORS+auth, not by the baseUrl guard. Guards against a future change # that would silently start blocking localhost (breaking Ollama). resp = self._post({ "providerType": "openai_compatible", "baseUrl": REDIS_SSRF_URL, "modelIdentifier": "x", }) # Reached the client (not a guard reject): proves localhost still allowed. self.assertIn(REDIS_SSRF_URL, _Recorder.constructed_base_urls) def test_tls_off_public_exfil_blocked(self): # I16 variant: point at a public host with TLS verification off (MITM # exposure of key+prompt). Must be blocked before any request. resp = self._post({ "providerType": "openai_compatible", "baseUrl": "https://8.8.8.8/v1", "modelIdentifier": "x", "sslVerify": False, }) self.assertEqual(resp.status_code, 400) self.assertEqual(_Recorder.ainvoke_called_for, []) class NoBreakingChange(_AppBase): """The legitimate self-hosted-model feature must behave exactly as before: the guard lets the request through to the network layer.""" def test_real_operator_ollama_endpoint_allowed(self): # The exact endpoint from internal/TODO.INTERNAL.md. resp = self._post({ "providerType": "openai_compatible", "baseUrl": REAL_OLLAMA_URL, "modelIdentifier": "qwen2.5:0.5b", "apiKey": "ollama", }) # Request reached the client and fired ainvoke -> behaves like pre-patch. self.assertIn(REAL_OLLAMA_URL, _Recorder.constructed_base_urls) self.assertIn(REAL_OLLAMA_URL, _Recorder.ainvoke_called_for) self.assertEqual(resp.status_code, 200) self.assertTrue(resp.json()["success"]) def test_real_operator_ollama_endpoint_allowed_tls_off(self): # Private host -> sslVerify=false is still permitted (internal self-signed). resp = self._post({ "providerType": "openai_compatible", "baseUrl": REAL_OLLAMA_URL, "modelIdentifier": "qwen2.5:0.5b", "sslVerify": False, }) self.assertIn(REAL_OLLAMA_URL, _Recorder.ainvoke_called_for) self.assertEqual(resp.status_code, 200) def test_localhost_ollama_default_allowed(self): resp = self._post({ "providerType": "openai_compatible", "baseUrl": "http://127.0.0.1:11434/v1", "modelIdentifier": "llama3", }) self.assertIn("http://127.0.0.1:11434/v1", _Recorder.ainvoke_called_for) self.assertTrue(resp.json()["success"]) if __name__ == "__main__": unittest.main()