Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -171,6 +171,8 @@ optillm

⚠️ **Security Note**: Only enable this on trusted networks where internal URL fetching is actually required.

**Known limitation (DNS rebinding)**: the guard validates DNS resolution before each request, but the underlying HTTP client performs a second resolution when it opens the connection, so a domain under attacker control could rebind to an internal address between the two lookups. For high-security deployments, additionally restrict the network egress of the optillm process (firewall or service-mesh policy) instead of relying on this check alone.

## Implemented techniques

| Approach | Slug | Description |
Expand Down
26 changes: 20 additions & 6 deletions optillm/plugins/readurls_plugin.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,16 @@
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).
reachable public address (loopback, private, link-local, reserved,
CGNAT/shared space, multicast or unspecified), or when the host cannot
be resolved (fail-closed).

Known limitation (DNS rebinding): the host is resolved here and the HTTP
client performs a second resolution when it opens the connection, so a
domain under attacker control could rebind to an internal address between
the two lookups. For high-security deployments, additionally restrict the
network egress of the optillm process instead of relying on this check
alone.
"""
try:
hostname = urlparse(url).hostname
Expand All @@ -30,10 +37,17 @@ def _is_blocked_host(url: str) -> bool:

# 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):
addr_infos = socket.getaddrinfo(hostname, None)
if not addr_infos:
# No resolution results: fail closed
return True
for addr_info in addr_infos:
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):
# ipaddress marks globally-routable multicast ranges as global,
# so multicast must be blocked explicitly. CGNAT/shared space
# (100.64.0.0/10, e.g. Tailscale) is not in the classic private
# ranges but is not globally routable either.
if ip.is_multicast or not ip.is_global:
return True
return False
except (OSError, ValueError, UnicodeError):
Expand Down
20 changes: 20 additions & 0 deletions tests/test_readurls_ssrf.py
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,26 @@ def test_allows_public_resolved_host(self):
return_value=_addrinfo('93.184.216.34')):
self.assertFalse(_is_blocked_host('https://example.com/article'))

def test_blocks_cgnat_shared_range(self):
"""100.64.0.0/10 (carrier-grade NAT, e.g. Tailscale) is not globally
routable even though it is outside the classic private ranges."""
from optillm.plugins.readurls_plugin import _is_blocked_host
self.assertTrue(_is_blocked_host('http://100.64.0.1/'))
self.assertTrue(_is_blocked_host('http://100.100.100.100/'))

def test_blocks_multicast(self):
"""Multicast ranges are marked global by ipaddress but must be blocked."""
from optillm.plugins.readurls_plugin import _is_blocked_host
self.assertTrue(_is_blocked_host('http://224.0.0.1/'))
self.assertTrue(_is_blocked_host('http://[ff02::1]/'))

def test_blocks_empty_resolution(self):
"""An empty resolution result must fail closed, not be allowed."""
from optillm.plugins.readurls_plugin import _is_blocked_host
with patch('optillm.plugins.readurls_plugin.socket.getaddrinfo',
return_value=[]):
self.assertTrue(_is_blocked_host('http://example.com/'))


class TestFetchWebpageContentGuard(unittest.TestCase):
"""Behavior tests: fetch_webpage_content refuses non-public URLs."""
Expand Down
Loading