Coverage for open_webui/retrieval/web/utils.py: 11%
668 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 05:07 +0000
1import asyncio
2import http.cookiejar
3import ipaddress
4import logging
5import socket
6import ssl
7import time
8import urllib.parse
9import urllib.request
10from datetime import datetime, timedelta
11from importlib import import_module
12from typing import (
13 Any,
14 AsyncIterator,
15 Dict,
16 Iterable,
17 Iterator,
18 List,
19 Literal,
20 Optional,
21 Sequence,
22 Tuple,
23 Union,
24)
26import aiohttp
27import certifi
28import requests
29import urllib3.connection
30import urllib3.connectionpool
31import validators
32from requests.adapters import HTTPAdapter
33from fastapi import HTTPException
34from fastapi.concurrency import run_in_threadpool
35from langchain_core.document_loaders import BaseLoader
36from langchain_core.documents import Document
37from open_webui.config import (
38 ENABLE_LOCAL_WEB_FETCH,
39 EXTERNAL_WEB_LOADER_API_KEY,
40 EXTERNAL_WEB_LOADER_URL,
41 FIRECRAWL_API_BASE_URL,
42 FIRECRAWL_API_KEY,
43 FIRECRAWL_TIMEOUT,
44 MICROSOFT_WEB_IQ_API_BASE_URL,
45 MICROSOFT_WEB_IQ_API_KEY,
46 MICROSOFT_WEB_IQ_LANGUAGE,
47 PLAYWRIGHT_TIMEOUT,
48 PLAYWRIGHT_WS_URL,
49 TAVILY_API_KEY,
50 TAVILY_EXTRACT_DEPTH,
51 WEB_FETCH_FILTER_LIST,
52 WEB_LOADER_ENGINE,
53 WEB_LOADER_TIMEOUT,
54)
55from open_webui.constants import ERROR_MESSAGES
56from open_webui.env import (
57 AIOHTTP_CLIENT_ALLOW_REDIRECTS,
58 AIOHTTP_CLIENT_SESSION_SSL,
59 AIOHTTP_CLIENT_SSL_CERT_FILE,
60 AIOHTTP_CLIENT_TIMEOUT,
61 USER_AGENT,
62 USE_SLIM,
63)
64from open_webui.retrieval.loaders.external_web import ExternalWebLoader
65from open_webui.retrieval.loaders.microsoft_web_iq import MicrosoftWebIQLoader
66from open_webui.retrieval.loaders.tavily import TavilyLoader
67from open_webui.retrieval.web.firecrawl import scrape_firecrawl_url
68from open_webui.utils.misc import is_host_allowed, is_host_blocked
70log = logging.getLogger(__name__)
73def resolve_hostname(hostname):
74 # Get address information
75 addr_info = socket.getaddrinfo(hostname, None)
77 # Extract IP addresses from address information
78 ipv4_addresses = [info[4][0] for info in addr_info if info[0] == socket.AF_INET]
79 ipv6_addresses = [info[4][0] for info in addr_info if info[0] == socket.AF_INET6]
81 return ipv4_addresses, ipv6_addresses
84def _embedded_ipv4(addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> list[ipaddress.IPv4Address]:
85 """The IPv4 addresses an IPv6 address carries: mapped, compatible, 6to4, teredo and NAT64."""
86 if not isinstance(addr, ipaddress.IPv6Address):
87 return []
89 embedded = []
90 if addr.ipv4_mapped:
91 embedded.append(addr.ipv4_mapped)
92 if addr.sixtofour:
93 embedded.append(addr.sixtofour)
94 if addr.teredo:
95 embedded.extend(addr.teredo)
97 b = addr.packed
98 # Prefixes that put the address in the last four bytes: v4-compatible and NAT64 /96.
99 if b[:12] in (b'\x00' * 12, b'\x00\x64\xff\x9b' + b'\x00' * 8):
100 embedded.append(ipaddress.IPv4Address(b[12:]))
101 elif b[:6] == b'\x00\x64\xff\x9b\x00\x01':
102 embedded.append(ipaddress.IPv4Address(bytes((b[6], b[7], b[9], b[10]))))
104 return embedded
107def _assert_host_allowed(host: str | None) -> None:
108 if WEB_FETCH_FILTER_LIST and not is_host_allowed(host, WEB_FETCH_FILTER_LIST):
109 log.warning(f'Blocked by filter list: {host}')
110 raise ValueError(ERROR_MESSAGES.INVALID_URL)
113def _assert_addresses_allowed(addresses: Sequence[str]) -> None:
114 # An IPv6 address can carry a blocked IPv4 address inside it, so judge both spellings.
115 parsed = [ipaddress.ip_address(address) for address in addresses]
116 candidates = [*parsed, *(ipv4 for address in parsed for ipv4 in _embedded_ipv4(address))]
118 # Block entries only: an allow entry names a host, so judging a resolved address against one
119 # would reject every allow-listed host.
120 if is_host_blocked([str(address) for address in candidates], WEB_FETCH_FILTER_LIST):
121 log.warning(f'Blocked by filter list: {", ".join(str(address) for address in candidates)}')
122 raise ValueError(ERROR_MESSAGES.INVALID_URL)
124 if not ENABLE_LOCAL_WEB_FETCH:
125 for address in candidates:
126 if not address.is_global:
127 log.warning(f'Blocked non-global address: {address}')
128 raise ValueError(ERROR_MESSAGES.INVALID_URL)
131def validate_url(url: Union[str, Sequence[str]]):
132 if isinstance(url, str): 132 ↛ 168line 132 didn't jump to line 168 because the condition on line 132 was always true
133 if isinstance(validators.url(url, simple_host=ENABLE_LOCAL_WEB_FETCH), validators.ValidationError): 133 ↛ 140line 133 didn't jump to line 140 because the condition on line 133 was always true
134 raise ValueError(ERROR_MESSAGES.INVALID_URL)
136 # Reject parser-confusing chars: urlparse and requests/aiohttp split
137 # on these differently, e.g. http://127.0.0.1\@1.1.1.1 → urlparse
138 # extracts 1.1.1.1 (public, passes filter) while requests connects
139 # to 127.0.0.1 (internal). Same shape with tab/CR/LF.
140 if any(ch in url for ch in ('\\', '\t', '\n', '\r')):
141 log.warning(f'Blocked URL with parser-confusing char: {url!r}')
142 raise ValueError(ERROR_MESSAGES.INVALID_URL)
144 parsed_url = urllib.parse.urlparse(url)
146 # Protocol validation - only allow http/https
147 if parsed_url.scheme not in ['http', 'https']:
148 log.warning(f'Blocked non-HTTP(S) protocol: {parsed_url.scheme} in URL: {url}')
149 raise ValueError(ERROR_MESSAGES.INVALID_URL)
151 # Match on the parsed hostname, not the full URL: a path component would
152 # otherwise let any URL slip past a hostname-based block/allow entry.
153 _assert_host_allowed(parsed_url.hostname)
155 try:
156 ipv4_addresses, ipv6_addresses = resolve_hostname(parsed_url.hostname)
157 except (socket.gaierror, UnicodeError) as e:
158 # With local fetch on, a proxied deployment can carry names only the proxy resolves.
159 if not ENABLE_LOCAL_WEB_FETCH:
160 log.warning(f'Could not resolve host {parsed_url.hostname}: {e}')
161 raise ValueError(ERROR_MESSAGES.INVALID_URL) from None
162 ipv4_addresses, ipv6_addresses = [], []
164 # A hostname match alone lets a DNS record point at a blocked address.
165 # DNS rebinding is mitigated at the connection layer; see _SSRFSafeConnector / _SSRFSafeAdapter
166 _assert_addresses_allowed(ipv4_addresses + ipv6_addresses)
167 return True
168 elif isinstance(url, Sequence):
169 return all(validate_url(u) for u in url)
170 else:
171 return False
174def safe_validate_urls(url: Sequence[str]) -> Sequence[str]:
175 valid_urls = []
176 for u in url:
177 try:
178 if validate_url(u):
179 valid_urls.append(u)
180 except Exception as e:
181 log.debug('Invalid URL %s: %s', u, e)
182 continue
183 return valid_urls
186def _ssrf_safe_new_conn(self):
187 """Resolve DNS, screen every resolved address, connect to one of them.
189 Replaces urllib3's _new_conn so the DNS lookup that feeds the actual TCP
190 connect is the same one we validate — no second resolution, no rebinding
191 window.
192 """
193 host = getattr(self, '_dns_host', self.host)
194 port = self.port
195 infos = socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM)
196 if not infos:
197 raise OSError(f'getaddrinfo for {host!r} returned empty list')
198 _assert_addresses_allowed([sa[0] for _, _, _, _, sa in infos])
199 err = None
200 for fam, typ, proto, _, sa in infos:
201 sock = None
202 try:
203 sock = socket.socket(fam, typ, proto)
204 if self.timeout is not socket._GLOBAL_DEFAULT_TIMEOUT:
205 sock.settimeout(self.timeout)
206 if getattr(self, 'source_address', None):
207 sock.bind(self.source_address)
208 for opt in getattr(self, 'socket_options', None) or ():
209 if len(opt) == 4 and isinstance(opt[3], str):
210 # urllib3-future per-protocol form: (level, optname, value, "tcp"/"udp")
211 if opt[3].lower() == 'tcp':
212 sock.setsockopt(*opt[:3])
213 continue
214 sock.setsockopt(*opt)
215 sock.connect(sa)
216 return sock
217 except OSError as exc:
218 err = exc
219 if sock is not None:
220 sock.close()
221 raise err or OSError(f'connect to {host!r}:{port} failed')
224class _SafeHTTPConn(urllib3.connection.HTTPConnection):
225 _new_conn = _ssrf_safe_new_conn
228class _SafeHTTPSConn(urllib3.connection.HTTPSConnection):
229 _new_conn = _ssrf_safe_new_conn
232class _SafeHTTPPool(urllib3.connectionpool.HTTPConnectionPool):
233 ConnectionCls = _SafeHTTPConn
236class _SafeHTTPSPool(urllib3.connectionpool.HTTPSConnectionPool):
237 ConnectionCls = _SafeHTTPSConn
240class _SSRFSafeAdapter(HTTPAdapter):
241 """requests adapter that rejects filter-listed request targets and non-global IPs at connect time."""
243 def init_poolmanager(self, *args, **kwargs):
244 super().init_poolmanager(*args, **kwargs)
245 self.poolmanager.pool_classes_by_scheme = {
246 'http': _SafeHTTPPool,
247 'https': _SafeHTTPSPool,
248 }
250 def send(self, request, *args, **kwargs):
251 # Per request, not per connection: the connection layer sees the proxy.
252 _assert_host_allowed(urllib.parse.urlparse(request.url).hostname)
253 return super().send(request, *args, **kwargs)
256class _SSRFSafeConnector(aiohttp.TCPConnector):
257 """Rejects filter-listed request targets, and non-global IPs on each new connection."""
259 async def connect(self, req, traces, timeout):
260 # Per request, not per connection: _resolve_host sees the proxy and pooled reuse skips it.
261 _assert_host_allowed(req.url.host)
262 return await super().connect(req, traces, timeout)
264 async def _resolve_host(self, host, port, traces=None):
265 # aiohttp answers IP-literal hosts itself without consulting a resolver.
266 results = await super()._resolve_host(host, port, traces=traces)
267 _assert_addresses_allowed([entry['host'] for entry in results])
268 return results
271def get_ssrf_safe_session(trust_env: bool = True, store_cookies: bool = True) -> aiohttp.ClientSession:
272 """A one-off aiohttp session that re-validates every connection via _SSRFSafeConnector,
273 defeating DNS rebinding. Use for validate_url-gated fetches of user-supplied URLs that must
274 not use the shared (rebinding-vulnerable) pool. Use as a context manager so it is closed:
275 ``async with get_ssrf_safe_session() as session: ...``.
277 trust_env also enables environment proxies, and proxied traffic bypasses the connect-time
278 IP check, because the proxy resolves the hostname instead.
279 """
280 return aiohttp.ClientSession(
281 connector=_SSRFSafeConnector(),
282 timeout=aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT),
283 trust_env=trust_env,
284 cookie_jar=None if store_cookies else aiohttp.DummyCookieJar(),
285 )
288def get_ssrf_safe_requests_session(trust_env: bool = True, store_cookies: bool = True) -> requests.Session:
289 """The requests counterpart of get_ssrf_safe_session, with the same proxy caveat."""
290 session = requests.Session()
291 session.trust_env = trust_env
292 if not store_cookies:
293 session.cookies.set_policy(http.cookiejar.DefaultCookiePolicy(allowed_domains=[]))
294 session.mount('http://', _SSRFSafeAdapter())
295 session.mount('https://', _SSRFSafeAdapter())
296 return session
299# accept-encoding goes because the client must advertise only codecs it can decode, the rest
300# because the client derives them from the URL and body it is actually given. content-encoding
301# stays: the browser's body is forwarded byte for byte, so its own labelling still applies.
302_DROPPED_REQUEST_HEADERS = {'accept-encoding', 'connection', 'content-length', 'host', 'transfer-encoding'}
304# The clients hand us a decoded body, so the sender's framing no longer describes it.
305_DROPPED_RESPONSE_HEADERS = {'connection', 'content-encoding', 'content-length', 'transfer-encoding'}
307# The Playwright loader only reads the page HTML, which none of these feed.
308_DROPPED_RESOURCE_TYPES = {'font', 'image', 'media'}
311def _forwardable_request_headers(headers: Dict[str, str]) -> Dict[str, str]:
312 return {name: value for name, value in headers.items() if name.lower() not in _DROPPED_REQUEST_HEADERS}
315def _fulfillable_response_headers(header_pairs: Iterable[Tuple[str, str]]) -> Dict[str, str]:
316 """Collapse repeated headers the way route.fulfill expects: set-cookie by newline, rest by comma.
318 Takes pairs rather than a mapping because reading either client's headers as a mapping loses
319 duplicate Set-Cookie values, leaving one malformed cookie or one of the two.
320 """
321 collected: Dict[str, List[str]] = {}
322 for name, value in header_pairs:
323 name = name.lower() # grouping by the sender's case would split a repeated header
324 if name not in _DROPPED_RESPONSE_HEADERS:
325 collected.setdefault(name, []).append(value)
326 return {name: ('\n' if name == 'set-cookie' else ', ').join(values) for name, values in collected.items()}
329def extract_metadata(soup, url):
330 metadata = {'source': url}
331 if title := soup.find('title'):
332 metadata['title'] = title.get_text()
333 if description := soup.find('meta', attrs={'name': 'description'}):
334 metadata['description'] = description.get('content', 'No description found.')
335 if html := soup.find('html'):
336 metadata['language'] = html.get('lang', 'No language found.')
337 return metadata
340def verify_ssl_cert(url: str) -> bool:
341 """Verify SSL certificate for the given URL."""
342 if not url.startswith('https://'):
343 return True
345 try:
346 hostname = url.split('://')[-1].split('/')[0]
347 context = ssl.create_default_context(cafile=certifi.where())
348 with context.wrap_socket(ssl.socket(), server_hostname=hostname) as s:
349 s.connect((hostname, 443))
350 return True
351 except ssl.SSLError:
352 return False
353 except Exception as e:
354 log.warning(f'SSL verification failed for {url}: {str(e)}')
355 return False
358class RateLimitMixin:
359 async def _wait_for_rate_limit(self):
360 """Wait to respect the rate limit if specified."""
361 if self.requests_per_second and self.last_request_time:
362 min_interval = timedelta(seconds=1.0 / self.requests_per_second)
363 time_since_last = datetime.now() - self.last_request_time
364 if time_since_last < min_interval:
365 await asyncio.sleep((min_interval - time_since_last).total_seconds())
366 self.last_request_time = datetime.now()
368 def _sync_wait_for_rate_limit(self):
369 """Synchronous version of rate limit wait."""
370 if self.requests_per_second and self.last_request_time:
371 min_interval = timedelta(seconds=1.0 / self.requests_per_second)
372 time_since_last = datetime.now() - self.last_request_time
373 if time_since_last < min_interval:
374 time.sleep((min_interval - time_since_last).total_seconds())
375 self.last_request_time = datetime.now()
378class URLProcessingMixin:
379 async def _verify_ssl_cert(self, url: str) -> bool:
380 """Verify SSL certificate for a URL."""
381 return await run_in_threadpool(verify_ssl_cert, url)
383 async def _safe_process_url(self, url: str) -> bool:
384 """Perform safety checks before processing a URL."""
385 if self.verify_ssl and not await self._verify_ssl_cert(url):
386 raise ValueError(f'SSL certificate verification failed for {url}')
387 await self._wait_for_rate_limit()
388 return True
390 def _safe_process_url_sync(self, url: str) -> bool:
391 """Synchronous version of safety checks."""
392 if self.verify_ssl and not verify_ssl_cert(url):
393 raise ValueError(f'SSL certificate verification failed for {url}')
394 self._sync_wait_for_rate_limit()
395 return True
398class SafeFireCrawlLoader(BaseLoader, RateLimitMixin, URLProcessingMixin):
399 def __init__(
400 self,
401 web_paths,
402 verify_ssl: bool = True,
403 trust_env: bool = False,
404 requests_per_second: Optional[float] = None,
405 continue_on_failure: bool = True,
406 api_key: Optional[str] = None,
407 api_url: Optional[str] = None,
408 timeout: Optional[int] = None,
409 mode: Literal['crawl', 'scrape', 'map'] = 'scrape',
410 proxy: Optional[Dict[str, str]] = None,
411 params: Optional[Dict] = None,
412 ):
413 proxy_server = proxy.get('server') if proxy else None
414 if trust_env and not proxy_server:
415 env_proxies = urllib.request.getproxies()
416 env_proxy_server = env_proxies.get('https') or env_proxies.get('http')
417 if env_proxy_server:
418 if proxy:
419 proxy['server'] = env_proxy_server
420 else:
421 proxy = {'server': env_proxy_server}
422 self.web_paths = web_paths
423 self.verify_ssl = verify_ssl
424 self.requests_per_second = requests_per_second
425 self.last_request_time = None
426 self.trust_env = trust_env
427 self.continue_on_failure = continue_on_failure
428 self.api_key = api_key
429 self.api_url = (api_url or 'https://api.firecrawl.dev').rstrip('/')
430 self.timeout = timeout
431 self.mode = mode
432 self.params = params or {}
434 def lazy_load(self) -> Iterator[Document]:
435 for url in self.web_paths:
436 try:
437 self._sync_wait_for_rate_limit()
438 doc = scrape_firecrawl_url(
439 self.api_url,
440 self.api_key,
441 url,
442 verify_ssl=self.verify_ssl,
443 timeout=self.timeout,
444 params=self.params,
445 )
446 if doc is not None:
447 yield doc
448 except Exception as e:
449 if self.continue_on_failure:
450 log.warning(f'Error extracting content from {url} with Firecrawl: {e}')
451 continue
452 raise
454 async def alazy_load(self):
455 for url in self.web_paths:
456 try:
457 await self._wait_for_rate_limit()
458 doc = await run_in_threadpool(
459 scrape_firecrawl_url,
460 self.api_url,
461 self.api_key,
462 url,
463 verify_ssl=self.verify_ssl,
464 timeout=self.timeout,
465 params=self.params,
466 )
467 if doc is not None:
468 yield doc
469 except Exception as e:
470 if self.continue_on_failure:
471 log.warning(f'Error extracting content from {url} with Firecrawl: {e}')
472 continue
473 raise
476class SafeTavilyLoader(BaseLoader, RateLimitMixin, URLProcessingMixin):
477 def __init__(
478 self,
479 web_paths: Union[str, List[str]],
480 api_key: str,
481 extract_depth: Literal['basic', 'advanced'] = 'basic',
482 continue_on_failure: bool = True,
483 requests_per_second: Optional[float] = None,
484 verify_ssl: bool = True,
485 trust_env: bool = False,
486 proxy: Optional[Dict[str, str]] = None,
487 ):
488 """Initialize SafeTavilyLoader with rate limiting and SSL verification support.
490 Args:
491 web_paths: List of URLs/paths to process.
492 api_key: The Tavily API key.
493 extract_depth: Depth of extraction ("basic" or "advanced").
494 continue_on_failure: Whether to continue if extraction of a URL fails.
495 requests_per_second: Number of requests per second to limit to.
496 verify_ssl: If True, verify SSL certificates.
497 trust_env: If True, use proxy settings from environment variables.
498 proxy: Optional proxy configuration.
499 """
500 # Initialize proxy configuration if using environment variables
501 proxy_server = proxy.get('server') if proxy else None
502 if trust_env and not proxy_server:
503 env_proxies = urllib.request.getproxies()
504 env_proxy_server = env_proxies.get('https') or env_proxies.get('http')
505 if env_proxy_server:
506 if proxy:
507 proxy['server'] = env_proxy_server
508 else:
509 proxy = {'server': env_proxy_server}
511 # Store parameters for creating TavilyLoader instances
512 self.web_paths = web_paths if isinstance(web_paths, list) else [web_paths]
513 self.api_key = api_key
514 self.extract_depth = extract_depth
515 self.continue_on_failure = continue_on_failure
516 self.verify_ssl = verify_ssl
517 self.trust_env = trust_env
518 self.proxy = proxy
520 # Add rate limiting
521 self.requests_per_second = requests_per_second
522 self.last_request_time = None
524 def lazy_load(self) -> Iterator[Document]:
525 """Load documents with rate limiting support, delegating to TavilyLoader."""
526 valid_urls = []
527 for url in self.web_paths:
528 try:
529 self._safe_process_url_sync(url)
530 valid_urls.append(url)
531 except Exception as e:
532 log.warning(f'SSL verification failed for {url}: {str(e)}')
533 if not self.continue_on_failure:
534 raise e
535 if not valid_urls:
536 if self.continue_on_failure:
537 log.warning('No valid URLs to process after SSL verification')
538 return
539 raise ValueError('No valid URLs to process after SSL verification')
540 try:
541 loader = TavilyLoader(
542 urls=valid_urls,
543 api_key=self.api_key,
544 extract_depth=self.extract_depth,
545 continue_on_failure=self.continue_on_failure,
546 )
547 yield from loader.lazy_load()
548 except Exception as e:
549 if self.continue_on_failure:
550 log.exception(f'Error extracting content from URLs: {e}')
551 else:
552 raise e
554 async def alazy_load(self) -> AsyncIterator[Document]:
555 """Async version with rate limiting and SSL verification."""
556 valid_urls = []
557 for url in self.web_paths:
558 try:
559 await self._safe_process_url(url)
560 valid_urls.append(url)
561 except Exception as e:
562 log.warning(f'SSL verification failed for {url}: {str(e)}')
563 if not self.continue_on_failure:
564 raise e
566 if not valid_urls:
567 if self.continue_on_failure:
568 log.warning('No valid URLs to process after SSL verification')
569 return
570 raise ValueError('No valid URLs to process after SSL verification')
572 try:
573 loader = TavilyLoader(
574 urls=valid_urls,
575 api_key=self.api_key,
576 extract_depth=self.extract_depth,
577 continue_on_failure=self.continue_on_failure,
578 )
579 async for document in loader.alazy_load():
580 yield document
581 except Exception as e:
582 if self.continue_on_failure:
583 log.exception(f'Error loading URLs: {e}')
584 else:
585 raise e
588class SafeMicrosoftWebIQLoader(BaseLoader, RateLimitMixin, URLProcessingMixin):
589 def __init__(
590 self,
591 web_paths: Union[str, List[str]],
592 api_key: str,
593 api_base_url: str = MICROSOFT_WEB_IQ_API_BASE_URL,
594 language: str = 'en',
595 verify_ssl: bool = True,
596 trust_env: bool = False,
597 requests_per_second: Optional[float] = None,
598 continue_on_failure: bool = True,
599 timeout: Optional[int] = None,
600 ):
601 self.web_paths = web_paths if isinstance(web_paths, list) else [web_paths]
602 self.api_key = api_key
603 self.api_base_url = api_base_url
604 self.language = language
605 self.verify_ssl = verify_ssl
606 self.trust_env = trust_env
607 self.requests_per_second = requests_per_second
608 self.last_request_time = None
609 self.continue_on_failure = continue_on_failure
610 self.timeout = timeout
612 def lazy_load(self) -> Iterator[Document]:
613 valid_urls = []
614 for url in self.web_paths:
615 try:
616 self._safe_process_url_sync(url)
617 valid_urls.append(url)
618 except Exception as e:
619 log.warning(f'SSL verification failed for {url}: {str(e)}')
620 if not self.continue_on_failure:
621 raise e
622 if not valid_urls:
623 if self.continue_on_failure:
624 log.warning('No valid URLs to process after SSL verification')
625 return
626 raise ValueError('No valid URLs to process after SSL verification')
628 loader = MicrosoftWebIQLoader(
629 urls=valid_urls,
630 api_base_url=self.api_base_url,
631 api_key=self.api_key,
632 language=self.language,
633 verify_ssl=self.verify_ssl,
634 timeout=self.timeout,
635 continue_on_failure=self.continue_on_failure,
636 )
637 yield from loader.lazy_load()
639 async def alazy_load(self) -> AsyncIterator[Document]:
640 try:
641 docs = await run_in_threadpool(lambda: list(self.lazy_load()))
642 for doc in docs:
643 yield doc
644 except Exception as e:
645 if self.continue_on_failure:
646 log.warning(f'Error browsing URLs with Microsoft Web IQ: {e}')
647 else:
648 raise e
651class SafePlaywrightURLLoader(BaseLoader, RateLimitMixin, URLProcessingMixin):
652 """Load HTML pages safely with Playwright, supporting SSL verification, rate limiting, and remote browser connection.
654 Attributes:
655 web_paths (List[str]): List of URLs to load.
656 verify_ssl (bool): If True, verify SSL certificates.
657 trust_env (bool): If True, use proxy settings from environment variables.
658 requests_per_second (Optional[float]): Number of requests per second to limit to.
659 continue_on_failure (bool): If True, continue loading other URLs on failure.
660 headless (bool): If True, the browser will run in headless mode.
661 proxy (dict): Proxy override settings for the Playwright session. Page requests are
662 issued outside the browser, so they follow the environment proxy via trust_env
663 rather than this setting.
664 playwright_ws_url (Optional[str]): WebSocket endpoint URI for remote browser connection.
665 playwright_timeout (Optional[int]): Maximum operation time in milliseconds.
666 """
668 def __init__(
669 self,
670 web_paths: List[str],
671 verify_ssl: bool = True,
672 trust_env: bool = False,
673 requests_per_second: Optional[float] = None,
674 continue_on_failure: bool = True,
675 headless: bool = True,
676 remove_selectors: Optional[List[str]] = None,
677 proxy: Optional[Dict[str, str]] = None,
678 playwright_ws_url: Optional[str] = None,
679 playwright_timeout: Optional[int] = 10000,
680 ):
681 """Initialize with additional safety parameters and remote browser support."""
682 if USE_SLIM:
683 raise HTTPException(
684 503, 'Playwright is unavailable in slim. Use basic HTTP fetching or an external web loader.'
685 )
687 for package in ('playwright', 'unstructured'):
688 import_module(package)
690 proxy_server = proxy.get('server') if proxy else None
691 if trust_env and not proxy_server:
692 env_proxies = urllib.request.getproxies()
693 env_proxy_server = env_proxies.get('https') or env_proxies.get('http')
694 if env_proxy_server:
695 if proxy:
696 proxy['server'] = env_proxy_server
697 else:
698 proxy = {'server': env_proxy_server}
700 self.urls = web_paths
701 self.continue_on_failure = continue_on_failure
702 self.headless = headless if playwright_ws_url is None else False
703 self.remove_selectors = remove_selectors or []
704 self.proxy = proxy
705 self.verify_ssl = verify_ssl
706 self.requests_per_second = requests_per_second
707 self.last_request_time = None
708 self.playwright_ws_url = playwright_ws_url
709 self.trust_env = trust_env
710 self.playwright_timeout = playwright_timeout
712 @staticmethod
713 def _extract_html(html):
714 from unstructured.partition.html import partition_html
716 return '\n\n'.join(str(element) for element in partition_html(text=html))
718 def _request_timeout(self) -> float:
719 # per-hop budget, since page.goto's timeout cannot reach into our own fetch and 0 disables
720 # it. aiohttp treats it as a total where requests only caps each read, so sync runs looser.
721 return (self.playwright_timeout or 30000) / 1000
723 def _requests_verify(self) -> Union[bool, str]:
724 """requests takes a CA path where aiohttp takes the parsed SSLContext.
726 A bundle named directly in AIOHTTP_CLIENT_SESSION_SSL reaches us already parsed and
727 cannot be expressed here, so that form falls back to the global bundle or certifi.
728 """
729 if not self.verify_ssl or AIOHTTP_CLIENT_SESSION_SSL is False:
730 return False
731 if AIOHTTP_CLIENT_SESSION_SSL is True:
732 return True # no usable global CA bundle, so both clients land on certifi
733 return AIOHTTP_CLIENT_SSL_CERT_FILE or True
735 def _intercept_navigation_sync(self, route, session):
736 req = route.request
737 if req.resource_type in _DROPPED_RESOURCE_TYPES:
738 route.abort()
739 return
741 hop_cookies: List[Tuple[str, str]] = []
743 try:
744 headers = _forwardable_request_headers(req.all_headers())
745 post_data = req.post_data_buffer
746 verify, timeout = self._requests_verify(), self._request_timeout()
748 # The browser would resolve the hostname again, after the check; fetch it ourselves.
749 def fetch(url):
750 validate_url(url)
751 return session.request(
752 req.method,
753 url,
754 headers=headers,
755 data=post_data,
756 allow_redirects=False,
757 verify=verify,
758 timeout=timeout,
759 )
761 resp = fetch(req.url)
763 if 300 <= resp.status_code < 400:
764 for _ in range(20):
765 if not AIOHTTP_CLIENT_ALLOW_REDIRECTS:
766 route.abort()
767 return
769 location = resp.headers.get('location')
770 if not location:
771 break
773 # only the last hop is fulfilled, so carry each hop's cookies to the browser
774 hop_cookies += [('set-cookie', v) for v in resp.raw.headers.getlist('set-cookie')]
775 resp = fetch(urllib.parse.urljoin(resp.url, location))
776 if not 300 <= resp.status_code < 400:
777 break
778 else:
779 route.abort()
780 return
781 except Exception as e:
782 log.debug('Playwright loader could not fetch %s: %s', req.url, e)
783 route.abort()
784 return
786 route.fulfill(
787 status=resp.status_code,
788 headers=_fulfillable_response_headers(hop_cookies + list(resp.raw.headers.items())),
789 body=resp.content,
790 )
792 async def _intercept_navigation(self, route, session):
793 req = route.request
794 if req.resource_type in _DROPPED_RESOURCE_TYPES:
795 await route.abort()
796 return
798 hop_cookies: List[Tuple[str, str]] = []
800 try:
801 headers = _forwardable_request_headers(await req.all_headers())
802 post_data = req.post_data_buffer
804 # The browser would resolve the hostname again, after the check; fetch it ourselves.
805 async def fetch(url):
806 await run_in_threadpool(validate_url, url)
807 response = await session.request(
808 req.method,
809 url,
810 headers=headers,
811 data=post_data,
812 allow_redirects=False,
813 ssl=AIOHTTP_CLIENT_SESSION_SSL if self.verify_ssl else False,
814 timeout=aiohttp.ClientTimeout(total=self._request_timeout()),
815 )
816 # aiohttp only returns the connection to the pool once the body is buffered
817 return response, await response.read()
819 resp, body = await fetch(req.url)
821 if 300 <= resp.status < 400:
822 for _ in range(20):
823 if not AIOHTTP_CLIENT_ALLOW_REDIRECTS:
824 await route.abort()
825 return
827 location = resp.headers.get('location')
828 if not location:
829 break
831 # only the last hop is fulfilled, so carry each hop's cookies to the browser
832 hop_cookies += [('set-cookie', v) for v in resp.headers.getall('Set-Cookie', [])]
833 resp, body = await fetch(urllib.parse.urljoin(str(resp.url), location))
834 if not 300 <= resp.status < 400:
835 break
836 else:
837 await route.abort()
838 return
839 except Exception as e:
840 log.debug('Playwright loader could not fetch %s: %s', req.url, e)
841 await route.abort()
842 return
844 await route.fulfill(
845 status=resp.status,
846 headers=_fulfillable_response_headers(hop_cookies + list(resp.headers.items())),
847 body=body,
848 )
850 def lazy_load(self) -> Iterator[Document]:
851 """Safely load URLs synchronously with support for remote browser."""
852 from playwright.sync_api import sync_playwright
854 with sync_playwright() as p:
855 # Use remote browser if ws_endpoint is provided, otherwise use local browser
856 if self.playwright_ws_url:
857 browser = p.chromium.connect(self.playwright_ws_url)
858 else:
859 browser = p.chromium.launch(headless=self.headless, proxy=self.proxy)
861 with browser:
862 for url in self.urls:
863 try:
864 self._safe_process_url_sync(url)
865 # opened before the page so it outlives any route still in flight at teardown
866 with (
867 get_ssrf_safe_requests_session(self.trust_env, store_cookies=False) as session,
868 browser.new_page(service_workers='block') as page,
869 ):
870 page.route('**/*', lambda route: self._intercept_navigation_sync(route, session))
871 # sync close() hangs the dispatcher; a no-op handler still never connects to the server
872 page.route_web_socket('**/*', lambda ws_route: None)
873 response = page.goto(url, timeout=self.playwright_timeout)
874 if response is None:
875 raise ValueError(f'page.goto() returned None for url {url}')
877 for selector in self.remove_selectors:
878 for element in page.locator(selector).all():
879 if element.is_visible():
880 element.evaluate('element => element.remove()')
881 text = self._extract_html(page.content())
882 page.unroute_all(behavior='ignoreErrors')
883 metadata = {'source': url}
884 yield Document(page_content=text, metadata=metadata)
885 except Exception as e:
886 if self.continue_on_failure:
887 log.exception(f'Error loading {url}: {e}')
888 continue
889 raise e
891 async def alazy_load(self) -> AsyncIterator[Document]:
892 """Safely load URLs asynchronously with support for remote browser."""
893 from playwright.async_api import async_playwright
895 async with async_playwright() as p:
896 # Use remote browser if ws_endpoint is provided, otherwise use local browser
897 if self.playwright_ws_url:
898 browser = await p.chromium.connect(self.playwright_ws_url)
899 else:
900 browser = await p.chromium.launch(headless=self.headless, proxy=self.proxy)
902 async with browser:
903 for url in self.urls:
904 try:
905 await self._safe_process_url(url)
906 # opened before the page so it outlives any route still in flight at teardown
907 async with (
908 get_ssrf_safe_session(self.trust_env, store_cookies=False) as session,
909 await browser.new_page(service_workers='block') as page,
910 ):
911 await page.route('**/*', lambda route: self._intercept_navigation(route, session))
912 await page.route_web_socket('**/*', lambda ws_route: ws_route.close())
913 response = await page.goto(url, timeout=self.playwright_timeout)
914 if response is None:
915 raise ValueError(f'page.goto() returned None for url {url}')
917 for selector in self.remove_selectors:
918 for element in await page.locator(selector).all():
919 if await element.is_visible():
920 await element.evaluate('element => element.remove()')
921 text = await asyncio.to_thread(self._extract_html, await page.content())
922 await page.unroute_all(behavior='ignoreErrors')
923 metadata = {'source': url}
924 yield Document(page_content=text, metadata=metadata)
925 except Exception as e:
926 if self.continue_on_failure:
927 log.exception(f'Error loading {url}: {e}')
928 continue
929 raise e
932class SafeWebBaseLoader(BaseLoader):
933 """Fetch pages with connect-time address checks and bounded concurrency."""
935 def __init__(
936 self,
937 web_paths,
938 verify_ssl=True,
939 trust_env=False,
940 requests_per_second=2,
941 continue_on_failure=False,
942 requests_kwargs=None,
943 raise_for_status=False,
944 default_parser='lxml',
945 bs_kwargs=None,
946 bs_get_text_kwargs=None,
947 ):
948 self.web_paths = list(web_paths)
949 self.trust_env = trust_env
950 self.requests_per_second = requests_per_second
951 self.continue_on_failure = continue_on_failure
952 self.requests_kwargs = {**(requests_kwargs or {}), 'allow_redirects': AIOHTTP_CLIENT_ALLOW_REDIRECTS}
953 self.raise_for_status = raise_for_status
954 self.default_parser = default_parser
955 self.bs_kwargs = bs_kwargs or {}
956 self.bs_get_text_kwargs = bs_get_text_kwargs or {}
957 # Preserve the synchronous loader's environment-proxy behavior.
958 self.session = get_ssrf_safe_requests_session()
959 self.session.verify = verify_ssl
960 self.session.headers = {
961 'User-Agent': USER_AGENT or 'DefaultLangchainUserAgent',
962 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
963 'Accept-Language': 'en-US,en;q=0.5',
964 'Referer': 'https://www.google.com/',
965 'DNT': '1',
966 'Connection': 'keep-alive',
967 'Upgrade-Insecure-Requests': '1',
968 }
970 async def fetch_all(self, urls):
971 semaphore = asyncio.Semaphore(self.requests_per_second)
973 async def fetch(url):
974 async with semaphore:
975 try:
976 return await self._fetch(url)
977 except Exception as e:
978 if not self.continue_on_failure:
979 raise
980 log.warning('Error fetching %s: %s', url, e)
981 return ''
983 return await asyncio.gather(*(fetch(url) for url in urls))
985 async def _fetch(self, url: str, retries: int = 3, cooldown: int = 2, backoff: float = 1.5) -> str:
986 connector = _SSRFSafeConnector()
987 async with aiohttp.ClientSession(trust_env=self.trust_env, connector=connector) as session:
988 for i in range(retries):
989 try:
990 kwargs: Dict = dict(
991 headers=self.session.headers,
992 cookies=self.session.cookies.get_dict(),
993 )
994 if not self.session.verify:
995 kwargs['ssl'] = False
996 else:
997 kwargs['ssl'] = AIOHTTP_CLIENT_SESSION_SSL
999 options = self.requests_kwargs | kwargs
1000 if isinstance(options.get('timeout'), (int, float)):
1001 options['timeout'] = aiohttp.ClientTimeout(total=options['timeout'])
1002 async with session.get(url, **options) as response:
1003 if self.raise_for_status:
1004 response.raise_for_status()
1005 return await response.text()
1006 except aiohttp.ClientConnectionError as e:
1007 if i == retries - 1:
1008 raise
1009 else:
1010 log.warning(f'Error fetching {url} with attempt {i + 1}/{retries}: {e}. Retrying...')
1011 await asyncio.sleep(cooldown * backoff**i)
1012 raise ValueError('retry count exceeded')
1014 def lazy_load(self) -> Iterator[Document]:
1015 """Lazy load text from the url(s) in web_path with error handling."""
1016 for path in self.web_paths:
1017 try:
1018 with self.session.get(path, **self.requests_kwargs) as response:
1019 if self.raise_for_status:
1020 response.raise_for_status()
1021 response.encoding = response.apparent_encoding
1022 yield self._document_from_html(response.text, path)
1023 except Exception as e:
1024 log.exception(f'Error loading {path}: {e}')
1026 def _document_from_html(self, html: str, url: str) -> Document:
1027 """Build one Document."""
1028 from bs4 import BeautifulSoup
1030 parser = 'xml' if url.endswith('.xml') else self.default_parser
1031 soup = BeautifulSoup(html, parser, **self.bs_kwargs)
1032 return Document(
1033 page_content=soup.get_text(**self.bs_get_text_kwargs),
1034 metadata=extract_metadata(soup, url),
1035 )
1037 async def alazy_load(self) -> AsyncIterator[Document]:
1038 """Async lazy load text from the url(s) in web_path."""
1039 results = await self.fetch_all(self.web_paths)
1040 for path, html in zip(self.web_paths, results):
1041 # parsing a large page costs hundreds of ms, keep it off the event loop
1042 yield await asyncio.to_thread(self._document_from_html, html, path)
1044 async def aload(self) -> list[Document]:
1045 """Load data into Document objects."""
1046 return [document async for document in self.alazy_load()]
1049def get_web_loader(
1050 urls: Union[str, Sequence[str]],
1051 verify_ssl: bool = True,
1052 requests_per_second: int = 2,
1053 trust_env: bool = False,
1054 loader_config: Optional[dict] = None,
1055):
1056 # Check if the URLs are valid
1057 safe_urls = safe_validate_urls([urls] if isinstance(urls, str) else urls)
1059 if not safe_urls:
1060 log.warning(f'All provided URLs were blocked or invalid: {urls}')
1061 raise ValueError(ERROR_MESSAGES.INVALID_URL)
1063 loader_config = loader_config or {}
1065 def cfg(key, env_value):
1066 # Admin-saved DB value wins; env constant covers keys never saved.
1067 value = loader_config.get(key)
1068 return env_value if value is None else value
1070 engine = cfg('web_loader_engine', WEB_LOADER_ENGINE)
1071 web_loader_timeout = cfg('web_loader_timeout', WEB_LOADER_TIMEOUT)
1073 web_loader_args = {
1074 'web_paths': safe_urls,
1075 'verify_ssl': verify_ssl,
1076 'requests_per_second': requests_per_second,
1077 'continue_on_failure': True,
1078 'trust_env': trust_env,
1079 }
1081 WebLoaderClass = None
1083 if engine == '' or engine == 'safe_web':
1084 WebLoaderClass = SafeWebBaseLoader
1086 request_kwargs = {}
1087 if web_loader_timeout:
1088 try:
1089 timeout_value = float(web_loader_timeout)
1090 except ValueError:
1091 timeout_value = None
1093 if timeout_value:
1094 request_kwargs['timeout'] = timeout_value
1096 if request_kwargs:
1097 web_loader_args['requests_kwargs'] = request_kwargs
1099 if engine == 'playwright':
1100 WebLoaderClass = SafePlaywrightURLLoader
1101 web_loader_args['playwright_timeout'] = cfg('playwright_timeout', PLAYWRIGHT_TIMEOUT)
1102 playwright_ws_url = cfg('playwright_ws_url', PLAYWRIGHT_WS_URL)
1103 if playwright_ws_url:
1104 web_loader_args['playwright_ws_url'] = playwright_ws_url
1106 if engine == 'firecrawl':
1107 WebLoaderClass = SafeFireCrawlLoader
1108 web_loader_args['api_key'] = cfg('firecrawl_api_key', FIRECRAWL_API_KEY)
1109 web_loader_args['api_url'] = cfg('firecrawl_api_url', FIRECRAWL_API_BASE_URL)
1110 firecrawl_timeout = cfg('firecrawl_timeout', FIRECRAWL_TIMEOUT)
1111 if firecrawl_timeout:
1112 try:
1113 web_loader_args['timeout'] = int(firecrawl_timeout)
1114 except ValueError:
1115 pass
1117 if engine == 'tavily':
1118 WebLoaderClass = SafeTavilyLoader
1119 web_loader_args['api_key'] = cfg('tavily_api_key', TAVILY_API_KEY)
1120 web_loader_args['extract_depth'] = cfg('tavily_extract_depth', TAVILY_EXTRACT_DEPTH)
1122 if engine == 'microsoft_web_iq':
1123 WebLoaderClass = SafeMicrosoftWebIQLoader
1124 web_loader_args['api_base_url'] = cfg('microsoft_web_iq_api_base_url', MICROSOFT_WEB_IQ_API_BASE_URL)
1125 web_loader_args['api_key'] = cfg('microsoft_web_iq_api_key', MICROSOFT_WEB_IQ_API_KEY)
1126 web_loader_args['language'] = cfg('microsoft_web_iq_language', MICROSOFT_WEB_IQ_LANGUAGE)
1127 if web_loader_timeout:
1128 try:
1129 web_loader_args['timeout'] = int(web_loader_timeout)
1130 except ValueError:
1131 pass
1133 if engine == 'external':
1134 WebLoaderClass = ExternalWebLoader
1135 web_loader_args['external_url'] = cfg('external_web_loader_url', EXTERNAL_WEB_LOADER_URL)
1136 web_loader_args['external_api_key'] = cfg('external_web_loader_api_key', EXTERNAL_WEB_LOADER_API_KEY)
1138 if WebLoaderClass:
1139 web_loader = WebLoaderClass(**web_loader_args)
1141 log.debug(
1142 'Using WEB_LOADER_ENGINE %s for %s URLs',
1143 web_loader.__class__.__name__,
1144 len(safe_urls),
1145 )
1147 return web_loader
1148 else:
1149 raise ValueError(
1150 f'Invalid WEB_LOADER_ENGINE: {engine}. '
1151 "Please set it to 'safe_web', 'playwright', 'firecrawl', 'tavily', 'external', or 'microsoft_web_iq'."
1152 )