CVE-2026-70485 (GHSA-8X5V-CPV7-8JJP)

@@ -19,9 +19,13 @@ )  import aiohttp+import aiohttp.resolver import certifi import requests+import urllib3.connection+import urllib3.connectionpool import validators+from requests.adapters import HTTPAdapter from fastapi.concurrency import run_in_threadpool from langchain_community.document_loaders import PlaywrightURLLoader, WebBaseLoader from langchain_community.document_loaders.base import BaseLoader@@ -94,7 +98,7 @@ def validate_url(url: Union[str, Sequence[str]]): # Get IPv4 and IPv6 addresses ipv4_addresses, ipv6_addresses = resolve_hostname(parsed_url.hostname) # Check if any of the resolved addresses are private- # This is technically still vulnerable to DNS rebinding attacks, as we don't control WebBaseLoader+ # DNS rebinding is mitigated at the connection layer; see _SSRFSafeResolver / _SSRFSafeAdapter for ip in ipv4_addresses + ipv6_addresses: addr = ipaddress.ip_address(ip) if not addr.is_global:@@ -118,6 +122,81 @@ def safe_validate_urls(url: Sequence[str]) -> Sequence[str]: return valid_urls  +def _ssrf_safe_new_conn(self):+ """Resolve DNS, validate all IPs are global, connect to validated IP.++ Replaces urllib3's _new_conn so the DNS lookup that feeds the actual TCP+ connect is the same one we validate — no second resolution, no rebinding+ window.+ """+ host = getattr(self, '_dns_host', self.host)+ port = self.port+ infos = socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM)+ if not infos:+ raise OSError(f'getaddrinfo for {host!r} returned empty list')+ if not ENABLE_RAG_LOCAL_WEB_FETCH:+ for _, _, _, _, sa in infos:+ if not ipaddress.ip_address(sa[0]).is_global:+ raise ValueError(ERROR_MESSAGES.INVALID_URL)+ err = None+ for fam, typ, proto, _, sa in infos:+ sock = None+ try:+ sock = socket.socket(fam, typ, proto)+ if self.timeout is not socket._GLOBAL_DEFAULT_TIMEOUT:+ sock.settimeout(self.timeout)+ if getattr(self, 'source_address', None):+ sock.bind(self.source_address)+ for opt in getattr(self, 'socket_options', None) or:+ sock.setsockopt(*opt)+ sock.connect(sa)+ return sock+ except OSError as exc:+ err = exc+ if sock is not None:+ sock.close()+ raise err or OSError(f'connect to {host!r}:{port} failed')+++class _SafeHTTPConn(urllib3.connection.HTTPConnection):+ _new_conn = _ssrf_safe_new_conn+++class _SafeHTTPSConn(urllib3.connection.HTTPSConnection):+ _new_conn = _ssrf_safe_new_conn+++class _SafeHTTPPool(urllib3.connectionpool.HTTPConnectionPool):+ ConnectionCls = _SafeHTTPConn+++class _SafeHTTPSPool(urllib3.connectionpool.HTTPSConnectionPool):+ ConnectionCls = _SafeHTTPSConn+++class _SSRFSafeAdapter(HTTPAdapter):+ """requests transport adapter that validates resolved IPs at connect time."""++ def init_poolmanager(self, *args, **kwargs):+ super().init_poolmanager(*args, **kwargs)+ self.poolmanager.pool_classes_by_scheme = {+ 'http': _SafeHTTPPool,+ 'https': _SafeHTTPSPool,+ }++

AI introduced this behavior: `# DNS rebinding is mitigated at the connection layer; see _SSRFSafeResolver / _SSRFSafeAdapter`

@@ -75,6 +75,34 @@ def resolve_hostname(hostname): return ipv4_addresses, ipv6_addresses  +def _is_global_addr(ip: str) -> bool:+ addr = ipaddress.ip_address(ip)+ if not addr.is_global:+ return False+ if not isinstance(addr, ipaddress.IPv6Address):+ return True++ embedded = []+ if addr.ipv4_mapped:+ embedded.append(addr.ipv4_mapped)+ if addr.sixtofour:+ embedded.append(addr.sixtofour)+ if addr.teredo:+ embedded.extend(addr.teredo)++ b = addr.packed+ if b[:12] == b"\x00" * 12:+ embedded.append(ipaddress.IPv4Address(b[12:]))+ elif b[:12] == b"\x00\x64\xff\x9b" + b"\x00" * 8:+ embedded.append(ipaddress.IPv4Address(b[12:]))+ elif b[:6] == b"\x00\x64\xff\x9b\x00\x01":+ if b[8] != 0:+ return False+ embedded.append(ipaddress.IPv4Address(bytes((b[6], b[7], b[9], b[10]))))++ return all(ip.is_global for ip in embedded)++ def validate_url(url: Union[str, Sequence[str]]): if isinstance(url, str): if isinstance(validators.url(url), validators.ValidationError):@@ -111,8 +139,7 @@ def validate_url(url: Union[str, Sequence[str]]): # Check if any of the resolved addresses are private # DNS rebinding is mitigated at the connection layer; see _SSRFSafeResolver / _SSRFSafeAdapter for ip in ipv4_addresses + ipv6_addresses:- addr = ipaddress.ip_address(ip)- if not addr.is_global:+ if not _is_global_addr(ip): raise ValueError(ERROR_MESSAGES.INVALID_URL) return True elif isinstance(url, Sequence):@@ -147,7 +174,7 @@ def _ssrf_safe_new_conn(self): raise OSError(f'getaddrinfo for {host!r} returned empty list') if not ENABLE_LOCAL_WEB_FETCH: for _, _, _, _, sa in infos:- if not ipaddress.ip_address(sa[0]).is_global:+ if not _is_global_addr(sa[0]): raise ValueError(ERROR_MESSAGES.INVALID_URL) err = None for fam, typ, proto, _, sa in infos:@@ -208,7 +235,7 @@ async def resolve(self, host, port=0, family=socket.AF_INET): results = await super().resolve(host, port, family) if not ENABLE_LOCAL_WEB_FETCH: for entry in results:- if not ipaddress.ip_address(entry['host']).is_global:+ if not _is_global_addr(entry['host']): raise ValueError(ERROR_MESSAGES.INVALID_URL) return results 

Open WebUI: Any authenticated user can reach internal services and cloud metadata via NAT64-encoded URLs