From a6c27de344b069c1d322dd900c4b4a7cb780d36b Mon Sep 17 00:00:00 2001 From: akushonkamen Date: Sat, 10 Oct 2026 14:47:56 +0800 Subject: [PATCH] fix(readurls): block SSRF destinations via stdlib host guard fetch_webpage_content() requested any URL extracted from the prompt with no restriction on the destination host, letting callers steer the server into private, loopback or link-local addresses (e.g. the cloud metadata IP 169.254.169.254). - Resolve the host with socket.getaddrinfo and refuse URLs where any resolved address is private, loopback, link-local, reserved, multicast or unspecified; fail closed on unresolvable/malformed hosts - Follow redirects manually (max 5) and re-run the guard on every hop so a public URL cannot redirect into an internal address - Operators can explicitly opt out with --readurls-allow-internal / OPTILLM_READURLS_ALLOW_INTERNAL for trusted internal fetches - Pure stdlib (ipaddress, socket); no new dependencies Fixes #323 --- README.md | 17 +++ optillm/plugins/readurls_plugin.py | 56 ++++++- optillm/server.py | 2 + tests/test_readurls_ssrf.py | 237 +++++++++++++++++++++++++++++ 4 files changed, 310 insertions(+), 2 deletions(-) create mode 100644 tests/test_readurls_ssrf.py diff --git a/README.md b/README.md index 9f44b5b7..1706482e 100644 --- a/README.md +++ b/README.md @@ -154,6 +154,23 @@ optillm ⚠️ **Security Note**: Disabling SSL verification is insecure and should only be used in development. For production environments with custom CAs, use `--ssl-cert-path` instead. See [SSL_CONFIGURATION.md](SSL_CONFIGURATION.md) for details. +## 🔒 URL Fetch Security (readurls plugin) + +The `readurls` plugin fetches URLs found in prompts and blocks requests to non-public destinations (private, loopback, link-local, reserved addresses such as the cloud metadata IP `169.254.169.254`) to prevent SSRF. Every redirect hop is re-validated and unresolvable hosts are rejected. + +If your deployment intentionally needs to read internal documentation pages, you can explicitly opt out: + +```bash +# Command line +optillm --readurls-allow-internal + +# Environment variable +export OPTILLM_READURLS_ALLOW_INTERNAL=true +optillm +``` + +⚠️ **Security Note**: Only enable this on trusted networks where internal URL fetching is actually required. + ## Implemented techniques | Approach | Slug | Description | diff --git a/optillm/plugins/readurls_plugin.py b/optillm/plugins/readurls_plugin.py index b799bca6..564e16b0 100644 --- a/optillm/plugins/readurls_plugin.py +++ b/optillm/plugins/readurls_plugin.py @@ -2,12 +2,44 @@ from typing import Tuple, List, Optional import requests import os +import ipaddress +import socket from bs4 import BeautifulSoup -from urllib.parse import urlparse +from urllib.parse import urlparse, urljoin from optillm import __version__, server_config SLUG = "readurls" +# Redirect statuses followed manually so each hop can be re-checked by the +# SSRF guard before the next request is issued +REDIRECT_STATUS_CODES = frozenset((301, 302, 303, 307, 308)) +MAX_REDIRECTS = 5 + + +def _is_blocked_host(url: str) -> bool: + """ + SSRF guard: return True when the URL does not point at a globally + reachable public address (private, loopback, link-local, reserved, + multicast or unspecified), or when the host cannot be resolved + (fail-closed). + """ + try: + hostname = urlparse(url).hostname + if not hostname: + return True + + # Resolve all addresses for the host and block the URL if any of + # them is non-public + for addr_info in socket.getaddrinfo(hostname, None): + ip = ipaddress.ip_address(addr_info[4][0]) + if (ip.is_private or ip.is_loopback or ip.is_link_local or + ip.is_reserved or ip.is_multicast or ip.is_unspecified): + return True + return False + except (OSError, ValueError, UnicodeError): + # Unresolvable or malformed hosts are blocked + return True + def extract_urls(text: str) -> List[str]: # Updated regex pattern to be more precise url_pattern = re.compile(r'https?://[^\s\'"]+') @@ -44,7 +76,27 @@ def fetch_webpage_content(url: str, max_length: int = 100000, verify_ssl: Option else: verify = True - response = requests.get(url, headers=headers, timeout=10, verify=verify) + # SSRF guard: resolve the host and refuse non-public destinations + # before fetching. Redirects are followed manually so every hop is + # re-checked, closing the bypass where a public URL redirects to an + # internal address. Operators can explicitly allow internal fetches + # via server_config['readurls_allow_internal']. + allow_internal = server_config.get('readurls_allow_internal', False) + current_url = url + response = None + for _ in range(MAX_REDIRECTS + 1): + if not allow_internal and _is_blocked_host(current_url): + return "Error fetching content: blocked non-public URL" + response = requests.get(current_url, headers=headers, timeout=10, verify=verify, allow_redirects=False) + if response.status_code not in REDIRECT_STATUS_CODES: + break + location = response.headers.get('Location') + if not location: + break + current_url = urljoin(current_url, location) + + if response.status_code in REDIRECT_STATUS_CODES: + return "Error fetching content: too many redirects" response.raise_for_status() # Make a soup diff --git a/optillm/server.py b/optillm/server.py index 8ed597fe..77d15862 100644 --- a/optillm/server.py +++ b/optillm/server.py @@ -195,6 +195,7 @@ def count_reasoning_tokens(text: str, tokenizer=None) -> int: 'log': 'info', 'ssl_verify': True, 'ssl_cert_path': '', + 'readurls_allow_internal': False, } # List of known approaches @@ -1094,6 +1095,7 @@ def parse_args(): ("--plugins-dir", "OPTILLM_PLUGINS_DIR", str, "", "Path to the plugins directory"), ("--log-conversations", "OPTILLM_LOG_CONVERSATIONS", bool, False, "Enable conversation logging with full metadata"), ("--conversation-log-dir", "OPTILLM_CONVERSATION_LOG_DIR", str, str(Path.home() / ".optillm" / "conversations"), "Directory to save conversation logs"), + ("--readurls-allow-internal", "OPTILLM_READURLS_ALLOW_INTERNAL", bool, False, "Allow the readurls plugin to fetch non-public/internal URLs (disables SSRF protection)"), ] for arg, env, type_, default, help_text, *extra in args_env: diff --git a/tests/test_readurls_ssrf.py b/tests/test_readurls_ssrf.py new file mode 100644 index 00000000..970e8213 --- /dev/null +++ b/tests/test_readurls_ssrf.py @@ -0,0 +1,237 @@ +""" +Unit tests for SSRF protection in the readurls plugin. + +Tests verify that the readurls plugin refuses to fetch non-public +destinations (private, loopback, link-local, reserved, multicast, +unspecified addresses), re-validates every redirect hop, fails closed on +DNS errors, and that operators can explicitly opt out via +server_config['readurls_allow_internal']. + +All tests are fully offline: DNS resolution is either exercised only on +literal IPs (which libc resolves without network access) or mocked. +""" + +import unittest +from unittest.mock import MagicMock, patch +import socket +import sys +import os + +# Add parent directory to path to import optillm modules +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from optillm import server_config + + +METADATA_URL = 'http://169.254.169.254/latest/meta-data/iam/security-credentials/' + + +def _addrinfo(ip): + """Build a getaddrinfo-shaped result for a single IP string.""" + af = socket.AF_INET6 if ':' in ip else socket.AF_INET + return [(af, socket.SOCK_STREAM, 6, '', (ip, 0))] + + +def _getaddrinfo_by_host(host_to_ip): + """Return a getaddrinfo stand-in: known hosts resolve via the mapping, + anything else (e.g. literal IPs) resolves to itself.""" + def fake_getaddrinfo(host, *args, **kwargs): + return _addrinfo(host_to_ip.get(host, host)) + return fake_getaddrinfo + + +class TestIsBlockedHost(unittest.TestCase): + """Unit tests for the _is_blocked_host guard.""" + + def test_blocks_loopback_ipv4(self): + from optillm.plugins.readurls_plugin import _is_blocked_host + self.assertTrue(_is_blocked_host('http://127.0.0.1/index.html')) + + def test_blocks_metadata_ip(self): + from optillm.plugins.readurls_plugin import _is_blocked_host + self.assertTrue(_is_blocked_host(METADATA_URL)) + + def test_blocks_private_10(self): + from optillm.plugins.readurls_plugin import _is_blocked_host + self.assertTrue(_is_blocked_host('http://10.0.0.1/')) + + def test_blocks_private_192_168(self): + from optillm.plugins.readurls_plugin import _is_blocked_host + self.assertTrue(_is_blocked_host('http://192.168.1.1/')) + + def test_blocks_private_172_16(self): + from optillm.plugins.readurls_plugin import _is_blocked_host + self.assertTrue(_is_blocked_host('http://172.16.0.5/')) + + def test_blocks_unspecified(self): + from optillm.plugins.readurls_plugin import _is_blocked_host + self.assertTrue(_is_blocked_host('http://0.0.0.0/')) + + def test_blocks_ipv6_loopback(self): + from optillm.plugins.readurls_plugin import _is_blocked_host + self.assertTrue(_is_blocked_host('http://[::1]/')) + + def test_blocks_ipv6_link_local(self): + from optillm.plugins.readurls_plugin import _is_blocked_host + self.assertTrue(_is_blocked_host('http://[fe80::1]/')) + + def test_blocks_localhost(self): + from optillm.plugins.readurls_plugin import _is_blocked_host + with patch('optillm.plugins.readurls_plugin.socket.getaddrinfo', + return_value=_addrinfo('127.0.0.1')): + self.assertTrue(_is_blocked_host('http://localhost/secret')) + + def test_blocks_missing_hostname(self): + from optillm.plugins.readurls_plugin import _is_blocked_host + self.assertTrue(_is_blocked_host('http:///path')) + + def test_blocks_unresolvable_host(self): + from optillm.plugins.readurls_plugin import _is_blocked_host + with patch('optillm.plugins.readurls_plugin.socket.getaddrinfo', + side_effect=socket.gaierror('name resolution failure')): + self.assertTrue(_is_blocked_host('http://internal.example.invalid/')) + + def test_allows_public_literal_ip(self): + from optillm.plugins.readurls_plugin import _is_blocked_host + self.assertFalse(_is_blocked_host('http://93.184.216.34/')) + + def test_allows_public_resolved_host(self): + from optillm.plugins.readurls_plugin import _is_blocked_host + with patch('optillm.plugins.readurls_plugin.socket.getaddrinfo', + return_value=_addrinfo('93.184.216.34')): + self.assertFalse(_is_blocked_host('https://example.com/article')) + + +class TestFetchWebpageContentGuard(unittest.TestCase): + """Behavior tests: fetch_webpage_content refuses non-public URLs.""" + + def setUp(self): + self.original_config = server_config.copy() + + def tearDown(self): + server_config.clear() + server_config.update(self.original_config) + + @patch('optillm.plugins.readurls_plugin.requests.get') + def test_fetch_blocks_metadata_url(self, mock_requests_get): + """The fetcher must short-circuit before issuing any request.""" + from optillm.plugins.readurls_plugin import fetch_webpage_content + + result = fetch_webpage_content(METADATA_URL) + + self.assertIn('Error fetching content', result) + self.assertIn('blocked non-public URL', result) + mock_requests_get.assert_not_called() + + @patch('optillm.plugins.readurls_plugin.requests.get') + def test_fetch_allows_public_url(self, mock_requests_get): + """Public URLs are still fetched normally.""" + from optillm.plugins.readurls_plugin import fetch_webpage_content + + mock_response = MagicMock() + mock_response.content = b'

