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

Author: akushonkamen

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                                                                                    |

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 

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:

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'<html><body><p>Public content</p></body></html>'
+        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'<html><body><p>Leaked</p></body></html>'
+        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'<html><body><p>Moved content</p></body></html>'
+        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'<html><body><p>Internal docs</p></body></html>'
+        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'<html><body><p>Public page text</p></body></html>'
+        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()