fix: allow for allowlisted redirect URIs (#27761)
* fix: allow for allowlisted redirect URIs * github comment addressing * Update litellm/proxy/_experimental/mcp_server/oauth_utils.py Co-authored-by: veria-ai[bot] <224490171+veria-ai[bot]@users.noreply.github.com> * harden oauth wildcard further * test: cover wildcard entry with dot-leading suffix rejection --------- Co-authored-by: veria-ai[bot] <224490171+veria-ai[bot]@users.noreply.github.com>
This commit is contained in:
Dennis Henry
GitHub
parent
7a462a4220
commit
9b6ab55c5f
@ -86,7 +86,9 @@ def decode_state_hash(encrypted_state: str) -> dict:
|
||||
def _get_validated_client_redirect_uri(
|
||||
request: Request, state_data: Dict[str, Any]
|
||||
) -> str:
|
||||
"""Return a trusted (same-origin or loopback) client redirect URI from OAuth state."""
|
||||
"""Return a trusted (same-origin, loopback, or ops-allowlisted)
|
||||
client redirect URI from OAuth state.
|
||||
"""
|
||||
redirect_uri = state_data.get("client_redirect_uri") or state_data.get("base_url")
|
||||
if not redirect_uri or not isinstance(redirect_uri, str):
|
||||
raise HTTPException(status_code=400, detail="Invalid redirect URI")
|
||||
@ -296,11 +298,10 @@ async def authorize_with_server(
|
||||
status_code=400, detail="MCP server authorization url is not set"
|
||||
)
|
||||
|
||||
# Loopback OR same-origin redirect_uri. The URI is encrypted into the
|
||||
# OAuth state and decoded on /callback to redirect the user back;
|
||||
# restricting to trusted origins blocks the open-redirect +
|
||||
# code-theft primitive (VERIA-57 root cause B). Loopback supports
|
||||
# native MCP clients; same-origin supports the proxy's own UI callback.
|
||||
# Trusted redirect_uri: same-origin, loopback, or ops-allowlisted.
|
||||
# The URI is encrypted into the OAuth state and decoded on
|
||||
# /callback to redirect the user back; a non-trusted URI would be
|
||||
# an open-redirect + code-theft primitive (VERIA-57 root cause B).
|
||||
validate_trusted_redirect_uri(request, redirect_uri)
|
||||
parsed = urlparse(redirect_uri)
|
||||
base_url = urlunparse(parsed._replace(query=""))
|
||||
@ -623,12 +624,12 @@ async def callback(request: Request, code: str, state: str):
|
||||
state_data = decode_state_hash(state)
|
||||
original_state = state_data["original_state"]
|
||||
|
||||
# Re-validate at the sink. /authorize rejects untrusted
|
||||
# redirect_uri before encoding into state, but encrypted states
|
||||
# minted before that check was added have no expiry and remain
|
||||
# valid indefinitely. Validating here (same-origin OR loopback)
|
||||
# blocks the open-redirect + code-theft primitive even for pre-fix
|
||||
# states while allowing the UI's same-origin callback to work.
|
||||
# Re-validate the client redirect URI at the sink. /authorize
|
||||
# rejects untrusted URIs before encoding them into state, but
|
||||
# encrypted states minted before that check was added have no
|
||||
# expiry and remain valid indefinitely. Validating here blocks
|
||||
# the open-redirect + code-theft primitive even for pre-fix
|
||||
# states while permitting same-origin / allowlisted clients.
|
||||
redirect_uri = _get_validated_client_redirect_uri(request, state_data)
|
||||
|
||||
params = {"code": code, "state": original_state}
|
||||
|
||||
@ -1,7 +1,9 @@
|
||||
"""Shared helpers for the MCP OAuth authorization endpoints
|
||||
(BYOK + discoverable / pass-through OAuth proxy)."""
|
||||
|
||||
import os
|
||||
from ipaddress import ip_address
|
||||
from typing import List, Optional
|
||||
from urllib.parse import urlparse, urlunparse
|
||||
|
||||
from fastapi import HTTPException, Request
|
||||
@ -13,6 +15,20 @@ from litellm.proxy.auth.ip_address_utils import IPAddressUtils
|
||||
# must not be cached — both success and error bodies may reveal secrets.
|
||||
TOKEN_NO_CACHE_HEADERS = {"Cache-Control": "no-store", "Pragma": "no-cache"}
|
||||
|
||||
# Stripped from netloc before same-origin comparison so
|
||||
# ``llm.example.com`` matches ``llm.example.com:443`` (load balancers
|
||||
# routinely set X-Forwarded-Port: 443 even when the client URL has no
|
||||
# explicit port, which would otherwise break a literal netloc compare).
|
||||
_DEFAULT_PORTS = {"http": 80, "https": 443}
|
||||
|
||||
# Env var for ops to allowlist additional redirect_uri origins beyond
|
||||
# same-origin + loopback — needed for first-party OAuth clients hosted
|
||||
# on sister domains (e.g. a web app on app.example.com registering as
|
||||
# an OAuth client of the MCP proxy on llm.example.com). Comma-separated;
|
||||
# each entry is ``host`` or ``host:port``; a ``*.`` prefix matches any
|
||||
# subdomain. HTTPS only.
|
||||
_TRUSTED_REDIRECT_ORIGINS_ENV = "MCP_TRUSTED_REDIRECT_ORIGINS"
|
||||
|
||||
|
||||
def get_request_base_url(request: Request) -> str:
|
||||
"""
|
||||
@ -96,22 +112,106 @@ def validate_loopback_redirect_uri(redirect_uri: str) -> None:
|
||||
raise HTTPException(status_code=400, detail="invalid_request")
|
||||
|
||||
|
||||
def _strip_default_port(scheme: str, netloc: str) -> str:
|
||||
"""Return ``netloc`` lowercased with the scheme's default port
|
||||
stripped. ``Llm.Example.com:443`` with scheme ``https`` becomes
|
||||
``llm.example.com``. Used so a literal netloc comparison between
|
||||
the proxy's origin and the client redirect_uri survives a load-
|
||||
balancer that sets ``X-Forwarded-Port: 443``.
|
||||
"""
|
||||
if not netloc:
|
||||
return netloc
|
||||
lowered = netloc.lower()
|
||||
if lowered.startswith("["):
|
||||
# IPv6 literal: port (if any) appears after the "]".
|
||||
close = lowered.rfind("]")
|
||||
if close != -1 and lowered[close + 1 :].startswith(":"):
|
||||
try:
|
||||
port = int(lowered[close + 2 :])
|
||||
except ValueError:
|
||||
return lowered
|
||||
if _DEFAULT_PORTS.get(scheme) == port:
|
||||
return lowered[: close + 1]
|
||||
return lowered
|
||||
if ":" in lowered:
|
||||
host, _, port_str = lowered.rpartition(":")
|
||||
try:
|
||||
port = int(port_str)
|
||||
except ValueError:
|
||||
return lowered
|
||||
if _DEFAULT_PORTS.get(scheme) == port:
|
||||
return host
|
||||
return lowered
|
||||
|
||||
|
||||
def _parse_trusted_redirect_origins() -> List[str]:
|
||||
"""Parse ``MCP_TRUSTED_REDIRECT_ORIGINS`` into normalized entries.
|
||||
Empty / unset env var → empty list. Entries are lowercased and any
|
||||
scheme / path component the operator included is stripped. Default
|
||||
``:443`` is also stripped from non-wildcard entries so
|
||||
``app.example.com:443`` matches a redirect_netloc whose own ``:443``
|
||||
has already been normalized away — the allowlist path is https-only,
|
||||
so ``:443`` is the only default port that can legitimately appear.
|
||||
"""
|
||||
raw = os.environ.get(_TRUSTED_REDIRECT_ORIGINS_ENV, "").strip()
|
||||
if not raw:
|
||||
return []
|
||||
entries: List[str] = []
|
||||
for token in raw.split(","):
|
||||
entry = token.strip().lower()
|
||||
if not entry:
|
||||
continue
|
||||
if "://" in entry:
|
||||
entry = entry.split("://", 1)[1]
|
||||
entry = entry.split("/", 1)[0]
|
||||
if not entry:
|
||||
continue
|
||||
# Wildcards don't express port constraints; leave them alone.
|
||||
if not entry.startswith("*."):
|
||||
entry = _strip_default_port("https", entry)
|
||||
if entry:
|
||||
entries.append(entry)
|
||||
return entries
|
||||
|
||||
|
||||
def _matches_trusted_origin_entry(netloc: str, entry: str) -> bool:
|
||||
"""``entry`` is either ``host[:port]`` (exact match after port
|
||||
normalization) or ``*.suffix`` (subdomain wildcard; matches any
|
||||
strictly-deeper subdomain of ``suffix`` but not ``suffix`` itself).
|
||||
``netloc`` is the already-port-normalized, lowercased netloc of
|
||||
the redirect_uri being validated.
|
||||
"""
|
||||
if entry.startswith("*."):
|
||||
suffix = entry[2:]
|
||||
if not suffix or suffix.startswith("."):
|
||||
return False
|
||||
# Strip port from netloc for wildcard host comparison;
|
||||
# wildcards don't express port constraints.
|
||||
host = netloc.split(":", 1)[0] if ":" in netloc else netloc
|
||||
return host != suffix and host.endswith("." + suffix)
|
||||
return netloc == entry
|
||||
|
||||
|
||||
def validate_trusted_redirect_uri(request: Request, redirect_uri: str) -> None:
|
||||
"""Accept same-origin (proxy's own origin) OR loopback ``redirect_uri``.
|
||||
"""Accept ``redirect_uri`` when it is (a) same-origin with the
|
||||
proxy's own request origin, (b) loopback, or (c) listed in the
|
||||
``MCP_TRUSTED_REDIRECT_ORIGINS`` ops allowlist.
|
||||
|
||||
Same-origin is required for the LiteLLM UI's OAuth flow: the UI
|
||||
redirects to ``<proxy>/ui/mcp/oauth/callback`` which is not loopback
|
||||
but is on the proxy's own trusted HTTPS origin. An attacker cannot
|
||||
host content on the proxy's own origin without already owning the
|
||||
proxy, so the open-redirect / code-theft primitive that motivated
|
||||
:func:`validate_loopback_redirect_uri` does not apply here.
|
||||
Same-origin is VERIA-57's threat-model-safe equivalent of loopback:
|
||||
an attacker who can host content on the proxy's own HTTPS origin
|
||||
has already compromised the proxy, so the open-redirect + code-
|
||||
theft primitive that motivated the loopback-only rule does not
|
||||
apply. The same reasoning extends to ops-trusted first-party
|
||||
hosts (e.g. an internal web app registering as an OAuth client of
|
||||
the proxy on a sister domain).
|
||||
|
||||
Loopback continues to be accepted for native MCP clients (per
|
||||
OAuth 2.1 §4.1.2.1 + RFC 8252 §7.3).
|
||||
Allowlisted non-loopback hosts are accepted only when the
|
||||
redirect_uri scheme is ``https`` — an attacker on the network
|
||||
cannot elevate to https without controlling the host's TLS key.
|
||||
|
||||
Use this in the discoverable OAuth proxy endpoints that serve both
|
||||
native clients and the proxy's own UI. BYOK endpoints that only
|
||||
support native clients should keep
|
||||
native clients and the proxy's UI / cross-origin web clients. The
|
||||
BYOK endpoints, which only serve native MCP clients, retain
|
||||
:func:`validate_loopback_redirect_uri`.
|
||||
"""
|
||||
try:
|
||||
@ -122,26 +222,53 @@ def validate_trusted_redirect_uri(request: Request, redirect_uri: str) -> None:
|
||||
raise HTTPException(status_code=400, detail="invalid_request")
|
||||
if parsed.fragment:
|
||||
raise HTTPException(status_code=400, detail="invalid_request")
|
||||
if not parsed.netloc or parsed.username is not None or parsed.password is not None:
|
||||
raise HTTPException(status_code=400, detail="invalid_request")
|
||||
# Reject userinfo (``user:pass@host``) outright: OAuth redirect_uris
|
||||
# have no legitimate reason to carry credentials, and allowing them
|
||||
# opens a host-confusion attack where the netloc *looks* allowlisted
|
||||
# (``app.example.com:443@attacker.example``) but the browser navigates
|
||||
# to the post-``@`` host and hands the authorization code to the
|
||||
# attacker. We compare against ``hostname`` after this, but defense in
|
||||
# depth keeps malformed netloc strings from reaching the wildcard
|
||||
# splitter.
|
||||
if parsed.username is not None or parsed.password is not None:
|
||||
raise HTTPException(status_code=400, detail="invalid_request")
|
||||
# Reject backslash in netloc: urlparse keeps ``\`` as part of netloc,
|
||||
# but browsers normalize ``\`` to ``/`` for http(s) URLs and treat it
|
||||
# as the start of the path. An attacker can exploit that split by
|
||||
# crafting ``https://attacker.net\app.example.com/cb`` — urlparse sees
|
||||
# ``attacker.net\app.example.com`` (matches ``*.example.com``) while
|
||||
# the browser navigates to ``attacker.net`` with the auth code.
|
||||
if "\\" in parsed.netloc:
|
||||
raise HTTPException(status_code=400, detail="invalid_request")
|
||||
|
||||
# Same-origin: scheme + netloc (host[:port]) must match the proxy's
|
||||
# own base URL at this request (honouring trusted X-Forwarded-*).
|
||||
redirect_netloc = _strip_default_port(parsed.scheme, parsed.netloc)
|
||||
|
||||
# (a) Same-origin. Swallow ``get_request_base_url`` failures so the
|
||||
# loopback + allowlist paths remain reachable when the origin can't
|
||||
# be determined (e.g. request came from an untrusted proxy and
|
||||
# ``get_request_base_url`` raised).
|
||||
proxy_base: Optional[str] = None
|
||||
try:
|
||||
proxy_base = urlparse(get_request_base_url(request))
|
||||
if (
|
||||
parsed.netloc
|
||||
and parsed.scheme == proxy_base.scheme
|
||||
and parsed.netloc.lower() == proxy_base.netloc.lower()
|
||||
):
|
||||
return
|
||||
proxy_base = get_request_base_url(request)
|
||||
except Exception as exc:
|
||||
# If we can't determine the proxy's origin, fall through to
|
||||
# loopback. Log so the failure is diagnosable in production.
|
||||
verbose_logger.warning(
|
||||
"validate_trusted_redirect_uri: could not determine proxy origin, "
|
||||
"falling back to loopback-only check. error=%s",
|
||||
"falling back to loopback + allowlist. error=%s",
|
||||
exc,
|
||||
)
|
||||
proxy_base = None
|
||||
if proxy_base:
|
||||
proxy_parsed = urlparse(proxy_base)
|
||||
if (
|
||||
parsed.scheme == proxy_parsed.scheme
|
||||
and redirect_netloc
|
||||
== _strip_default_port(proxy_parsed.scheme, proxy_parsed.netloc)
|
||||
):
|
||||
return
|
||||
|
||||
# (b) Loopback — same rule as validate_loopback_redirect_uri.
|
||||
host = (parsed.hostname or "").lower()
|
||||
if host == "localhost":
|
||||
return
|
||||
@ -150,4 +277,11 @@ def validate_trusted_redirect_uri(request: Request, redirect_uri: str) -> None:
|
||||
return
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# (c) Ops allowlist. https only.
|
||||
if parsed.scheme == "https":
|
||||
for entry in _parse_trusted_redirect_origins():
|
||||
if _matches_trusted_origin_entry(redirect_netloc, entry):
|
||||
return
|
||||
|
||||
raise HTTPException(status_code=400, detail="invalid_request")
|
||||
|
||||
@ -1137,67 +1137,445 @@ def test_validate_loopback_redirect_uri_rejects_malformed_cleanly():
|
||||
assert exc.value.status_code == 400
|
||||
|
||||
|
||||
def _mock_request_with_base_url(base_url: str):
|
||||
req = MagicMock()
|
||||
req.base_url = base_url
|
||||
req.headers = {}
|
||||
return req
|
||||
# ---------------------------------------------------------------------------
|
||||
# validate_trusted_redirect_uri — same-origin + loopback + env allowlist
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_trusted_request(base_url: str = "https://llm.example.com/"):
|
||||
"""Build a request-like object whose same-origin is ``base_url``.
|
||||
|
||||
``get_request_base_url`` defers to ``request.base_url`` unless the
|
||||
caller is a trusted proxy, so passing the target origin as
|
||||
``base_url`` is sufficient here — no X-Forwarded headers needed.
|
||||
"""
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
mock = MagicMock()
|
||||
mock.base_url = base_url
|
||||
mock.headers = {}
|
||||
return mock
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_accepts_same_origin():
|
||||
"""UI OAuth flow: redirect_uri on the proxy's own origin is allowed."""
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
validate_trusted_redirect_uri,
|
||||
)
|
||||
|
||||
req = _mock_request_with_base_url("https://proxy.example.com/")
|
||||
# Should not raise.
|
||||
validate_trusted_redirect_uri(
|
||||
req, "https://proxy.example.com/ui/mcp/oauth/callback"
|
||||
req = _make_trusted_request("https://llm.example.com/")
|
||||
validate_trusted_redirect_uri(req, "https://llm.example.com/ui/mcp/callback")
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_same_origin_normalizes_default_port():
|
||||
"""Regression: a load balancer that sets X-Forwarded-Port: 443 would
|
||||
otherwise produce a proxy_base of ``https://llm.example.com:443``
|
||||
which wouldn't literally match the browser's port-less ``llm.example.com``
|
||||
redirect_uri even though both represent the same origin."""
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
validate_trusted_redirect_uri,
|
||||
)
|
||||
|
||||
# Proxy base with explicit :443 — redirect_uri without a port.
|
||||
req = _make_trusted_request("https://llm.example.com:443/")
|
||||
validate_trusted_redirect_uri(req, "https://llm.example.com/cb")
|
||||
|
||||
# And the symmetric case — redirect_uri has the explicit port.
|
||||
req2 = _make_trusted_request("https://llm.example.com/")
|
||||
validate_trusted_redirect_uri(req2, "https://llm.example.com:443/cb")
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_accepts_loopback():
|
||||
"""Native MCP client flow: loopback is still allowed."""
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
validate_trusted_redirect_uri,
|
||||
)
|
||||
|
||||
req = _mock_request_with_base_url("https://proxy.example.com/")
|
||||
req = _make_trusted_request("https://llm.example.com/")
|
||||
for uri in (
|
||||
"http://localhost:3000/cb",
|
||||
"http://127.0.0.1:3000/cb",
|
||||
"http://127.0.0.55/cb",
|
||||
"http://[::1]/cb",
|
||||
):
|
||||
validate_trusted_redirect_uri(req, uri)
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_rejects_cross_origin_by_default(
|
||||
monkeypatch,
|
||||
):
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
validate_trusted_redirect_uri,
|
||||
)
|
||||
|
||||
monkeypatch.delenv("MCP_TRUSTED_REDIRECT_ORIGINS", raising=False)
|
||||
req = _make_trusted_request("https://llm.example.com/")
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
validate_trusted_redirect_uri(req, "https://attacker.example.net/cb")
|
||||
assert exc.value.status_code == 400
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_rejects_fragment_and_bad_scheme():
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
validate_trusted_redirect_uri,
|
||||
)
|
||||
|
||||
req = _make_trusted_request("https://llm.example.com/")
|
||||
for uri in (
|
||||
"https://llm.example.com/cb#frag", # fragment
|
||||
"ftp://llm.example.com/cb", # unsupported scheme
|
||||
"https:///no-netloc", # missing netloc
|
||||
):
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
validate_trusted_redirect_uri(req, uri)
|
||||
assert exc.value.status_code == 400, uri
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_rejects_scheme_mismatch_on_same_host():
|
||||
"""Regression: an attacker who can serve http on the proxy's own
|
||||
host (e.g. by MITMing an unencrypted LAN hop) must not be able to
|
||||
pass same-origin validation — scheme must match as well as host."""
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
validate_trusted_redirect_uri,
|
||||
)
|
||||
|
||||
req = _make_trusted_request("https://llm.example.com/")
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
validate_trusted_redirect_uri(req, "http://llm.example.com/ui/callback")
|
||||
assert exc.value.status_code == 400
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_rejects_userinfo(monkeypatch):
|
||||
"""VERIA finding: an attacker can hide the real destination host in
|
||||
the post-``@`` portion of the URL, while the pre-``@`` userinfo is
|
||||
styled to look like an allowlisted host. Without an explicit
|
||||
username/password check, a wildcard allowlist that splits the raw
|
||||
netloc on ``:`` sees ``app.example.com`` and accepts; the browser
|
||||
then navigates to ``attacker.example`` with the authorization code.
|
||||
|
||||
Reject userinfo at every tier — same-origin, loopback, exact-entry
|
||||
allowlist, and wildcard allowlist — so the bypass is closed on
|
||||
every path through the validator.
|
||||
"""
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
validate_trusted_redirect_uri,
|
||||
)
|
||||
|
||||
# (1) Wildcard allowlist — the original VERIA vector, including the
|
||||
# ``:443`` inside userinfo that makes the raw netloc split deceptive.
|
||||
monkeypatch.setenv("MCP_TRUSTED_REDIRECT_ORIGINS", "*.example.com")
|
||||
req = _make_trusted_request("https://llm.other-proxy.com/")
|
||||
for uri in (
|
||||
"https://app.example.com:443@attacker.example/cb",
|
||||
"https://app.example.com@attacker.example/cb",
|
||||
):
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
validate_trusted_redirect_uri(req, uri)
|
||||
assert exc.value.status_code == 400, uri
|
||||
|
||||
# (2) Exact-entry allowlist — same class of bypass, different path.
|
||||
monkeypatch.setenv("MCP_TRUSTED_REDIRECT_ORIGINS", "app.example.com")
|
||||
req = _make_trusted_request("https://llm.other-proxy.com/")
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
validate_trusted_redirect_uri(
|
||||
req, "https://app.example.com@attacker.example/cb"
|
||||
)
|
||||
assert exc.value.status_code == 400
|
||||
|
||||
# (3) Same-origin path — userinfo that mimics the proxy's host.
|
||||
monkeypatch.delenv("MCP_TRUSTED_REDIRECT_ORIGINS", raising=False)
|
||||
req = _make_trusted_request("https://llm.example.com/")
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
validate_trusted_redirect_uri(
|
||||
req, "https://llm.example.com@attacker.example/cb"
|
||||
)
|
||||
assert exc.value.status_code == 400
|
||||
|
||||
# (4) Loopback path — userinfo that mimics 127.0.0.1.
|
||||
req = _make_trusted_request("https://llm.example.com/")
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
validate_trusted_redirect_uri(req, "http://127.0.0.1@attacker.example/cb")
|
||||
assert exc.value.status_code == 400
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_rejects_backslash_in_netloc(monkeypatch):
|
||||
"""VERIA finding: urlparse keeps backslashes in ``netloc``, but
|
||||
browsers normalize ``\\`` to ``/`` on http(s) URLs and treat it as
|
||||
the start of the path. An allowlist of ``*.example.com`` would
|
||||
accept ``https://attacker.net\\app.example.com/cb`` (the raw netloc
|
||||
ends with ``.example.com``) while the browser navigates to
|
||||
``attacker.net`` and delivers the authorization code there.
|
||||
|
||||
Reject on every path through the validator — same-origin,
|
||||
exact-entry, and wildcard — by bouncing the netloc before any
|
||||
matching runs.
|
||||
"""
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
validate_trusted_redirect_uri,
|
||||
)
|
||||
|
||||
# (1) Wildcard allowlist — the VERIA vector.
|
||||
monkeypatch.setenv("MCP_TRUSTED_REDIRECT_ORIGINS", "*.example.com")
|
||||
req = _make_trusted_request("https://llm.other-proxy.com/")
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
validate_trusted_redirect_uri(req, "https://attacker.net\\app.example.com/cb")
|
||||
assert exc.value.status_code == 400
|
||||
|
||||
# (2) Exact-entry allowlist — same split, different match path.
|
||||
monkeypatch.setenv("MCP_TRUSTED_REDIRECT_ORIGINS", "app.example.com")
|
||||
req = _make_trusted_request("https://llm.other-proxy.com/")
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
validate_trusted_redirect_uri(req, "https://attacker.net\\app.example.com/cb")
|
||||
assert exc.value.status_code == 400
|
||||
|
||||
# (3) Same-origin path — backslash that mimics the proxy's host.
|
||||
monkeypatch.delenv("MCP_TRUSTED_REDIRECT_ORIGINS", raising=False)
|
||||
req = _make_trusted_request("https://llm.example.com/")
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
validate_trusted_redirect_uri(req, "https://attacker.net\\llm.example.com/cb")
|
||||
assert exc.value.status_code == 400
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_allowlist_entry_with_default_port(monkeypatch):
|
||||
"""Regression: operators who write ``app.example.com:443`` in
|
||||
``MCP_TRUSTED_REDIRECT_ORIGINS`` (natural when copy-pasting from a
|
||||
browser address bar or load-balancer log) must still match a
|
||||
port-less redirect_uri. The redirect_uri's ``:443`` is normalized
|
||||
away for the same-origin compare; the allowlist side has to apply
|
||||
the same normalization or the comparison is asymmetric and silently
|
||||
fails."""
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
_parse_trusted_redirect_origins,
|
||||
validate_trusted_redirect_uri,
|
||||
)
|
||||
|
||||
monkeypatch.setenv("MCP_TRUSTED_REDIRECT_ORIGINS", "app.example.com:443")
|
||||
# Verify the parse step itself drops the default port.
|
||||
assert _parse_trusted_redirect_origins() == ["app.example.com"]
|
||||
|
||||
req = _make_trusted_request("https://llm.example.com/")
|
||||
# Port-less redirect_uri — should match the :443 env entry.
|
||||
validate_trusted_redirect_uri(req, "https://app.example.com/cb")
|
||||
# Explicit :443 on both sides — should still match.
|
||||
validate_trusted_redirect_uri(req, "https://app.example.com:443/cb")
|
||||
# Non-default port on the redirect_uri — must NOT match a default-port entry.
|
||||
with pytest.raises(HTTPException):
|
||||
validate_trusted_redirect_uri(req, "https://app.example.com:8443/cb")
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_accepts_exact_allowlisted_host(monkeypatch):
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
validate_trusted_redirect_uri,
|
||||
)
|
||||
|
||||
monkeypatch.setenv(
|
||||
"MCP_TRUSTED_REDIRECT_ORIGINS",
|
||||
"app.example.com, https://other.example.com/",
|
||||
)
|
||||
req = _make_trusted_request("https://llm.example.com/")
|
||||
# Exact allowlisted host — accepted.
|
||||
validate_trusted_redirect_uri(req, "https://app.example.com/oauth/cb")
|
||||
# Path component on the env entry should be stripped at parse time;
|
||||
# the URL still resolves to an allowlisted host.
|
||||
validate_trusted_redirect_uri(req, "https://other.example.com/anything")
|
||||
# An unrelated host still fails.
|
||||
with pytest.raises(HTTPException):
|
||||
validate_trusted_redirect_uri(req, "https://different.example.com/cb")
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_allowlist_rejects_http_even_on_listed_host(
|
||||
monkeypatch,
|
||||
):
|
||||
"""An attacker must not be able to elevate to the allowlist by
|
||||
serving http:// on the listed host — only https is accepted for
|
||||
non-loopback allowlist entries."""
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
validate_trusted_redirect_uri,
|
||||
)
|
||||
|
||||
monkeypatch.setenv("MCP_TRUSTED_REDIRECT_ORIGINS", "app.example.com")
|
||||
req = _make_trusted_request("https://llm.example.com/")
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
validate_trusted_redirect_uri(req, "http://app.example.com/cb")
|
||||
assert exc.value.status_code == 400
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_wildcard_allowlist(monkeypatch):
|
||||
"""``*.suffix`` entries match any strictly-deeper subdomain of
|
||||
``suffix`` but must not match the bare suffix, nor unrelated domains
|
||||
that happen to end with the same characters."""
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
validate_trusted_redirect_uri,
|
||||
)
|
||||
|
||||
monkeypatch.setenv("MCP_TRUSTED_REDIRECT_ORIGINS", "*.example.com")
|
||||
req = _make_trusted_request("https://llm.other-proxy.com/")
|
||||
|
||||
# Direct subdomain — accepted.
|
||||
validate_trusted_redirect_uri(req, "https://app.example.com/cb")
|
||||
# Nested subdomain — accepted.
|
||||
validate_trusted_redirect_uri(req, "https://foo.bar.example.com/cb")
|
||||
|
||||
# Bare suffix — NOT accepted (wildcard requires a proper subdomain).
|
||||
with pytest.raises(HTTPException):
|
||||
validate_trusted_redirect_uri(req, "https://example.com/cb")
|
||||
|
||||
# Similar-looking domain that isn't a subdomain — NOT accepted.
|
||||
with pytest.raises(HTTPException):
|
||||
validate_trusted_redirect_uri(req, "https://evil-example.com/cb")
|
||||
with pytest.raises(HTTPException):
|
||||
validate_trusted_redirect_uri(req, "https://example.com.attacker.net/cb")
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_wildcard_rejects_http(monkeypatch):
|
||||
"""The https-only gate applies to wildcard entries too."""
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
validate_trusted_redirect_uri,
|
||||
)
|
||||
|
||||
monkeypatch.setenv("MCP_TRUSTED_REDIRECT_ORIGINS", "*.example.com")
|
||||
req = _make_trusted_request("https://llm.other-proxy.com/")
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
validate_trusted_redirect_uri(req, "http://app.example.com/cb")
|
||||
assert exc.value.status_code == 400
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_wildcard_host_with_port_still_matches(
|
||||
monkeypatch,
|
||||
):
|
||||
"""Wildcard entries don't express port constraints — an allowlisted
|
||||
subdomain should match regardless of explicit port on the URL."""
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
validate_trusted_redirect_uri,
|
||||
)
|
||||
|
||||
monkeypatch.setenv("MCP_TRUSTED_REDIRECT_ORIGINS", "*.example.com")
|
||||
req = _make_trusted_request("https://llm.other-proxy.com/")
|
||||
validate_trusted_redirect_uri(req, "https://app.example.com:8443/cb")
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_accepts_ipv6_loopback_with_default_port():
|
||||
"""IPv6 loopback with explicit ``:443`` on an ``https`` URL should
|
||||
still match — exercises ``_strip_default_port``'s IPv6 branch."""
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
validate_trusted_redirect_uri,
|
||||
)
|
||||
|
||||
req = _make_trusted_request("https://[::1]/")
|
||||
validate_trusted_redirect_uri(req, "https://[::1]:443/cb")
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_tolerates_malformed_env_entries(monkeypatch):
|
||||
"""Operators occasionally mis-type env values (empty items, bare
|
||||
``*.``, non-numeric ports). None of those should raise; unmatched
|
||||
entries must simply fail to grant access while well-formed entries
|
||||
in the same list continue to work."""
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
validate_trusted_redirect_uri,
|
||||
)
|
||||
|
||||
monkeypatch.setenv(
|
||||
"MCP_TRUSTED_REDIRECT_ORIGINS",
|
||||
", ,*., foo:notaport, app.example.com",
|
||||
)
|
||||
req = _make_trusted_request("https://llm.example.com/")
|
||||
# Well-formed entry still works.
|
||||
validate_trusted_redirect_uri(req, "https://app.example.com/cb")
|
||||
# Bare ``*.`` grants nothing.
|
||||
with pytest.raises(HTTPException):
|
||||
validate_trusted_redirect_uri(req, "https://example.com/cb")
|
||||
# Non-numeric port entry is ignored (doesn't grant access).
|
||||
with pytest.raises(HTTPException):
|
||||
validate_trusted_redirect_uri(req, "https://foo.example.net/cb")
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_rejects_wildcard_entry_with_dot_leading_suffix(
|
||||
monkeypatch,
|
||||
):
|
||||
"""A wildcard entry like ``*..example.com`` has a suffix that starts
|
||||
with ``.``, which would otherwise match ``anything.example.com`` via
|
||||
the ``host.endswith("." + suffix)`` branch by accepting a netloc
|
||||
whose own leading ``.`` makes it look like a deeper subdomain.
|
||||
Operators who mistype an extra dot should get an ignored entry, not
|
||||
a broader match than they intended."""
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
validate_trusted_redirect_uri,
|
||||
)
|
||||
|
||||
monkeypatch.setenv("MCP_TRUSTED_REDIRECT_ORIGINS", "*..example.com")
|
||||
req = _make_trusted_request("https://llm.example.com/")
|
||||
|
||||
# None of these should resolve against the malformed wildcard entry.
|
||||
for uri in (
|
||||
"https://app.example.com/cb",
|
||||
"https://foo.bar.example.com/cb",
|
||||
"https://example.com/cb",
|
||||
):
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
validate_trusted_redirect_uri(req, uri)
|
||||
assert exc.value.status_code == 400
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_falls_through_when_origin_lookup_fails():
|
||||
"""If ``get_request_base_url`` can't determine the proxy's origin,
|
||||
same-origin is skipped silently but loopback + allowlist paths are
|
||||
still reachable."""
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
validate_trusted_redirect_uri,
|
||||
)
|
||||
|
||||
class _ExplodingRequest:
|
||||
# Accessing ``.base_url`` is what ``get_request_base_url``
|
||||
# reaches for first; raising here lets us exercise the swallowed-
|
||||
# error fallback without monkey-patching imports.
|
||||
base_url = property(lambda self: (_ for _ in ()).throw(RuntimeError("boom")))
|
||||
headers: dict = {}
|
||||
|
||||
req = _ExplodingRequest()
|
||||
# Loopback still accepted despite origin lookup failure.
|
||||
validate_trusted_redirect_uri(req, "http://127.0.0.1:3000/cb")
|
||||
validate_trusted_redirect_uri(req, "http://localhost:3000/cb")
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_rejects_external_origin():
|
||||
"""An attacker-controlled origin must still be rejected."""
|
||||
def test_strip_default_port_empty_netloc():
|
||||
"""``_strip_default_port("", "")`` should round-trip — validator
|
||||
rejects empty-netloc URLs upstream so this is purely a defensive
|
||||
contract on the helper itself."""
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import _strip_default_port
|
||||
|
||||
assert _strip_default_port("https", "") == ""
|
||||
|
||||
|
||||
def test_strip_default_port_handles_non_numeric_port():
|
||||
"""Raw netloc with a non-numeric port is returned unchanged. Reached
|
||||
in practice when a malformed ``Host`` header survives upstream
|
||||
parsing — we stay out of its way rather than 500ing."""
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import _strip_default_port
|
||||
|
||||
assert _strip_default_port("https", "foo.com:bar") == "foo.com:bar"
|
||||
assert _strip_default_port("https", "[::1]:bar") == "[::1]:bar"
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_rejects_public_ip_without_allowlist():
|
||||
"""A redirect_uri whose host is a public IP (parseable by
|
||||
``ip_address`` but not loopback) must fail all three tiers and 400."""
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
validate_trusted_redirect_uri,
|
||||
)
|
||||
|
||||
req = _mock_request_with_base_url("https://proxy.example.com/")
|
||||
req = _make_trusted_request("https://llm.example.com/")
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
validate_trusted_redirect_uri(req, "https://attacker.example.com/cb")
|
||||
validate_trusted_redirect_uri(req, "https://1.2.3.4/cb")
|
||||
assert exc.value.status_code == 400
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_rejects_scheme_mismatch():
|
||||
"""https→http (or vice versa) on the same host is not same-origin."""
|
||||
def test_parse_trusted_redirect_origins_drops_bare_path_entries(monkeypatch):
|
||||
"""``/foo`` has a scheme-less leading slash and would strip to the
|
||||
empty string — drop silently rather than allowlisting empty
|
||||
origins."""
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
validate_trusted_redirect_uri,
|
||||
_parse_trusted_redirect_origins,
|
||||
)
|
||||
|
||||
req = _mock_request_with_base_url("https://proxy.example.com/")
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
validate_trusted_redirect_uri(req, "http://proxy.example.com/ui/callback")
|
||||
assert exc.value.status_code == 400
|
||||
|
||||
|
||||
def test_validate_trusted_redirect_uri_rejects_fragment():
|
||||
from litellm.proxy._experimental.mcp_server.oauth_utils import (
|
||||
validate_trusted_redirect_uri,
|
||||
monkeypatch.setenv(
|
||||
"MCP_TRUSTED_REDIRECT_ORIGINS", "https:///, /foo, app.example.com"
|
||||
)
|
||||
|
||||
req = _mock_request_with_base_url("https://proxy.example.com/")
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
validate_trusted_redirect_uri(req, "https://proxy.example.com/ui/cb#code=1")
|
||||
assert exc.value.status_code == 400
|
||||
# The two malformed entries drop out; only the real host survives.
|
||||
assert _parse_trusted_redirect_origins() == ["app.example.com"]
|
||||
|
||||
Loading…
Reference in New Issue
Block a user