Match WEB_FETCH_FILTER_LIST on hostnames with label boundaries, not URL suffix (CWE-693) (#25949)
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:
co-authored by
addcontent
Claude Opus 4.8
parent
4f69c33de0
commit
087878ce84
@@ -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
|
||||
|
||||
|
||||
@@ -55,7 +55,7 @@ from open_webui.env import (
|
||||
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__)
|
||||
|
||||
@@ -93,7 +93,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)
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user