open-webui/backend/open_webui/retrieval/web/utils.py

import asyncio
import http.cookiejar
import ipaddress
import logging
import socket
import ssl
import time
import urllib.parse
import urllib.request
from datetime import datetime, timedelta
from typing import (
    Any,
    AsyncIterator,
    Dict,
    Iterable,
    Iterator,
    List,
    Literal,
    Optional,
    Sequence,
    Tuple,
    Union,
)

import aiohttp
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
from langchain_core.documents import Document
from open_webui.config import (
    ENABLE_LOCAL_WEB_FETCH,
    EXTERNAL_WEB_LOADER_API_KEY,
    EXTERNAL_WEB_LOADER_URL,
    FIRECRAWL_API_BASE_URL,
    FIRECRAWL_API_KEY,
    FIRECRAWL_TIMEOUT,
    MICROSOFT_WEB_IQ_API_BASE_URL,
    MICROSOFT_WEB_IQ_API_KEY,
    MICROSOFT_WEB_IQ_LANGUAGE,
    PLAYWRIGHT_TIMEOUT,
    PLAYWRIGHT_WS_URL,
    TAVILY_API_KEY,
    TAVILY_EXTRACT_DEPTH,
    WEB_FETCH_FILTER_LIST,
    WEB_LOADER_ENGINE,
    WEB_LOADER_TIMEOUT,
)
from open_webui.constants import ERROR_MESSAGES
from open_webui.env import (
    AIOHTTP_CLIENT_ALLOW_REDIRECTS,
    AIOHTTP_CLIENT_SESSION_SSL,
    AIOHTTP_CLIENT_SSL_CERT_FILE,
    AIOHTTP_CLIENT_TIMEOUT,
    USER_AGENT,
)
from open_webui.retrieval.loaders.external_web import ExternalWebLoader
from open_webui.retrieval.loaders.microsoft_web_iq import MicrosoftWebIQLoader
from open_webui.retrieval.loaders.tavily import TavilyLoader
from open_webui.retrieval.web.firecrawl import scrape_firecrawl_url
from open_webui.utils.misc import is_host_allowed, is_host_blocked

log = logging.getLogger(__name__)


def resolve_hostname(hostname):
    # Get address information
    addr_info = socket.getaddrinfo(hostname, None)

    # Extract IP addresses from address information
    ipv4_addresses = [info[4][0] for info in addr_info if info[0] == socket.AF_INET]
    ipv6_addresses = [info[4][0] for info in addr_info if info[0] == socket.AF_INET6]

    return ipv4_addresses, ipv6_addresses


