Match WEB_FETCH_FILTER_LIST on hostnames with label boundaries, not URL suffix (CWE-693)

is_string_allowed does endswith() matching and was called with the full URL
(retrieval/web/utils.py) against WEB_FETCH_FILTER_LIST, so a blocklisted host with any
path (https://blocked.example/x) ended with /x, not the host, and slipped through; the
allowlist direction false-rejected legitimate URLs and admitted attacker URLs ending in
an allowed string. The same endswith caused label confusion at the hostname call site
(retrieval/web/main.py): corp.com matched evilcorp.com, 10.0.0.1 matched 110.0.0.1.

Add is_host_allowed(host, ...) matching on DNS label boundaries (host == pattern or
host.endswith('.' + pattern)), called with the parsed hostname at both web-fetch call
sites. is_string_allowed is left unchanged for the unrelated function-name filters
(utils/middleware.py, utils/tools.py).

The separate is_global guard (validate_url / _ssrf_safe_new_conn, active when
ENABLE_RAG_LOCAL_WEB_FETCH is off) already blocks RFC1918/loopback/link-local, so this
restores the admin's intended blocking of specific public hosts.

Co-authored-by: addcontent <59762500+addcontent@users.noreply.github.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Classic298 2026-06-11 14:55:35 +02:00
parent b1d40f3409
commit 834a2cf109
3 changed files with 44 additions and 4 deletions

View file

@ -4,7 +4,7 @@ from urllib.parse import urlparse
import validators
from open_webui.retrieval.web.utils import resolve_hostname
from open_webui.utils.misc import is_string_allowed
from open_webui.utils.misc import is_host_allowed
from pydantic import BaseModel
@ -32,7 +32,7 @@ def get_filtered_results(results, filter_list):
except Exception:
pass
if is_string_allowed(hostnames, filter_list):
if is_host_allowed(hostnames, filter_list):
filtered_results.append(result)
continue

View file

@ -50,7 +50,7 @@ from open_webui.env import AIOHTTP_CLIENT_ALLOW_REDIRECTS, AIOHTTP_CLIENT_SESSIO
from open_webui.retrieval.loaders.external_web import ExternalWebLoader
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_string_allowed
from open_webui.utils.misc import is_host_allowed
log = logging.getLogger(__name__)
@ -88,7 +88,9 @@ def validate_url(url: Union[str, Sequence[str]]):
# Blocklist check using unified filtering logic
if WEB_FETCH_FILTER_LIST:
if not is_string_allowed(url, WEB_FETCH_FILTER_LIST):
# 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.
if not is_host_allowed(parsed_url.hostname, WEB_FETCH_FILTER_LIST):
log.warning(f'URL blocked by filter list: {url}')
raise ValueError(ERROR_MESSAGES.INVALID_URL)

View file

@ -69,6 +69,44 @@ def is_string_allowed(string: Union[str, Sequence[str]], filter_list: list[str |
return True
def _host_matches_pattern(host: str, pattern: str) -> bool:
"""Match a hostname against a filter entry on DNS label boundaries.
`pattern` matches `host` when equal or a parent domain of it, so `corp.com`
matches `api.corp.com` but not `evilcorp.com`, and an IP literal matches only
itself. Avoids the raw-suffix confusion of a plain endswith.
"""
host = (host or '').strip().lower().rstrip('.')
pattern = (pattern or '').strip().lower().rstrip('.')
if not host or not pattern:
return False
return host == pattern or host.endswith('.' + pattern)
def is_host_allowed(host: Union[str, Sequence[str]], filter_list: list[str | None] = None) -> bool:
"""Allow/block a hostname (or list of hostnames / resolved IPs) against a
WEB_FETCH_FILTER_LIST-style filter, matching on label boundaries.
Pass a parsed hostname, never a full URL: matching against a URL lets a path
component defeat the filter (e.g. ``https://blocked.example/x`` ends with ``/x``,
not the blocked host). Entries prefixed with ``!`` are blocked; the rest form an allowlist.
"""
if not filter_list:
return True
allow_list, block_list = get_allow_block_lists(filter_list)
hosts = [host] if isinstance(host, str) else list(host or [])
if allow_list:
if not any(_host_matches_pattern(h, allowed) for h in hosts for allowed in allow_list):
return False
if any(_host_matches_pattern(h, blocked) for h in hosts for blocked in block_list):
return False
return True
def get_message_list(messages_map, message_id):
"""
Reconstructs a list of messages in order up to the specified message_id.