Public content

' + mock_response.raise_for_status = MagicMock() + mock_requests_get.return_value = mock_response + + with patch('optillm.plugins.readurls_plugin.socket.getaddrinfo', + return_value=_addrinfo('93.184.216.34')): + result = fetch_webpage_content('https://example.com/article') + + self.assertEqual(result, 'Public content') + mock_requests_get.assert_called_once() + + @patch('optillm.plugins.readurls_plugin.requests.get') + def test_fetch_blocks_redirect_to_internal(self, mock_requests_get): + """A public URL redirecting to an internal address must be blocked.""" + from optillm.plugins.readurls_plugin import fetch_webpage_content + + redirect_response = MagicMock() + redirect_response.status_code = 302 + redirect_response.headers = {'Location': METADATA_URL} + redirect_response.raise_for_status = MagicMock() + + final_response = MagicMock() + final_response.content = b'

Leaked

' + final_response.raise_for_status = MagicMock() + mock_requests_get.side_effect = [redirect_response, final_response] + + with patch('optillm.plugins.readurls_plugin.socket.getaddrinfo', + side_effect=_getaddrinfo_by_host({'example.com': '93.184.216.34'})): + result = fetch_webpage_content('https://example.com/redirect') + + self.assertIn('blocked non-public URL', result) + self.assertNotIn('Leaked', result) + # Only the initial public request may have been issued + self.assertEqual(mock_requests_get.call_count, 1) + + @patch('optillm.plugins.readurls_plugin.requests.get') + def test_fetch_follows_public_redirect(self, mock_requests_get): + """Redirects between public hosts keep working.""" + from optillm.plugins.readurls_plugin import fetch_webpage_content + + redirect_response = MagicMock() + redirect_response.status_code = 301 + redirect_response.headers = {'Location': 'https://example.com/moved'} + final_response = MagicMock() + final_response.content = b'