def _embedded_ipv4(addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> list[ipaddress.IPv4Address]:
    """The IPv4 addresses an IPv6 address carries: mapped, compatible, 6to4, teredo and NAT64."""
    if not isinstance(addr, ipaddress.IPv6Address):
        return []

    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
    # Prefixes that put the address in the last four bytes: v4-compatible and NAT64 /96.
    if b[:12] in (b'\x00' * 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':
        embedded.append(ipaddress.IPv4Address(bytes((b[6], b[7], b[9], b[10]))))

    return embedded


def _assert_host_allowed(host: str | None) -> None:
    if WEB_FETCH_FILTER_LIST and not is_host_allowed(host, WEB_FETCH_FILTER_LIST):
        log.warning(f'Blocked by filter list: {host}')
        raise ValueError(ERROR_MESSAGES.INVALID_URL)


def _assert_addresses_allowed(addresses: Sequence[str]) -> None:
    # An IPv6 address can carry a blocked IPv4 address inside it, so judge both spellings.
    parsed = [ipaddress.ip_address(address) for address in addresses]
    candidates = [*parsed, *(ipv4 for address in parsed for ipv4 in _embedded_ipv4(address))]

    # Block entries only: an allow entry names a host, so judging a resolved address against one
    # would reject every allow-listed host.
    if is_host_blocked([str(address) for address in candidates], WEB_FETCH_FILTER_LIST):
        log.warning(f'Blocked by filter list: {", ".join(str(address) for address in candidates)}')
        raise ValueError(ERROR_MESSAGES.INVALID_URL)

    if not ENABLE_LOCAL_WEB_FETCH:
        for address in candidates:
            if not address.is_global:
                log.warning(f'Blocked non-global address: {address}')
                raise ValueError(ERROR_MESSAGES.INVALID_URL)


def validate_url(url: Union[str, Sequence[str]]):
    if isinstance(url, str):
        if isinstance(validators.url(url), validators.ValidationError):
            raise ValueError(ERROR_MESSAGES.INVALID_URL)

        # Reject parser-confusing chars: urlparse and requests/aiohttp split
        # on these differently, e.g. http://127.0.0.1\@1.1.1.1 → urlparse
        # extracts 1.1.1.1 (public, passes filter) while requests connects
        # to 127.0.0.1 (internal). Same shape with tab/CR/LF.
        if any(ch in url for ch in ('\\', '\t', '\n', '\r')):
            log.warning(f'Blocked URL with parser-confusing char: {url!r}')
            raise ValueError(ERROR_MESSAGES.INVALID_URL)

        parsed_url = urllib.parse.urlparse(url)

        # Protocol validation - only allow http/https
        if parsed_url.scheme not in ['http', 'https']:
            log.warning(f'Blocked non-HTTP(S) protocol: {parsed_url.scheme} in URL: {url}')
            raise ValueError(ERROR_MESSAGES.INVALID_URL)

        # Match on the parsed hostname, not the full URL: a path component would
        # otherwise let any URL slip past a hostname-based block/allow entry.
        _assert_host_allowed(parsed_url.hostname)

        try:
            ipv4_addresses, ipv6_addresses = resolve_hostname(parsed_url.hostname)
        except (socket.gaierror, UnicodeError) as e:
            # With local fetch on, a proxied deployment can carry names only the proxy resolves.
            if not ENABLE_LOCAL_WEB_FETCH:
                log.warning(f'Could not resolve host {parsed_url.hostname}: {e}')
                raise ValueError(ERROR_MESSAGES.INVALID_URL) from None
            ipv4_addresses, ipv6_addresses = [], []

        # A hostname match alone lets a DNS record point at a blocked address.
        # DNS rebinding is mitigated at the connection layer; see _SSRFSafeConnector / _SSRFSafeAdapter
        _assert_addresses_allowed(ipv4_addresses + ipv6_addresses)
        return True
    elif isinstance(url, Sequence):
        return all(validate_url(u) for u in url)
    else:
        return False


def safe_validate_urls(url: Sequence[str]) -> Sequence[str]:
    valid_urls = []
    for u in url:
        try:
            if validate_url(u):
                valid_urls.append(u)
        except Exception as e:
            log.debug('Invalid URL %s: %s', u, e)
            continue
    return valid_urls


def _ssrf_safe_new_conn(self):
    """Resolve DNS, screen every resolved address, connect to one of them.

    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')
    _assert_addresses_allowed([sa[0] for _, _, _, _, sa in infos])
    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 ():
                if len(opt) == 4 and isinstance(opt[3], str):
                    # urllib3-future per-protocol form: (level, optname, value, "tcp"/"udp")
                    if opt[3].lower() == 'tcp':
                        sock.setsockopt(*opt[:3])
                    continue
                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 adapter that rejects filter-listed request targets and non-global 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,
        }

    def send(self, request, *args, **kwargs):
        # Per request, not per connection: the connection layer sees the proxy.
        _assert_host_allowed(urllib.parse.urlparse(request.url).hostname)
        return super().send(request, *args, **kwargs)


class _SSRFSafeConnector(aiohttp.TCPConnector):
    """Rejects filter-listed request targets, and non-global IPs on each new connection."""

    async def connect(self, req, traces, timeout):
        # Per request, not per connection: _resolve_host sees the proxy and pooled reuse skips it.
        _assert_host_allowed(req.url.host)
        return await super().connect(req, traces, timeout)

    async def _resolve_host(self, host, port, traces=None):
        # aiohttp answers IP-literal hosts itself without consulting a resolver.
        results = await super()._resolve_host(host, port, traces=traces)
        _assert_addresses_allowed([entry['host'] for entry in results])
        return results


def get_ssrf_safe_session(trust_env: bool = True, store_cookies: bool = True) -> aiohttp.ClientSession:
    """A one-off aiohttp session that re-validates every connection via _SSRFSafeConnector,
    defeating DNS rebinding. Use for validate_url-gated fetches of user-supplied URLs that must
    not use the shared (rebinding-vulnerable) pool. Use as a context manager so it is closed:
    ``async with get_ssrf_safe_session() as session: ...``.

    trust_env also enables environment proxies, and proxied traffic bypasses the connect-time
    IP check, because the proxy resolves the hostname instead.
    """
    return aiohttp.ClientSession(
        connector=_SSRFSafeConnector(),
        timeout=aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT),
        trust_env=trust_env,
        cookie_jar=None if store_cookies else aiohttp.DummyCookieJar(),
    )


def get_ssrf_safe_requests_session(trust_env: bool = True, store_cookies: bool = True) -> requests.Session:
    """The requests counterpart of get_ssrf_safe_session, with the same proxy caveat."""
    session = requests.Session()
    session.trust_env = trust_env
    if not store_cookies:
        session.cookies.set_policy(http.cookiejar.DefaultCookiePolicy(allowed_domains=[]))
    session.mount('http://', _SSRFSafeAdapter())
    session.mount('https://', _SSRFSafeAdapter())
    return session


# accept-encoding goes because the client must advertise only codecs it can decode, the rest
# because the client derives them from the URL and body it is actually given. content-encoding
# stays: the browser's body is forwarded byte for byte, so its own labelling still applies.
_DROPPED_REQUEST_HEADERS = {'accept-encoding', 'connection', 'content-length', 'host', 'transfer-encoding'}

# The clients hand us a decoded body, so the sender's framing no longer describes it.
_DROPPED_RESPONSE_HEADERS = {'connection', 'content-encoding', 'content-length', 'transfer-encoding'}


def _forwardable_request_headers(headers: Dict[str, str]) -> Dict[str, str]:
    return {name: value for name, value in headers.items() if name.lower() not in _DROPPED_REQUEST_HEADERS}


def _fulfillable_response_headers(header_pairs: Iterable[Tuple[str, str]]) -> Dict[str, str]:
    """Collapse repeated headers the way route.fulfill expects: set-cookie by newline, rest by comma.

    Takes pairs rather than a mapping because reading either client's headers as a mapping loses
    duplicate Set-Cookie values, leaving one malformed cookie or one of the two.
    """
    collected: Dict[str, List[str]] = {}
    for name, value in header_pairs:
        name = name.lower()  # grouping by the sender's case would split a repeated header
        if name not in _DROPPED_RESPONSE_HEADERS:
            collected.setdefault(name, []).append(value)
    return {name: ('\n' if name == 'set-cookie' else ', ').join(values) for name, values in collected.items()}


def extract_metadata(soup, url):
    metadata = {'source': url}
    if title := soup.find('title'):
        metadata['title'] = title.get_text()
    if description := soup.find('meta', attrs={'name': 'description'}):
        metadata['description'] = description.get('content', 'No description found.')
    if html := soup.find('html'):
        metadata['language'] = html.get('lang', 'No language found.')
    return metadata


def verify_ssl_cert(url: str) -> bool:
    """Verify SSL certificate for the given URL."""
    if not url.startswith('https://'):
        return True

    try:
        hostname = url.split('://')[-1].split('/')[0]
        context = ssl.create_default_context(cafile=certifi.where())
        with context.wrap_socket(ssl.socket(), server_hostname=hostname) as s:
            s.connect((hostname, 443))
        return True
    except ssl.SSLError:
        return False
    except Exception as e:
        log.warning(f'SSL verification failed for {url}: {str(e)}')
        return False


class RateLimitMixin:
    async def _wait_for_rate_limit(self):
        """Wait to respect the rate limit if specified."""
        if self.requests_per_second and self.last_request_time:
            min_interval = timedelta(seconds=1.0 / self.requests_per_second)
            time_since_last = datetime.now() - self.last_request_time
            if time_since_last < min_interval:
                await asyncio.sleep((min_interval - time_since_last).total_seconds())
        self.last_request_time = datetime.now()

    def _sync_wait_for_rate_limit(self):
        """Synchronous version of rate limit wait."""
        if self.requests_per_second and self.last_request_time:
            min_interval = timedelta(seconds=1.0 / self.requests_per_second)
            time_since_last = datetime.now() - self.last_request_time
            if time_since_last < min_interval:
                time.sleep((min_interval - time_since_last).total_seconds())
        self.last_request_time = datetime.now()


class URLProcessingMixin:
    async def _verify_ssl_cert(self, url: str) -> bool:
        """Verify SSL certificate for a URL."""
        return await run_in_threadpool(verify_ssl_cert, url)

    async def _safe_process_url(self, url: str) -> bool:
        """Perform safety checks before processing a URL."""
        if self.verify_ssl and not await self._verify_ssl_cert(url):
            raise ValueError(f'SSL certificate verification failed for {url}')
        await self._wait_for_rate_limit()
        return True

    def _safe_process_url_sync(self, url: str) -> bool:
        """Synchronous version of safety checks."""
        if self.verify_ssl and not verify_ssl_cert(url):
            raise ValueError(f'SSL certificate verification failed for {url}')
        self._sync_wait_for_rate_limit()
        return True


class SafeFireCrawlLoader(BaseLoader, RateLimitMixin, URLProcessingMixin):
    def __init__(
        self,
        web_paths,
        verify_ssl: bool = True,
        trust_env: bool = False,
        requests_per_second: Optional[float] = None,
        continue_on_failure: bool = True,
        api_key: Optional[str] = None,
        api_url: Optional[str] = None,
        timeout: Optional[int] = None,
        mode: Literal['crawl', 'scrape', 'map'] = 'scrape',
        proxy: Optional[Dict[str, str]] = None,
        params: Optional[Dict] = None,
    ):
        proxy_server = proxy.get('server') if proxy else None
        if trust_env and not proxy_server:
            env_proxies = urllib.request.getproxies()
            env_proxy_server = env_proxies.get('https') or env_proxies.get('http')
            if env_proxy_server:
                if proxy:
                    proxy['server'] = env_proxy_server
                else:
                    proxy = {'server': env_proxy_server}
        self.web_paths = web_paths
        self.verify_ssl = verify_ssl
        self.requests_per_second = requests_per_second
        self.last_request_time = None
        self.trust_env = trust_env
        self.continue_on_failure = continue_on_failure
        self.api_key = api_key
        self.api_url = (api_url or 'https://api.firecrawl.dev').rstrip('/')
        self.timeout = timeout
        self.mode = mode
        self.params = params or {}

    def lazy_load(self) -> Iterator[Document]:
        for url in self.web_paths:
            try:
                self._sync_wait_for_rate_limit()
                doc = scrape_firecrawl_url(
                    self.api_url,
                    self.api_key,
                    url,
                    verify_ssl=self.verify_ssl,
                    timeout=self.timeout,
                    params=self.params,
                )
                if doc is not None:
                    yield doc
            except Exception as e:
                if self.continue_on_failure:
                    log.warning(f'Error extracting content from {url} with Firecrawl: {e}')
                    continue
                raise

    async def alazy_load(self):
        for url in self.web_paths:
            try:
                await self._wait_for_rate_limit()
                doc = await run_in_threadpool(
                    scrape_firecrawl_url,
                    self.api_url,
                    self.api_key,
                    url,
                    verify_ssl=self.verify_ssl,
                    timeout=self.timeout,
                    params=self.params,
                )
                if doc is not None:
                    yield doc
            except Exception as e:
                if self.continue_on_failure:
                    log.warning(f'Error extracting content from {url} with Firecrawl: {e}')
                    continue
                raise


class SafeTavilyLoader(BaseLoader, RateLimitMixin, URLProcessingMixin):
    def __init__(
        self,
        web_paths: Union[str, List[str]],
        api_key: str,
        extract_depth: Literal['basic', 'advanced'] = 'basic',
        continue_on_failure: bool = True,
        requests_per_second: Optional[float] = None,
        verify_ssl: bool = True,
        trust_env: bool = False,
        proxy: Optional[Dict[str, str]] = None,
    ):
        """Initialize SafeTavilyLoader with rate limiting and SSL verification support.

        Args:
            web_paths: List of URLs/paths to process.
            api_key: The Tavily API key.
            extract_depth: Depth of extraction ("basic" or "advanced").
            continue_on_failure: Whether to continue if extraction of a URL fails.
            requests_per_second: Number of requests per second to limit to.
            verify_ssl: If True, verify SSL certificates.
            trust_env: If True, use proxy settings from environment variables.
            proxy: Optional proxy configuration.
        """
        # Initialize proxy configuration if using environment variables
        proxy_server = proxy.get('server') if proxy else None
        if trust_env and not proxy_server:
            env_proxies = urllib.request.getproxies()
            env_proxy_server = env_proxies.get('https') or env_proxies.get('http')
            if env_proxy_server:
                if proxy:
                    proxy['server'] = env_proxy_server
                else:
                    proxy = {'server': env_proxy_server}

        # Store parameters for creating TavilyLoader instances
        self.web_paths = web_paths if isinstance(web_paths, list) else [web_paths]
        self.api_key = api_key
        self.extract_depth = extract_depth
        self.continue_on_failure = continue_on_failure
        self.verify_ssl = verify_ssl
        self.trust_env = trust_env
        self.proxy = proxy

        # Add rate limiting
        self.requests_per_second = requests_per_second
        self.last_request_time = None

    def lazy_load(self) -> Iterator[Document]:
        """Load documents with rate limiting support, delegating to TavilyLoader."""
        valid_urls = []
        for url in self.web_paths:
            try:
                self._safe_process_url_sync(url)
                valid_urls.append(url)
            except Exception as e:
                log.warning(f'SSL verification failed for {url}: {str(e)}')
                if not self.continue_on_failure:
                    raise e
        if not valid_urls:
            if self.continue_on_failure:
                log.warning('No valid URLs to process after SSL verification')
                return
            raise ValueError('No valid URLs to process after SSL verification')
        try:
            loader = TavilyLoader(
                urls=valid_urls,
                api_key=self.api_key,
                extract_depth=self.extract_depth,
                continue_on_failure=self.continue_on_failure,
            )
            yield from loader.lazy_load()
        except Exception as e:
            if self.continue_on_failure:
                log.exception(f'Error extracting content from URLs: {e}')
            else:
                raise e

    async def alazy_load(self) -> AsyncIterator[Document]:
        """Async version with rate limiting and SSL verification."""
        valid_urls = []
        for url in self.web_paths:
            try:
                await self._safe_process_url(url)
                valid_urls.append(url)
            except Exception as e:
                log.warning(f'SSL verification failed for {url}: {str(e)}')
                if not self.continue_on_failure:
                    raise e

        if not valid_urls:
            if self.continue_on_failure:
                log.warning('No valid URLs to process after SSL verification')
                return
            raise ValueError('No valid URLs to process after SSL verification')

        try:
            loader = TavilyLoader(
                urls=valid_urls,
                api_key=self.api_key,
                extract_depth=self.extract_depth,
                continue_on_failure=self.continue_on_failure,
            )
            async for document in loader.alazy_load():
                yield document
        except Exception as e:
            if self.continue_on_failure:
                log.exception(f'Error loading URLs: {e}')
            else:
                raise e


class SafeMicrosoftWebIQLoader(BaseLoader, RateLimitMixin, URLProcessingMixin):
    def __init__(
        self,
        web_paths: Union[str, List[str]],
        api_key: str,
        language: str = 'en',
        verify_ssl: bool = True,
        trust_env: bool = False,
        requests_per_second: Optional[float] = None,
        continue_on_failure: bool = True,
        timeout: Optional[int] = None,
    ):
        self.web_paths = web_paths if isinstance(web_paths, list) else [web_paths]
        self.api_key = api_key
        self.language = language
        self.verify_ssl = verify_ssl
        self.trust_env = trust_env
        self.requests_per_second = requests_per_second
        self.last_request_time = None
        self.continue_on_failure = continue_on_failure
        self.timeout = timeout

    def lazy_load(self) -> Iterator[Document]:
        valid_urls = []
        for url in self.web_paths:
            try:
                self._safe_process_url_sync(url)
                valid_urls.append(url)
            except Exception as e:
                log.warning(f'SSL verification failed for {url}: {str(e)}')
                if not self.continue_on_failure:
                    raise e
        if not valid_urls:
            if self.continue_on_failure:
                log.warning('No valid URLs to process after SSL verification')
                return
            raise ValueError('No valid URLs to process after SSL verification')

        loader = MicrosoftWebIQLoader(
            urls=valid_urls,
            api_base_url=self.api_base_url,
            api_key=self.api_key,
            language=self.language,
            verify_ssl=self.verify_ssl,
            timeout=self.timeout,
            continue_on_failure=self.continue_on_failure,
        )
        yield from loader.lazy_load()

    async def alazy_load(self) -> AsyncIterator[Document]:
        try:
            docs = await run_in_threadpool(lambda: list(self.lazy_load()))
            for doc in docs:
                yield doc
        except Exception as e:
            if self.continue_on_failure:
                log.warning(f'Error browsing URLs with Microsoft Web IQ: {e}')
            else:
                raise e


class SafePlaywrightURLLoader(PlaywrightURLLoader, RateLimitMixin, URLProcessingMixin):
    """Load HTML pages safely with Playwright, supporting SSL verification, rate limiting, and remote browser connection.

    Attributes:
        web_paths (List[str]): List of URLs to load.
        verify_ssl (bool): If True, verify SSL certificates.
        trust_env (bool): If True, use proxy settings from environment variables.
        requests_per_second (Optional[float]): Number of requests per second to limit to.
        continue_on_failure (bool): If True, continue loading other URLs on failure.
        headless (bool): If True, the browser will run in headless mode.
        proxy (dict): Proxy override settings for the Playwright session. Page requests are
            issued outside the browser, so they follow the environment proxy via trust_env
            rather than this setting.
        playwright_ws_url (Optional[str]): WebSocket endpoint URI for remote browser connection.
        playwright_timeout (Optional[int]): Maximum operation time in milliseconds.
    """

    def __init__(
        self,
        web_paths: List[str],
        verify_ssl: bool = True,
        trust_env: bool = False,
        requests_per_second: Optional[float] = None,
        continue_on_failure: bool = True,
        headless: bool = True,
        remove_selectors: Optional[List[str]] = None,
        proxy: Optional[Dict[str, str]] = None,
        playwright_ws_url: Optional[str] = None,
        playwright_timeout: Optional[int] = 10000,
    ):
        """Initialize with additional safety parameters and remote browser support."""

        proxy_server = proxy.get('server') if proxy else None
        if trust_env and not proxy_server:
            env_proxies = urllib.request.getproxies()
            env_proxy_server = env_proxies.get('https') or env_proxies.get('http')
            if env_proxy_server:
                if proxy:
                    proxy['server'] = env_proxy_server
                else:
                    proxy = {'server': env_proxy_server}

        # We'll set headless to False if using playwright_ws_url since it's handled by the remote browser
        super().__init__(
            urls=web_paths,
            continue_on_failure=continue_on_failure,
            headless=headless if playwright_ws_url is None else False,
            remove_selectors=remove_selectors,
            proxy=proxy,
        )
        self.verify_ssl = verify_ssl
        self.requests_per_second = requests_per_second
        self.last_request_time = None
        self.playwright_ws_url = playwright_ws_url
        self.trust_env = trust_env
        self.playwright_timeout = playwright_timeout

    def _request_timeout(self) -> float:
        # per-hop budget, since page.goto's timeout cannot reach into our own fetch and 0 disables
        # it. aiohttp treats it as a total where requests only caps each read, so sync runs looser.
        return (self.playwright_timeout or 30000) / 1000

    def _requests_verify(self) -> Union[bool, str]:
        """requests takes a CA path where aiohttp takes the parsed SSLContext.

        A bundle named directly in AIOHTTP_CLIENT_SESSION_SSL reaches us already parsed and
        cannot be expressed here, so that form falls back to the global bundle or certifi.
        """
        if not self.verify_ssl or AIOHTTP_CLIENT_SESSION_SSL is False:
            return False
        if AIOHTTP_CLIENT_SESSION_SSL is True:
            return True  # no usable global CA bundle, so both clients land on certifi
        return AIOHTTP_CLIENT_SSL_CERT_FILE or True

    def _intercept_navigation_sync(self, route, session):
        req = route.request

        hop_cookies: List[Tuple[str, str]] = []

        try:
            headers = _forwardable_request_headers(req.all_headers())
            post_data = req.post_data_buffer
            verify, timeout = self._requests_verify(), self._request_timeout()

            # The browser would resolve the hostname again, after the check; fetch it ourselves.
            def fetch(url):
                validate_url(url)
                return session.request(
                    req.method,
                    url,
                    headers=headers,
                    data=post_data,
                    allow_redirects=False,
                    verify=verify,
                    timeout=timeout,
                )

            resp = fetch(req.url)

            if 300 <= resp.status_code < 400:
                for _ in range(20):
                    if not AIOHTTP_CLIENT_ALLOW_REDIRECTS:
                        route.abort()
                        return

                    location = resp.headers.get('location')
                    if not location:
                        break

                    # only the last hop is fulfilled, so carry each hop's cookies to the browser
                    hop_cookies += [('set-cookie', v) for v in resp.raw.headers.getlist('set-cookie')]
                    resp = fetch(urllib.parse.urljoin(resp.url, location))
                    if not 300 <= resp.status_code < 400:
                        break
                else:
                    route.abort()
                    return
        except Exception as e:
            log.debug('Playwright loader could not fetch %s: %s', req.url, e)
            route.abort()
            return

        route.fulfill(
            status=resp.status_code,
            headers=_fulfillable_response_headers(hop_cookies + list(resp.raw.headers.items())),
            body=resp.content,
        )

    async def _intercept_navigation(self, route, session):
        req = route.request

        hop_cookies: List[Tuple[str, str]] = []

        try:
            headers = _forwardable_request_headers(await req.all_headers())
            post_data = req.post_data_buffer

            # The browser would resolve the hostname again, after the check; fetch it ourselves.
            async def fetch(url):
                await run_in_threadpool(validate_url, url)
                response = await session.request(
                    req.method,
                    url,
                    headers=headers,
                    data=post_data,
                    allow_redirects=False,
                    ssl=AIOHTTP_CLIENT_SESSION_SSL if self.verify_ssl else False,
                    timeout=aiohttp.ClientTimeout(total=self._request_timeout()),
                )
                # aiohttp only returns the connection to the pool once the body is buffered
                return response, await response.read()

            resp, body = await fetch(req.url)

            if 300 <= resp.status < 400:
                for _ in range(20):
                    if not AIOHTTP_CLIENT_ALLOW_REDIRECTS:
                        await route.abort()
                        return

                    location = resp.headers.get('location')
                    if not location:
                        break

                    # only the last hop is fulfilled, so carry each hop's cookies to the browser
                    hop_cookies += [('set-cookie', v) for v in resp.headers.getall('Set-Cookie', [])]
                    resp, body = await fetch(urllib.parse.urljoin(str(resp.url), location))
                    if not 300 <= resp.status < 400:
                        break
                else:
                    await route.abort()
                    return
        except Exception as e:
            log.debug('Playwright loader could not fetch %s: %s', req.url, e)
            await route.abort()
            return

        await route.fulfill(
            status=resp.status,
            headers=_fulfillable_response_headers(hop_cookies + list(resp.headers.items())),
            body=body,
        )

    def lazy_load(self) -> Iterator[Document]:
        """Safely load URLs synchronously with support for remote browser."""
        from playwright.sync_api import sync_playwright

        with sync_playwright() as p:
            # Use remote browser if ws_endpoint is provided, otherwise use local browser
            if self.playwright_ws_url:
                browser = p.chromium.connect(self.playwright_ws_url)
            else:
                browser = p.chromium.launch(headless=self.headless, proxy=self.proxy)

            with browser:
                for url in self.urls:
                    try:
                        self._safe_process_url_sync(url)
                        # opened before the page so it outlives any route still in flight at teardown
                        with (
                            get_ssrf_safe_requests_session(self.trust_env, store_cookies=False) as session,
                            browser.new_page(service_workers='block') as page,
                        ):
                            page.route('**/*', lambda route: self._intercept_navigation_sync(route, session))
                            page.route_web_socket('**/*', lambda ws_route: ws_route.close())
                            response = page.goto(url, timeout=self.playwright_timeout)
                            if response is None:
                                raise ValueError(f'page.goto() returned None for url {url}')

                            text = self.evaluator.evaluate(page, browser, response)
                            metadata = {'source': url}
                            yield Document(page_content=text, metadata=metadata)
                    except Exception as e:
                        if self.continue_on_failure:
                            log.exception(f'Error loading {url}: {e}')
                            continue
                        raise e

    async def alazy_load(self) -> AsyncIterator[Document]:
        """Safely load URLs asynchronously with support for remote browser."""
        from playwright.async_api import async_playwright

        async with async_playwright() as p:
            # Use remote browser if ws_endpoint is provided, otherwise use local browser
            if self.playwright_ws_url:
                browser = await p.chromium.connect(self.playwright_ws_url)
            else:
                browser = await p.chromium.launch(headless=self.headless, proxy=self.proxy)

            async with browser:
                for url in self.urls:
                    try:
                        await self._safe_process_url(url)
                        # opened before the page so it outlives any route still in flight at teardown
                        async with (
                            get_ssrf_safe_session(self.trust_env, store_cookies=False) as session,
                            await browser.new_page(service_workers='block') as page,
                        ):
                            await page.route('**/*', lambda route: self._intercept_navigation(route, session))
                            await page.route_web_socket('**/*', lambda ws_route: ws_route.close())
                            response = await page.goto(url, timeout=self.playwright_timeout)
                            if response is None:
                                raise ValueError(f'page.goto() returned None for url {url}')

                            text = await self.evaluator.evaluate_async(page, browser, response)
                            metadata = {'source': url}
                            yield Document(page_content=text, metadata=metadata)
                    except Exception as e:
                        if self.continue_on_failure:
                            log.exception(f'Error loading {url}: {e}')
                            continue
                        raise e


class SafeWebBaseLoader(WebBaseLoader):
    """WebBaseLoader with enhanced error handling for URLs."""

    def __init__(self, trust_env: bool = False, *args, **kwargs):
        """Initialize SafeWebBaseLoader
        Args:
            trust_env (bool, optional): set to True if using proxy to make web requests, for example
                using http(s)_proxy environment variables. Defaults to False.
        """
        # lxml parses scraped pages far faster than the html.parser default
        kwargs.setdefault('default_parser', 'lxml')
        super().__init__(*args, **kwargs)
        self.trust_env = trust_env

        # Propagate USER_AGENT env var so that both the sync _scrape() and
        # async _fetch() paths present a real UA instead of python-requests/2.x
        # which gets blocked by Cloudflare, Wikipedia, and similar bot-detection.
        # _fetch() forwards self.session.headers to the aiohttp session, so
        # setting it here covers both code-paths.
        if USER_AGENT:
            self.session.headers['User-Agent'] = USER_AGENT

        # Prevent redirect-based SSRF on the synchronous _scrape() path.
        # validate_url() is called once on the originally-submitted URL, but the
        # parent WebBaseLoader's _scrape() invokes self.session.get(url, **self.requests_kwargs)
        # which by default follows redirects. Without the override below, an attacker
        # can submit a public URL that 302-redirects to an internal address (RFC1918,
        # 127.0.0.1, 169.254.169.254, etc.) and the redirected target is fetched without
        # re-validation. Matches the policy enforced on the async _fetch() path below.
        self.requests_kwargs = {
            **(self.requests_kwargs or {}),
            'allow_redirects': AIOHTTP_CLIENT_ALLOW_REDIRECTS,
        }

        self.session.mount('http://', _SSRFSafeAdapter())
        self.session.mount('https://', _SSRFSafeAdapter())

    async def _fetch(self, url: str, retries: int = 3, cooldown: int = 2, backoff: float = 1.5) -> str:
        connector = _SSRFSafeConnector()
        async with aiohttp.ClientSession(trust_env=self.trust_env, connector=connector) as session:
            for i in range(retries):
                try:
                    kwargs: Dict = dict(
                        headers=self.session.headers,
                        cookies=self.session.cookies.get_dict(),
                    )
                    if not self.session.verify:
                        kwargs['ssl'] = False
                    else:
                        kwargs['ssl'] = AIOHTTP_CLIENT_SESSION_SSL

                    async with session.get(
                        url,
                        **(self.requests_kwargs | kwargs),
                    ) as response:
                        if self.raise_for_status:
                            response.raise_for_status()
                        return await response.text()
                except aiohttp.ClientConnectionError as e:
                    if i == retries - 1:
                        raise
                    else:
                        log.warning(f'Error fetching {url} with attempt {i + 1}/{retries}: {e}. Retrying...')
                        await asyncio.sleep(cooldown * backoff**i)
        raise ValueError('retry count exceeded')

    def _unpack_fetch_results(self, results: Any, urls: List[str], parser: Union[str, None] = None) -> List[Any]:
        """Unpack fetch results into BeautifulSoup objects."""
        from bs4 import BeautifulSoup

        final_results = []
        for i, result in enumerate(results):
            url = urls[i]
            url_parser = parser
            if url_parser is None:
                url_parser = 'xml' if url.endswith('.xml') else self.default_parser
                self._check_parser(url_parser)
            final_results.append(BeautifulSoup(result, url_parser, **self.bs_kwargs))
        return final_results

    def lazy_load(self) -> Iterator[Document]:
        """Lazy load text from the url(s) in web_path with error handling."""
        for path in self.web_paths:
            try:
                soup = self._scrape(path, bs_kwargs=self.bs_kwargs)
                text = soup.get_text(**self.bs_get_text_kwargs)

                # Build metadata
                metadata = extract_metadata(soup, path)

                yield Document(page_content=text, metadata=metadata)
            except Exception as e:
                # Log the error and continue with the next URL
                log.exception(f'Error loading {path}: {e}')

    def _document_from_html(self, html: str, url: str) -> Document:
        """Build one Document."""
        soup = self._unpack_fetch_results([html], [url])[0]
        return Document(
            page_content=soup.get_text(**self.bs_get_text_kwargs),
            metadata=extract_metadata(soup, url),
        )

    async def alazy_load(self) -> AsyncIterator[Document]:
        """Async lazy load text from the url(s) in web_path."""
        results = await self.fetch_all(self.web_paths)
        for path, html in zip(self.web_paths, results):
            # parsing a large page costs hundreds of ms, keep it off the event loop
            yield await asyncio.to_thread(self._document_from_html, html, path)

    async def aload(self) -> list[Document]:
        """Load data into Document objects."""
        return [document async for document in self.alazy_load()]


def get_web_loader(
    urls: Union[str, Sequence[str]],
    verify_ssl: bool = True,
    requests_per_second: int = 2,
    trust_env: bool = False,
    loader_config: Optional[dict] = None,
):
    # Check if the URLs are valid
    safe_urls = safe_validate_urls([urls] if isinstance(urls, str) else urls)

    if not safe_urls:
        log.warning(f'All provided URLs were blocked or invalid: {urls}')
        raise ValueError(ERROR_MESSAGES.INVALID_URL)

    loader_config = loader_config or {}

    def cfg(key, env_value):
        # Admin-saved DB value wins; env constant covers keys never saved.
        value = loader_config.get(key)
        return env_value if value is None else value

    engine = cfg('web_loader_engine', WEB_LOADER_ENGINE)
    web_loader_timeout = cfg('web_loader_timeout', WEB_LOADER_TIMEOUT)

    web_loader_args = {
        'web_paths': safe_urls,
        'verify_ssl': verify_ssl,
        'requests_per_second': requests_per_second,
        'continue_on_failure': True,
        'trust_env': trust_env,
    }

    WebLoaderClass = None

    if engine == '' or engine == 'safe_web':
        WebLoaderClass = SafeWebBaseLoader

        request_kwargs = {}
        if web_loader_timeout:
            try:
                timeout_value = float(web_loader_timeout)
            except ValueError:
                timeout_value = None

            if timeout_value:
                request_kwargs['timeout'] = timeout_value

        if request_kwargs:
            web_loader_args['requests_kwargs'] = request_kwargs

    if engine == 'playwright':
        WebLoaderClass = SafePlaywrightURLLoader
        web_loader_args['playwright_timeout'] = cfg('playwright_timeout', PLAYWRIGHT_TIMEOUT)
        playwright_ws_url = cfg('playwright_ws_url', PLAYWRIGHT_WS_URL)
        if playwright_ws_url:
            web_loader_args['playwright_ws_url'] = playwright_ws_url

    if engine == 'firecrawl':
        WebLoaderClass = SafeFireCrawlLoader
        web_loader_args['api_key'] = cfg('firecrawl_api_key', FIRECRAWL_API_KEY)
        web_loader_args['api_url'] = cfg('firecrawl_api_url', FIRECRAWL_API_BASE_URL)
        firecrawl_timeout = cfg('firecrawl_timeout', FIRECRAWL_TIMEOUT)
        if firecrawl_timeout:
            try:
                web_loader_args['timeout'] = int(firecrawl_timeout)
            except ValueError:
                pass

    if engine == 'tavily':
        WebLoaderClass = SafeTavilyLoader
        web_loader_args['api_key'] = cfg('tavily_api_key', TAVILY_API_KEY)
        web_loader_args['extract_depth'] = cfg('tavily_extract_depth', TAVILY_EXTRACT_DEPTH)

    if engine == 'microsoft_web_iq':
        WebLoaderClass = SafeMicrosoftWebIQLoader
        web_loader_args['api_base_url'] = cfg('microsoft_web_iq_api_base_url', MICROSOFT_WEB_IQ_API_BASE_URL)
        web_loader_args['api_key'] = cfg('microsoft_web_iq_api_key', MICROSOFT_WEB_IQ_API_KEY)
        web_loader_args['language'] = cfg('microsoft_web_iq_language', MICROSOFT_WEB_IQ_LANGUAGE)
        if web_loader_timeout:
            try:
                web_loader_args['timeout'] = int(web_loader_timeout)
            except ValueError:
                pass

    if engine == 'external':
        WebLoaderClass = ExternalWebLoader
        web_loader_args['external_url'] = cfg('external_web_loader_url', EXTERNAL_WEB_LOADER_URL)
        web_loader_args['external_api_key'] = cfg('external_web_loader_api_key', EXTERNAL_WEB_LOADER_API_KEY)

    if WebLoaderClass:
        web_loader = WebLoaderClass(**web_loader_args)

        log.debug(
            'Using WEB_LOADER_ENGINE %s for %s URLs',
            web_loader.__class__.__name__,
            len(safe_urls),
        )

        return web_loader
    else:
        raise ValueError(
            f'Invalid WEB_LOADER_ENGINE: {engine}. '
            "Please set it to 'safe_web', 'playwright', 'firecrawl', 'tavily', 'external', or 'microsoft_web_iq'."
        )