Moved content

' + mock_requests_get.side_effect = [redirect_response, final_response] + + with patch('optillm.plugins.readurls_plugin.socket.getaddrinfo', + return_value=_addrinfo('93.184.216.34')): + result = fetch_webpage_content('https://example.com/old') + + self.assertEqual(result, 'Moved content') + self.assertEqual(mock_requests_get.call_count, 2) + self.assertEqual(mock_requests_get.call_args_list[1][0][0], + 'https://example.com/moved') + + @patch('optillm.plugins.readurls_plugin.requests.get') + def test_fetch_allow_internal_opt_out(self, mock_requests_get): + """Operators can explicitly allow internal fetches.""" + from optillm.plugins.readurls_plugin import fetch_webpage_content + + server_config['readurls_allow_internal'] = True + + mock_response = MagicMock() + mock_response.content = b'

Internal docs

' + mock_response.raise_for_status = MagicMock() + mock_requests_get.return_value = mock_response + + result = fetch_webpage_content('http://127.0.0.1:8080/internal-docs') + + self.assertEqual(result, 'Internal docs') + mock_requests_get.assert_called_once() + + +class TestReadurlsRun(unittest.TestCase): + """run()-level tests with mixed public and internal URLs.""" + + def setUp(self): + self.original_config = server_config.copy() + + def tearDown(self): + server_config.clear() + server_config.update(self.original_config) + + @patch('optillm.plugins.readurls_plugin.requests.get') + def test_run_blocks_internal_and_keeps_public(self, mock_requests_get): + from optillm.plugins.readurls_plugin import run + + mock_response = MagicMock() + mock_response.content = b'

Public page text

' + mock_response.raise_for_status = MagicMock() + mock_requests_get.return_value = mock_response + + query = f"Summarize {METADATA_URL} and https://example.com/page" + with patch('optillm.plugins.readurls_plugin.socket.getaddrinfo', + side_effect=_getaddrinfo_by_host({'example.com': '93.184.216.34'})): + modified_query, _ = run("", query) + + # Only the public URL reached the network layer + self.assertEqual(mock_requests_get.call_count, 1) + self.assertIn('blocked non-public URL', modified_query) + self.assertIn('Public page text', modified_query) + + +if __name__ == '__main__': + unittest.main()