From 3c9a8690d1e61495916f1a0cef5e07020921e848 Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Wed, 29 Apr 2026 22:18:18 +0000
Subject: [PATCH 1/6] fix(auth): gate oauth2-proxy header trust on premium +
 privileged-field denylist
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

``handle_oauth2_proxy_request`` reads HTTP request headers per the
admin-set ``oauth2_config_mappings`` and constructs a
``UserAPIKeyAuth`` from the values. Two failure modes:

1. **Premium parity.** Sibling auth paths
   (``enable_oauth2_auth``, ``enable_jwt_auth``) require
   ``premium_user``; this path did not, so any open-source deployment
   could turn the feature on without realising it requires a hardened
   reverse-proxy topology. Added the ``premium_user`` gate.

2. **Privileged-field denylist.** Without a denylist, an admin who
   maps the wrong header to ``user_role`` (or whose reverse proxy
   leaks the header from upstream user input) lets any caller send
   ``X-User-Role: proxy_admin`` and gain full admin access — Pydantic
   coerces the string into the ``LitellmUserRoles.PROXY_ADMIN`` enum.
   Mapping any field in ``PRIVILEGED_OAUTH2_PROXY_FIELDS``
   (``user_role``, ``api_key``, ``token``, ``permissions``,
   ``allowed_routes``, budget/limit fields, ``metadata``) raises at
   request time so the misconfiguration surfaces loudly rather than
   as a silent privesc.

Operators who genuinely need a trusted upstream to assert one of
these privileged fields should switch to JWT auth (signature-validated)
rather than header-trust.

Tests:

- ``test_returns_auth_for_simple_user_id_mapping``: legitimate
  identity-only mapping still works.
- ``test_rejects_when_not_premium``: open-source deployments get a
  clear enterprise-feature error.
- ``test_refuses_to_map_privileged_fields``: parametrized over every
  entry in the denylist — each is rejected at request time.
- ``test_user_role_header_forgery_attack_is_blocked``: end-to-end
  shape of the GHSA-5c3m-qffq-4r9m attack; rejected before auth
  object construction.
- ``test_safe_fields_still_pass_through``: documented usage
  (``user_id``, ``user_email``, ``team_id``, ``models``) is
  unaffected.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 litellm/proxy/auth/oauth2_proxy_hook.py       | 106 ++++++++--
 .../proxy/auth/test_oauth2_proxy_hook.py      | 189 ++++++++++++++++++
 2 files changed, 277 insertions(+), 18 deletions(-)
 create mode 100644 tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py

diff --git a/litellm/proxy/auth/oauth2_proxy_hook.py b/litellm/proxy/auth/oauth2_proxy_hook.py
index 0dc696bc45..341d2b477b 100644
--- a/litellm/proxy/auth/oauth2_proxy_hook.py
+++ b/litellm/proxy/auth/oauth2_proxy_hook.py
@@ -1,19 +1,78 @@
-from typing import Any, Dict
+from typing import Any, Dict, FrozenSet
 
 from fastapi import Request
 
 from litellm._logging import verbose_proxy_logger
-from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy._types import CommonProxyErrors, UserAPIKeyAuth
+
+# Fields on ``UserAPIKeyAuth`` that grant privileges directly (``user_role``
+# is the canonical privesc — coerced from the string ``"proxy_admin"`` into
+# ``LitellmUserRoles.PROXY_ADMIN`` by Pydantic) or break trust assumptions
+# (``api_key`` / ``token`` short-circuit the validated-key contract;
+# ``permissions`` / ``allowed_routes`` directly grant route access; budget
+# and limit fields can be set to wild values to bypass enforcement;
+# ``metadata`` is too broad to safely admit from caller-controlled headers).
+#
+# Operators who legitimately need any of these to flow from a trusted
+# upstream proxy should switch to JWT authentication, which validates a
+# signature on the assertion rather than blindly trusting headers.
+PRIVILEGED_OAUTH2_PROXY_FIELDS: FrozenSet[str] = frozenset(
+    {
+        "user_role",
+        "api_key",
+        "token",
+        "key_alias",
+        "key_name",
+        "permissions",
+        "allowed_routes",
+        "max_budget",
+        "spend",
+        "model_max_budget",
+        "model_spend",
+        "tpm_limit",
+        "rpm_limit",
+        "team_max_budget",
+        "team_spend",
+        "blocked",
+        "metadata",
+    }
+)
 
 
 async def handle_oauth2_proxy_request(request: Request) -> UserAPIKeyAuth:
     """
-    Handle request from oauth2 proxy.
+    Resolve a ``UserAPIKeyAuth`` from request headers per the admin-set
+    ``oauth2_config_mappings``.
+
+    The auth model assumes the proxy is deployed behind a trusted OAuth2
+    reverse proxy that injects authenticated identity headers (e.g.
+    oauth2-proxy, Authelia). Two safeguards above and beyond that
+    deployment assumption:
+
+    1. **Premium gate.** The sibling auth paths (``enable_oauth2_auth``
+       and ``enable_jwt_auth``) require ``premium_user``; this path
+       previously did not, which let any open-source deployment turn
+       the feature on without realising it requires a hardened
+       deployment topology.
+    2. **Privileged-field denylist.** ``oauth2_config_mappings`` maps
+       header names to ``UserAPIKeyAuth`` fields. Without a denylist,
+       an admin who maps the wrong header to ``user_role`` (or who
+       hasn't fully locked down their reverse proxy) lets any caller
+       set the ``user_role`` header to ``"proxy_admin"`` and gain full
+       admin privileges — Pydantic coerces the string into the enum.
+       Mapping any privileged field is rejected at startup-style auth
+       time so the misconfiguration surfaces loudly rather than as a
+       silent privesc.
     """
-    from litellm.proxy.proxy_server import general_settings
+    from litellm.proxy.proxy_server import general_settings, premium_user
+
+    if premium_user is not True:
+        raise ValueError(
+            "Oauth2 proxy auth is an enterprise-only feature. "
+            + CommonProxyErrors.not_premium_user.value
+        )
 
     verbose_proxy_logger.debug("Handling oauth2 proxy request")
-    # Define the OAuth2 config mappings
     oauth2_config_mappings: Dict[str, str] = (
         general_settings.get("oauth2_config_mappings") or {}
     )
@@ -21,21 +80,33 @@ async def handle_oauth2_proxy_request(request: Request) -> UserAPIKeyAuth:
 
     if not oauth2_config_mappings:
         raise ValueError("Oauth2 config mappings not found in general_settings")
-    # Initialize a dictionary to store the mapped values
-    auth_data: Dict[str, Any] = {}
 
-    # Extract values from headers based on the mappings
+    privileged_mapped = sorted(
+        set(oauth2_config_mappings.keys()) & PRIVILEGED_OAUTH2_PROXY_FIELDS
+    )
+    if privileged_mapped:
+        raise ValueError(
+            "Oauth2 proxy auth refuses to map privileged UserAPIKeyAuth "
+            f"fields from request headers: {privileged_mapped}. These "
+            "fields would grant privileges (e.g. proxy_admin), bypass "
+            "budget enforcement, or short-circuit key validation if a "
+            "caller can spoof the corresponding header. If you need a "
+            "trusted upstream to assert one of these, use JWT auth "
+            "(signature-validated) instead of header-trust."
+        )
+
+    auth_data: Dict[str, Any] = {}
     for key, header in oauth2_config_mappings.items():
         value = request.headers.get(header)
-        if value:
-            # Convert max_budget to float if present
-            if key == "max_budget":
-                auth_data[key] = float(value)
-            # Convert models to list if present
-            elif key == "models":
-                auth_data[key] = [model.strip() for model in value.split(",")]
-            else:
-                auth_data[key] = value
+        if not value:
+            continue
+        if key == "max_budget":
+            auth_data[key] = float(value)
+        elif key == "models":
+            auth_data[key] = [model.strip() for model in value.split(",")]
+        else:
+            auth_data[key] = value
+
     verbose_proxy_logger.debug(
         "Auth data before creating UserAPIKeyAuth object: keys=%s",
         list(auth_data.keys()),
@@ -45,5 +116,4 @@ async def handle_oauth2_proxy_request(request: Request) -> UserAPIKeyAuth:
         "UserAPIKeyAuth object created with keys: %s",
         list(user_api_key_auth.__fields_set__),
     )
-    # Create and return UserAPIKeyAuth object
     return user_api_key_auth
diff --git a/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py b/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py
new file mode 100644
index 0000000000..87f8c45087
--- /dev/null
+++ b/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py
@@ -0,0 +1,189 @@
+"""
+Regression tests for the OAuth2-proxy header-forgery fix
+(GHSA-5c3m-qffq-4r9m).
+
+The hook reads HTTP request headers per ``oauth2_config_mappings`` and
+constructs a ``UserAPIKeyAuth`` from them. Two separate failure modes
+the fix closes:
+
+1. The path was not gated on ``premium_user`` (the sibling
+   ``enable_oauth2_auth`` and ``enable_jwt_auth`` paths are). Open-source
+   deployments could enable the feature without realising it requires
+   a hardened deployment topology.
+2. Any ``UserAPIKeyAuth`` field could be mapped from a header — including
+   ``user_role``, which Pydantic coerces from the string ``"proxy_admin"``
+   into ``LitellmUserRoles.PROXY_ADMIN``. An attacker who reaches the
+   proxy directly (or via a misconfigured reverse proxy) sets the mapped
+   header and gains full admin privileges.
+"""
+
+import os
+import sys
+from unittest.mock import patch
+
+import pytest
+from fastapi import Request
+from starlette.datastructures import Headers
+
+sys.path.insert(0, os.path.abspath("../../../.."))
+
+from litellm.proxy._types import LitellmUserRoles
+from litellm.proxy.auth.oauth2_proxy_hook import (
+    PRIVILEGED_OAUTH2_PROXY_FIELDS,
+    handle_oauth2_proxy_request,
+)
+
+
+def _request_with_headers(headers: dict) -> Request:
+    scope = {
+        "type": "http",
+        "headers": [(k.lower().encode(), v.encode()) for k, v in headers.items()],
+    }
+    request = Request(scope=scope)
+    request._headers = Headers(headers)
+    return request
+
+
+@pytest.fixture
+def premium_proxy_settings(monkeypatch):
+    """
+    Patch the proxy_server module attributes the hook reads so each test
+    starts from "premium=True, mapping={user_id: x-user-id}".
+    """
+    import litellm.proxy.proxy_server as proxy_server
+
+    monkeypatch.setattr(proxy_server, "premium_user", True, raising=False)
+    monkeypatch.setattr(
+        proxy_server,
+        "general_settings",
+        {"oauth2_config_mappings": {"user_id": "x-user-id"}},
+        raising=False,
+    )
+
+
+@pytest.mark.asyncio
+async def test_returns_auth_for_simple_user_id_mapping(premium_proxy_settings):
+    request = _request_with_headers({"x-user-id": "alice"})
+
+    auth = await handle_oauth2_proxy_request(request)
+
+    assert auth.user_id == "alice"
+    assert auth.user_role is None
+
+
+@pytest.mark.asyncio
+async def test_rejects_when_not_premium(monkeypatch):
+    import litellm.proxy.proxy_server as proxy_server
+
+    monkeypatch.setattr(proxy_server, "premium_user", False, raising=False)
+    monkeypatch.setattr(
+        proxy_server,
+        "general_settings",
+        {"oauth2_config_mappings": {"user_id": "x-user-id"}},
+        raising=False,
+    )
+    request = _request_with_headers({"x-user-id": "alice"})
+
+    with pytest.raises(ValueError, match="enterprise"):
+        await handle_oauth2_proxy_request(request)
+
+
+@pytest.mark.parametrize(
+    "privileged_field",
+    sorted(PRIVILEGED_OAUTH2_PROXY_FIELDS),
+)
+@pytest.mark.asyncio
+async def test_refuses_to_map_privileged_fields(monkeypatch, privileged_field):
+    """
+    The exact privesc shape from GHSA-5c3m-qffq-4r9m: an admin maps
+    ``user_role`` (or any other privileged field) to a header and a
+    caller forges ``X-User-Role: proxy_admin``. The hook must reject
+    this configuration outright at request time.
+    """
+    import litellm.proxy.proxy_server as proxy_server
+
+    monkeypatch.setattr(proxy_server, "premium_user", True, raising=False)
+    monkeypatch.setattr(
+        proxy_server,
+        "general_settings",
+        {"oauth2_config_mappings": {privileged_field: f"x-{privileged_field}"}},
+        raising=False,
+    )
+    request = _request_with_headers({f"x-{privileged_field}": "proxy_admin"})
+
+    with pytest.raises(ValueError) as exc:
+        await handle_oauth2_proxy_request(request)
+    assert privileged_field in str(exc.value)
+
+
+@pytest.mark.asyncio
+async def test_user_role_header_forgery_attack_is_blocked(monkeypatch):
+    """
+    End-to-end shape from the GHSA: with ``user_role`` mapped, a forged
+    ``X-User-Role: proxy_admin`` header would have produced a
+    ``UserAPIKeyAuth`` with PROXY_ADMIN role. Now the request raises
+    before any auth object is constructed.
+    """
+    import litellm.proxy.proxy_server as proxy_server
+
+    monkeypatch.setattr(proxy_server, "premium_user", True, raising=False)
+    monkeypatch.setattr(
+        proxy_server,
+        "general_settings",
+        {
+            "oauth2_config_mappings": {
+                "user_id": "x-user-id",
+                "user_role": "x-user-role",
+            }
+        },
+        raising=False,
+    )
+    request = _request_with_headers(
+        {
+            "x-user-id": "attacker",
+            "x-user-role": LitellmUserRoles.PROXY_ADMIN.value,
+        }
+    )
+
+    with pytest.raises(ValueError, match="user_role"):
+        await handle_oauth2_proxy_request(request)
+
+
+@pytest.mark.asyncio
+async def test_safe_fields_still_pass_through(monkeypatch):
+    """
+    Sanity check that non-privileged fields (the documented use case
+    for OAuth2 proxy auth — asserting identity from a trusted upstream)
+    still work after the fix.
+    """
+    import litellm.proxy.proxy_server as proxy_server
+
+    monkeypatch.setattr(proxy_server, "premium_user", True, raising=False)
+    monkeypatch.setattr(
+        proxy_server,
+        "general_settings",
+        {
+            "oauth2_config_mappings": {
+                "user_id": "x-user-id",
+                "user_email": "x-user-email",
+                "team_id": "x-team-id",
+                "models": "x-models",
+            }
+        },
+        raising=False,
+    )
+    request = _request_with_headers(
+        {
+            "x-user-id": "alice",
+            "x-user-email": "alice@example.com",
+            "x-team-id": "team-corp",
+            "x-models": "gpt-4, gpt-3.5-turbo",
+        }
+    )
+
+    auth = await handle_oauth2_proxy_request(request)
+
+    assert auth.user_id == "alice"
+    assert auth.user_email == "alice@example.com"
+    assert auth.team_id == "team-corp"
+    assert auth.models == ["gpt-4", "gpt-3.5-turbo"]

From e6867c143ae831ed9c6927034f4233048711820a Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Wed, 29 Apr 2026 22:23:29 +0000
Subject: [PATCH 2/6] =?UTF-8?q?chore(oauth2-proxy):=20/simplify=20pass=20?=
 =?UTF-8?q?=E2=80=94=20drop=20dead=20max=5Fbudget=20branch=20+=20DRY=20tes?=
 =?UTF-8?q?ts?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two cleanups from the /simplify review pass:

* The header-mapping loop had a special-case ``if key == "max_budget":
  auth_data[key] = float(value)`` branch. Since ``max_budget`` is now
  in ``PRIVILEGED_OAUTH2_PROXY_FIELDS``, the denylist check rejects
  the configuration before the loop runs — the float-conversion
  branch is unreachable. Removed.

* Four tests independently called
  ``monkeypatch.setattr(proxy_server, "premium_user", ...)`` and
  ``monkeypatch.setattr(proxy_server, "general_settings", ...)`` with
  almost-identical bodies. Replaced with a ``configure_proxy`` fixture
  that yields a single callable —
  ``configure_proxy(premium=False)`` /
  ``configure_proxy(mappings={...})`` — so each test's setup is one
  line. The previously-unused ``premium_proxy_settings`` fixture is
  removed.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 litellm/proxy/auth/oauth2_proxy_hook.py       |   4 +-
 .../proxy/auth/test_oauth2_proxy_hook.py      | 119 +++++++-----------
 2 files changed, 43 insertions(+), 80 deletions(-)

diff --git a/litellm/proxy/auth/oauth2_proxy_hook.py b/litellm/proxy/auth/oauth2_proxy_hook.py
index 341d2b477b..0f7cfa4222 100644
--- a/litellm/proxy/auth/oauth2_proxy_hook.py
+++ b/litellm/proxy/auth/oauth2_proxy_hook.py
@@ -100,9 +100,7 @@ async def handle_oauth2_proxy_request(request: Request) -> UserAPIKeyAuth:
         value = request.headers.get(header)
         if not value:
             continue
-        if key == "max_budget":
-            auth_data[key] = float(value)
-        elif key == "models":
+        if key == "models":
             auth_data[key] = [model.strip() for model in value.split(",")]
         else:
             auth_data[key] = value
diff --git a/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py b/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py
index 87f8c45087..e51882a3b1 100644
--- a/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py
+++ b/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py
@@ -45,24 +45,32 @@ def _request_with_headers(headers: dict) -> Request:
 
 
 @pytest.fixture
-def premium_proxy_settings(monkeypatch):
+def configure_proxy(monkeypatch):
     """
-    Patch the proxy_server module attributes the hook reads so each test
-    starts from "premium=True, mapping={user_id: x-user-id}".
+    Yields a callable that sets ``premium_user`` and
+    ``oauth2_config_mappings`` on the proxy_server module for the
+    duration of one test. Default is premium=True with a single
+    ``user_id -> x-user-id`` mapping.
     """
     import litellm.proxy.proxy_server as proxy_server
 
-    monkeypatch.setattr(proxy_server, "premium_user", True, raising=False)
-    monkeypatch.setattr(
-        proxy_server,
-        "general_settings",
-        {"oauth2_config_mappings": {"user_id": "x-user-id"}},
-        raising=False,
-    )
+    def _configure(*, premium=True, mappings=None):
+        if mappings is None:
+            mappings = {"user_id": "x-user-id"}
+        monkeypatch.setattr(proxy_server, "premium_user", premium, raising=False)
+        monkeypatch.setattr(
+            proxy_server,
+            "general_settings",
+            {"oauth2_config_mappings": mappings},
+            raising=False,
+        )
+
+    return _configure
 
 
 @pytest.mark.asyncio
-async def test_returns_auth_for_simple_user_id_mapping(premium_proxy_settings):
+async def test_returns_auth_for_simple_user_id_mapping(configure_proxy):
+    configure_proxy()
     request = _request_with_headers({"x-user-id": "alice"})
 
     auth = await handle_oauth2_proxy_request(request)
@@ -72,16 +80,8 @@ async def test_returns_auth_for_simple_user_id_mapping(premium_proxy_settings):
 
 
 @pytest.mark.asyncio
-async def test_rejects_when_not_premium(monkeypatch):
-    import litellm.proxy.proxy_server as proxy_server
-
-    monkeypatch.setattr(proxy_server, "premium_user", False, raising=False)
-    monkeypatch.setattr(
-        proxy_server,
-        "general_settings",
-        {"oauth2_config_mappings": {"user_id": "x-user-id"}},
-        raising=False,
-    )
+async def test_rejects_when_not_premium(configure_proxy):
+    configure_proxy(premium=False)
     request = _request_with_headers({"x-user-id": "alice"})
 
     with pytest.raises(ValueError, match="enterprise"):
@@ -93,22 +93,11 @@ async def test_rejects_when_not_premium(monkeypatch):
     sorted(PRIVILEGED_OAUTH2_PROXY_FIELDS),
 )
 @pytest.mark.asyncio
-async def test_refuses_to_map_privileged_fields(monkeypatch, privileged_field):
-    """
-    The exact privesc shape from GHSA-5c3m-qffq-4r9m: an admin maps
-    ``user_role`` (or any other privileged field) to a header and a
-    caller forges ``X-User-Role: proxy_admin``. The hook must reject
-    this configuration outright at request time.
-    """
-    import litellm.proxy.proxy_server as proxy_server
-
-    monkeypatch.setattr(proxy_server, "premium_user", True, raising=False)
-    monkeypatch.setattr(
-        proxy_server,
-        "general_settings",
-        {"oauth2_config_mappings": {privileged_field: f"x-{privileged_field}"}},
-        raising=False,
-    )
+async def test_refuses_to_map_privileged_fields(configure_proxy, privileged_field):
+    # GHSA-5c3m-qffq-4r9m attack shape: admin maps a privileged field
+    # to a header and a caller forges the value. The hook must reject
+    # the misconfiguration outright at request time.
+    configure_proxy(mappings={privileged_field: f"x-{privileged_field}"})
     request = _request_with_headers({f"x-{privileged_field}": "proxy_admin"})
 
     with pytest.raises(ValueError) as exc:
@@ -117,26 +106,13 @@ async def test_refuses_to_map_privileged_fields(monkeypatch, privileged_field):
 
 
 @pytest.mark.asyncio
-async def test_user_role_header_forgery_attack_is_blocked(monkeypatch):
-    """
-    End-to-end shape from the GHSA: with ``user_role`` mapped, a forged
-    ``X-User-Role: proxy_admin`` header would have produced a
-    ``UserAPIKeyAuth`` with PROXY_ADMIN role. Now the request raises
-    before any auth object is constructed.
-    """
-    import litellm.proxy.proxy_server as proxy_server
-
-    monkeypatch.setattr(proxy_server, "premium_user", True, raising=False)
-    monkeypatch.setattr(
-        proxy_server,
-        "general_settings",
-        {
-            "oauth2_config_mappings": {
-                "user_id": "x-user-id",
-                "user_role": "x-user-role",
-            }
-        },
-        raising=False,
+async def test_user_role_header_forgery_attack_is_blocked(configure_proxy):
+    # End-to-end form of the privesc: with ``user_role`` mapped, the
+    # forged ``X-User-Role: proxy_admin`` header would have produced
+    # a ``UserAPIKeyAuth(user_role=PROXY_ADMIN)``. Now rejected before
+    # any auth object is constructed.
+    configure_proxy(
+        mappings={"user_id": "x-user-id", "user_role": "x-user-role"},
     )
     request = _request_with_headers(
         {
@@ -150,27 +126,16 @@ async def test_user_role_header_forgery_attack_is_blocked(monkeypatch):
 
 
 @pytest.mark.asyncio
-async def test_safe_fields_still_pass_through(monkeypatch):
-    """
-    Sanity check that non-privileged fields (the documented use case
-    for OAuth2 proxy auth — asserting identity from a trusted upstream)
-    still work after the fix.
-    """
-    import litellm.proxy.proxy_server as proxy_server
-
-    monkeypatch.setattr(proxy_server, "premium_user", True, raising=False)
-    monkeypatch.setattr(
-        proxy_server,
-        "general_settings",
-        {
-            "oauth2_config_mappings": {
-                "user_id": "x-user-id",
-                "user_email": "x-user-email",
-                "team_id": "x-team-id",
-                "models": "x-models",
-            }
+async def test_safe_fields_still_pass_through(configure_proxy):
+    # The documented use case for OAuth2 proxy auth: identity assertion
+    # from a trusted upstream. Must remain unaffected by the denylist.
+    configure_proxy(
+        mappings={
+            "user_id": "x-user-id",
+            "user_email": "x-user-email",
+            "team_id": "x-team-id",
+            "models": "x-models",
         },
-        raising=False,
     )
     request = _request_with_headers(
         {

From b35287a062dbdd99eac053223f099200b12db9c0 Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Wed, 29 Apr 2026 22:28:57 +0000
Subject: [PATCH 3/6] fix(oauth2-proxy): switch privileged-field denylist to
 identity-only allowlist
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Greptile flagged that the denylist was incomplete: ``user_max_budget``,
``user_tpm_limit``, ``user_rpm_limit``, and ``user_spend`` were not on
it. Inspection of the auth model showed dozens more privileged fields
across the ``LiteLLM_VerificationTokenView`` hierarchy (team / org /
end-user / region budget / spend / limit fields, plus
``allowed_model_region``, ``rpm_limit_per_model``, etc.) — a denylist
of "privileged fields" is unmaintainable here.

Inverted the model. ``ALLOWED_OAUTH2_PROXY_FIELDS`` is now an
identity-only allowlist: ``user_id``, ``user_email``, ``team_id``,
``team_alias``, ``org_id``, ``models``. Any mapping to a non-identity
field is rejected at request time. Default-secure: a future field
added to ``UserAPIKeyAuth`` is automatically blocked from
header-trust.

Use case for OAuth2-proxy auth is identity assertion from a trusted
upstream. Anything beyond that (privileges, budgets, rate limits) is
policy and should be authenticated with a signature, not a header —
operators who need this should switch to JWT auth.

Tests:

- ``test_refuses_to_map_non_identity_fields`` parametrized over 22
  fields including all four ``user_*`` Greptile flagged, plus
  team/org/end-user budget/limit fields, plus a fabricated field name
  to confirm "anything not on the allowlist" is the rule.
- ``test_allowlist_is_identity_only`` locks in the allowlist's intent
  so future additions of budget / role / permission entries are caught
  in review.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 litellm/proxy/auth/oauth2_proxy_hook.py       | 87 +++++++++----------
 .../proxy/auth/test_oauth2_proxy_hook.py      | 59 +++++++++++--
 2 files changed, 96 insertions(+), 50 deletions(-)

diff --git a/litellm/proxy/auth/oauth2_proxy_hook.py b/litellm/proxy/auth/oauth2_proxy_hook.py
index 0f7cfa4222..1ba1b100a8 100644
--- a/litellm/proxy/auth/oauth2_proxy_hook.py
+++ b/litellm/proxy/auth/oauth2_proxy_hook.py
@@ -5,36 +5,32 @@ from fastapi import Request
 from litellm._logging import verbose_proxy_logger
 from litellm.proxy._types import CommonProxyErrors, UserAPIKeyAuth
 
-# Fields on ``UserAPIKeyAuth`` that grant privileges directly (``user_role``
-# is the canonical privesc — coerced from the string ``"proxy_admin"`` into
-# ``LitellmUserRoles.PROXY_ADMIN`` by Pydantic) or break trust assumptions
-# (``api_key`` / ``token`` short-circuit the validated-key contract;
-# ``permissions`` / ``allowed_routes`` directly grant route access; budget
-# and limit fields can be set to wild values to bypass enforcement;
-# ``metadata`` is too broad to safely admit from caller-controlled headers).
+# OAuth2-proxy header trust is for **identity assertion** from a trusted
+# upstream auth proxy (oauth2-proxy, Authelia, etc.). The allowlist below
+# is the only safe surface — anything else (``user_role``, ``api_key``,
+# ``permissions``, ``max_budget``, ``user_max_budget``,
+# ``team_tpm_limit``, ``end_user_max_budget``, ``allowed_model_region``,
+# and dozens of similar policy fields scattered across the
+# ``LiteLLM_VerificationTokenView`` hierarchy) is a privilege grant that
+# would let a caller forge their own enforcement parameters by sending
+# the matching header.
 #
-# Operators who legitimately need any of these to flow from a trusted
-# upstream proxy should switch to JWT authentication, which validates a
+# A denylist of "privileged fields" is unmaintainable in this codebase:
+# the auth model has ~50 budget/spend/limit/permission fields and gains
+# more with each release. An allowlist scoped to identity assertion is
+# default-secure — new fields are blocked automatically.
+#
+# Operators who need a trusted upstream to assert anything beyond
+# identity should switch to JWT authentication, which validates a
 # signature on the assertion rather than blindly trusting headers.
-PRIVILEGED_OAUTH2_PROXY_FIELDS: FrozenSet[str] = frozenset(
+ALLOWED_OAUTH2_PROXY_FIELDS: FrozenSet[str] = frozenset(
     {
-        "user_role",
-        "api_key",
-        "token",
-        "key_alias",
-        "key_name",
-        "permissions",
-        "allowed_routes",
-        "max_budget",
-        "spend",
-        "model_max_budget",
-        "model_spend",
-        "tpm_limit",
-        "rpm_limit",
-        "team_max_budget",
-        "team_spend",
-        "blocked",
-        "metadata",
+        "user_id",
+        "user_email",
+        "team_id",
+        "team_alias",
+        "org_id",
+        "models",
     }
 )
 
@@ -54,15 +50,15 @@ async def handle_oauth2_proxy_request(request: Request) -> UserAPIKeyAuth:
        previously did not, which let any open-source deployment turn
        the feature on without realising it requires a hardened
        deployment topology.
-    2. **Privileged-field denylist.** ``oauth2_config_mappings`` maps
-       header names to ``UserAPIKeyAuth`` fields. Without a denylist,
-       an admin who maps the wrong header to ``user_role`` (or who
-       hasn't fully locked down their reverse proxy) lets any caller
-       set the ``user_role`` header to ``"proxy_admin"`` and gain full
-       admin privileges — Pydantic coerces the string into the enum.
-       Mapping any privileged field is rejected at startup-style auth
-       time so the misconfiguration surfaces loudly rather than as a
-       silent privesc.
+    2. **Identity-only allowlist.** ``oauth2_config_mappings`` maps
+       header names to ``UserAPIKeyAuth`` fields. Without an allowlist,
+       an admin who maps the wrong header to ``user_role`` lets any
+       caller send ``X-User-Role: proxy_admin`` and gain full admin
+       privileges (Pydantic coerces the string into the enum). Only
+       fields in ``ALLOWED_OAUTH2_PROXY_FIELDS`` (identity assertion
+       only — see the constant's comment) may be mapped; any other
+       mapping is rejected at request time so the misconfiguration
+       surfaces loudly rather than as a silent privesc.
     """
     from litellm.proxy.proxy_server import general_settings, premium_user
 
@@ -81,17 +77,18 @@ async def handle_oauth2_proxy_request(request: Request) -> UserAPIKeyAuth:
     if not oauth2_config_mappings:
         raise ValueError("Oauth2 config mappings not found in general_settings")
 
-    privileged_mapped = sorted(
-        set(oauth2_config_mappings.keys()) & PRIVILEGED_OAUTH2_PROXY_FIELDS
+    disallowed = sorted(
+        set(oauth2_config_mappings.keys()) - ALLOWED_OAUTH2_PROXY_FIELDS
     )
-    if privileged_mapped:
+    if disallowed:
         raise ValueError(
-            "Oauth2 proxy auth refuses to map privileged UserAPIKeyAuth "
-            f"fields from request headers: {privileged_mapped}. These "
-            "fields would grant privileges (e.g. proxy_admin), bypass "
-            "budget enforcement, or short-circuit key validation if a "
-            "caller can spoof the corresponding header. If you need a "
-            "trusted upstream to assert one of these, use JWT auth "
+            "Oauth2 proxy auth refuses to map non-identity UserAPIKeyAuth "
+            f"fields from request headers: {disallowed}. Only identity "
+            f"fields are accepted ({sorted(ALLOWED_OAUTH2_PROXY_FIELDS)}); "
+            "anything else (privileges, budgets, rate limits, metadata) "
+            "would let a caller forge enforcement parameters by spoofing "
+            "the matching header. If you need a trusted upstream to "
+            "assert anything beyond identity, use JWT auth "
             "(signature-validated) instead of header-trust."
         )
 
diff --git a/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py b/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py
index e51882a3b1..e73ac571d5 100644
--- a/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py
+++ b/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py
@@ -29,7 +29,7 @@ sys.path.insert(0, os.path.abspath("../../../.."))
 
 from litellm.proxy._types import LitellmUserRoles
 from litellm.proxy.auth.oauth2_proxy_hook import (
-    PRIVILEGED_OAUTH2_PROXY_FIELDS,
+    ALLOWED_OAUTH2_PROXY_FIELDS,
     handle_oauth2_proxy_request,
 )
 
@@ -90,13 +90,46 @@ async def test_rejects_when_not_premium(configure_proxy):
 
 @pytest.mark.parametrize(
     "privileged_field",
-    sorted(PRIVILEGED_OAUTH2_PROXY_FIELDS),
+    [
+        # The GHSA-5c3m-qffq-4r9m primary privesc field.
+        "user_role",
+        # Key-level enforcement bypass shapes.
+        "api_key",
+        "token",
+        "permissions",
+        "allowed_routes",
+        "max_budget",
+        "spend",
+        "tpm_limit",
+        "rpm_limit",
+        "model_max_budget",
+        "metadata",
+        # User-level enforcement bypass — flagged by Greptile as a denylist gap.
+        "user_max_budget",
+        "user_tpm_limit",
+        "user_rpm_limit",
+        "user_spend",
+        # Team / org / end-user / region — same class, all denied by the
+        # identity-only allowlist.
+        "team_max_budget",
+        "team_spend",
+        "team_member_tpm_limit",
+        "organization_max_budget",
+        "organization_tpm_limit",
+        "end_user_max_budget",
+        "allowed_model_region",
+        # Anything not on ALLOWED_OAUTH2_PROXY_FIELDS is blocked, even
+        # fabricated field names admins might try.
+        "definitely_not_a_real_field",
+    ],
 )
 @pytest.mark.asyncio
-async def test_refuses_to_map_privileged_fields(configure_proxy, privileged_field):
+async def test_refuses_to_map_non_identity_fields(configure_proxy, privileged_field):
     # GHSA-5c3m-qffq-4r9m attack shape: admin maps a privileged field
-    # to a header and a caller forges the value. The hook must reject
-    # the misconfiguration outright at request time.
+    # to a header and a caller forges the value. The allowlist rejects
+    # any non-identity mapping at request time, regardless of whether
+    # the field ever appeared on a denylist — which is the whole reason
+    # we use an allowlist instead.
     configure_proxy(mappings={privileged_field: f"x-{privileged_field}"})
     request = _request_with_headers({f"x-{privileged_field}": "proxy_admin"})
 
@@ -105,6 +138,22 @@ async def test_refuses_to_map_privileged_fields(configure_proxy, privileged_fiel
     assert privileged_field in str(exc.value)
 
 
+@pytest.mark.parametrize("identity_field", sorted(ALLOWED_OAUTH2_PROXY_FIELDS))
+def test_allowlist_is_identity_only(identity_field):
+    # Lock in the allowlist's intent: only identity-assertion fields are
+    # safe to populate from a header. If anyone proposes adding budget /
+    # spend / role / permission to ``ALLOWED_OAUTH2_PROXY_FIELDS``, this
+    # assertion forces them to update the test deliberately.
+    assert identity_field in {
+        "user_id",
+        "user_email",
+        "team_id",
+        "team_alias",
+        "org_id",
+        "models",
+    }
+
+
 @pytest.mark.asyncio
 async def test_user_role_header_forgery_attack_is_blocked(configure_proxy):
     # End-to-end form of the privesc: with ``user_role`` mapped, the

From fbcfd59b1a23edc17d6a93880726f26e308efedd Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Wed, 29 Apr 2026 23:04:05 +0000
Subject: [PATCH 4/6] fix(oauth2-proxy): drop premium gate; identity-only
 allowlist is the security fix
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Greptile flagged the ``premium_user is not True`` check as a hard
backwards-incompatible break for OSS users currently running
``enable_oauth2_proxy_auth=True``. They were right: unlike the
api_base case (where the docs already required admin opt-in), this
path was documented as available to OSS users. Adding the gate would
have closed a documented feature, not fixed a vuln.

Reframed the change:

* The **identity-only allowlist** (``ALLOWED_OAUTH2_PROXY_FIELDS`` =
  ``{user_id, user_email, team_id, team_alias, org_id, models}``) is
  the actual security fix — it closes the privesc by rejecting any
  mapping to a non-identity field at request time. This is unchanged.
* The **premium gate** was parity-with-siblings (a product decision,
  not a security one). Removed. BerriAI can re-add it on their own
  schedule with a proper deprecation cycle if they want enterprise-
  only gating.

Tests: removed ``test_rejects_when_not_premium``; everything else
(allowlist enforcement, identity passthrough, attack-shape
regression) still passes — 14 tests.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 litellm/proxy/auth/oauth2_proxy_hook.py       | 36 +++++++------------
 .../proxy/auth/test_oauth2_proxy_hook.py      | 19 +++-------
 2 files changed, 16 insertions(+), 39 deletions(-)

diff --git a/litellm/proxy/auth/oauth2_proxy_hook.py b/litellm/proxy/auth/oauth2_proxy_hook.py
index 1ba1b100a8..389a5b2b9e 100644
--- a/litellm/proxy/auth/oauth2_proxy_hook.py
+++ b/litellm/proxy/auth/oauth2_proxy_hook.py
@@ -3,7 +3,7 @@ from typing import Any, Dict, FrozenSet
 from fastapi import Request
 
 from litellm._logging import verbose_proxy_logger
-from litellm.proxy._types import CommonProxyErrors, UserAPIKeyAuth
+from litellm.proxy._types import UserAPIKeyAuth
 
 # OAuth2-proxy header trust is for **identity assertion** from a trusted
 # upstream auth proxy (oauth2-proxy, Authelia, etc.). The allowlist below
@@ -42,31 +42,19 @@ async def handle_oauth2_proxy_request(request: Request) -> UserAPIKeyAuth:
 
     The auth model assumes the proxy is deployed behind a trusted OAuth2
     reverse proxy that injects authenticated identity headers (e.g.
-    oauth2-proxy, Authelia). Two safeguards above and beyond that
-    deployment assumption:
+    oauth2-proxy, Authelia).
 
-    1. **Premium gate.** The sibling auth paths (``enable_oauth2_auth``
-       and ``enable_jwt_auth``) require ``premium_user``; this path
-       previously did not, which let any open-source deployment turn
-       the feature on without realising it requires a hardened
-       deployment topology.
-    2. **Identity-only allowlist.** ``oauth2_config_mappings`` maps
-       header names to ``UserAPIKeyAuth`` fields. Without an allowlist,
-       an admin who maps the wrong header to ``user_role`` lets any
-       caller send ``X-User-Role: proxy_admin`` and gain full admin
-       privileges (Pydantic coerces the string into the enum). Only
-       fields in ``ALLOWED_OAUTH2_PROXY_FIELDS`` (identity assertion
-       only — see the constant's comment) may be mapped; any other
-       mapping is rejected at request time so the misconfiguration
-       surfaces loudly rather than as a silent privesc.
+    **Identity-only allowlist.** ``oauth2_config_mappings`` maps header
+    names to ``UserAPIKeyAuth`` fields. Without an allowlist, an admin
+    who maps the wrong header to ``user_role`` lets any caller send
+    ``X-User-Role: proxy_admin`` and gain full admin privileges
+    (Pydantic coerces the string into the enum). Only fields in
+    ``ALLOWED_OAUTH2_PROXY_FIELDS`` (identity assertion only — see the
+    constant's comment) may be mapped; any other mapping is rejected at
+    request time so the misconfiguration surfaces loudly rather than as
+    a silent privesc.
     """
-    from litellm.proxy.proxy_server import general_settings, premium_user
-
-    if premium_user is not True:
-        raise ValueError(
-            "Oauth2 proxy auth is an enterprise-only feature. "
-            + CommonProxyErrors.not_premium_user.value
-        )
+    from litellm.proxy.proxy_server import general_settings
 
     verbose_proxy_logger.debug("Handling oauth2 proxy request")
     oauth2_config_mappings: Dict[str, str] = (
diff --git a/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py b/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py
index e73ac571d5..42af9e6f03 100644
--- a/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py
+++ b/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py
@@ -47,17 +47,15 @@ def _request_with_headers(headers: dict) -> Request:
 @pytest.fixture
 def configure_proxy(monkeypatch):
     """
-    Yields a callable that sets ``premium_user`` and
-    ``oauth2_config_mappings`` on the proxy_server module for the
-    duration of one test. Default is premium=True with a single
-    ``user_id -> x-user-id`` mapping.
+    Yields a callable that sets ``oauth2_config_mappings`` on the
+    proxy_server module for the duration of one test. Default mapping
+    is a single ``user_id -> x-user-id`` (identity-only).
     """
     import litellm.proxy.proxy_server as proxy_server
 
-    def _configure(*, premium=True, mappings=None):
+    def _configure(*, mappings=None):
         if mappings is None:
             mappings = {"user_id": "x-user-id"}
-        monkeypatch.setattr(proxy_server, "premium_user", premium, raising=False)
         monkeypatch.setattr(
             proxy_server,
             "general_settings",
@@ -79,15 +77,6 @@ async def test_returns_auth_for_simple_user_id_mapping(configure_proxy):
     assert auth.user_role is None
 
 
-@pytest.mark.asyncio
-async def test_rejects_when_not_premium(configure_proxy):
-    configure_proxy(premium=False)
-    request = _request_with_headers({"x-user-id": "alice"})
-
-    with pytest.raises(ValueError, match="enterprise"):
-        await handle_oauth2_proxy_request(request)
-
-
 @pytest.mark.parametrize(
     "privileged_field",
     [

From 722bc63e37d5a3773f95bb6c22b11f74bb3cb65e Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Wed, 29 Apr 2026 23:47:29 +0000
Subject: [PATCH 5/6] chore(oauth2-proxy): drop unused patch import + tighten
 docstring

Greptile flagged the unused ``from unittest.mock import patch``
left over from before the ``configure_proxy`` fixture refactor (the
fixture uses ``monkeypatch``, no ``patch`` calls remain). Also pruned
the now-stale "premium gate" paragraph from the module docstring
since that gate was removed in fbcfd59b1a.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 .../proxy/auth/test_oauth2_proxy_hook.py      | 19 ++++++-------------
 1 file changed, 6 insertions(+), 13 deletions(-)

diff --git a/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py b/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py
index 42af9e6f03..9d0bdcf351 100644
--- a/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py
+++ b/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py
@@ -3,23 +3,16 @@ Regression tests for the OAuth2-proxy header-forgery fix
 (GHSA-5c3m-qffq-4r9m).
 
 The hook reads HTTP request headers per ``oauth2_config_mappings`` and
-constructs a ``UserAPIKeyAuth`` from them. Two separate failure modes
-the fix closes:
-
-1. The path was not gated on ``premium_user`` (the sibling
-   ``enable_oauth2_auth`` and ``enable_jwt_auth`` paths are). Open-source
-   deployments could enable the feature without realising it requires
-   a hardened deployment topology.
-2. Any ``UserAPIKeyAuth`` field could be mapped from a header — including
-   ``user_role``, which Pydantic coerces from the string ``"proxy_admin"``
-   into ``LitellmUserRoles.PROXY_ADMIN``. An attacker who reaches the
-   proxy directly (or via a misconfigured reverse proxy) sets the mapped
-   header and gains full admin privileges.
+constructs a ``UserAPIKeyAuth`` from them. Without the
+identity-only allowlist any field could be mapped — including
+``user_role``, which Pydantic coerces from the string
+``"proxy_admin"`` into ``LitellmUserRoles.PROXY_ADMIN``. An attacker
+who reaches the proxy directly (or via a misconfigured reverse
+proxy) sets the mapped header and gains full admin privileges.
 """
 
 import os
 import sys
-from unittest.mock import patch
 
 import pytest
 from fastapi import Request

From 2f4641752bd9f8bf325c43c1dc6cd5d85e322bcb Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Wed, 29 Apr 2026 21:20:21 -0700
Subject: [PATCH 6/6] chore(auth): require trusted proxy for header identity
 auth

---
 .../proxy/auth/custom_sso_handler.py          |  54 ++++---
 litellm/integrations/custom_sso_handler.py    |  11 ++
 litellm/proxy/_types.py                       |   4 +
 litellm/proxy/auth/oauth2_proxy_hook.py       |   7 +
 litellm/proxy/auth/trusted_proxy_utils.py     | 118 +++++++++++++++
 .../proxy/auth/test_oauth2_proxy_hook.py      |  50 +++++--
 .../proxy/management_endpoints/test_ui_sso.py | 141 +++++++++++-------
 7 files changed, 300 insertions(+), 85 deletions(-)
 create mode 100644 litellm/proxy/auth/trusted_proxy_utils.py

diff --git a/enterprise/litellm_enterprise/proxy/auth/custom_sso_handler.py b/enterprise/litellm_enterprise/proxy/auth/custom_sso_handler.py
index a368232038..e8f104c262 100644
--- a/enterprise/litellm_enterprise/proxy/auth/custom_sso_handler.py
+++ b/enterprise/litellm_enterprise/proxy/auth/custom_sso_handler.py
@@ -10,28 +10,21 @@ has already authenticated the user) and you need to extract user information fro
 custom headers or other request attributes.
 """
 
-from typing import TYPE_CHECKING, Dict, Optional, Union, cast
+from typing import cast
 
 from fastapi import Request
 from fastapi.responses import RedirectResponse
 
-if TYPE_CHECKING:
-    from fastapi_sso.sso.base import OpenID
-else:
-    from typing import Any as OpenID
-
-from litellm.proxy.management_endpoints.types import CustomOpenID
-
 
 class EnterpriseCustomSSOHandler:
     """
     Enterprise Custom SSO Handler for LiteLLM Proxy
-    
+
     This class provides methods for handling custom SSO authentication flows
     where users can implement their own authentication logic by processing
     request headers and returning user information in OpenID format.
     """
-    
+
     @staticmethod
     async def handle_custom_ui_sso_sign_in(
         request: Request,
@@ -40,16 +33,16 @@ class EnterpriseCustomSSOHandler:
         Allow a user to execute their custom code to parse incoming request headers and return a OpenID object
 
         Use this when you have an OAuth proxy in front of LiteLLM (where the OAuth proxy has already authenticated the user)
-        
+
         Args:
             request: The FastAPI request object containing headers and other request data
-            
+
         Returns:
             RedirectResponse: Redirect response that sends the user to the LiteLLM UI with authentication token
-            
+
         Raises:
             ValueError: If custom_ui_sso_sign_in_handler is not configured
-            
+
         Example:
             This method is typically called when a user has already been authenticated by an
             external OAuth proxy and the proxy has added custom headers containing user information.
@@ -60,27 +53,44 @@ class EnterpriseCustomSSOHandler:
         from litellm.integrations.custom_sso_handler import CustomSSOLoginHandler
         from litellm.proxy.proxy_server import (
             CommonProxyErrors,
+            general_settings,
             premium_user,
             user_custom_ui_sso_sign_in_handler,
         )
+        from litellm.proxy.auth.trusted_proxy_utils import (
+            require_trusted_proxy_request,
+        )
+
         if premium_user is not True:
             raise ValueError(CommonProxyErrors.not_premium_user.value)
-        
+
         if user_custom_ui_sso_sign_in_handler is None:
-            raise ValueError("custom_ui_sso_sign_in_handler is not configured. Please set it in general_settings.")
-        
-        custom_sso_login_handler = cast(CustomSSOLoginHandler, user_custom_ui_sso_sign_in_handler)
-        openid_response: OpenID = await custom_sso_login_handler.handle_custom_ui_sso_sign_in(
+            raise ValueError(
+                "custom_ui_sso_sign_in_handler is not configured. Please set it in general_settings."
+            )
+
+        require_trusted_proxy_request(
             request=request,
+            general_settings=general_settings,
+            feature_name="Custom UI SSO",
         )
-        
+
+        custom_sso_login_handler = cast(
+            CustomSSOLoginHandler, user_custom_ui_sso_sign_in_handler
+        )
+        openid_response: OpenID = (
+            await custom_sso_login_handler.handle_custom_ui_sso_sign_in(
+                request=request,
+            )
+        )
+
         # Import here to avoid circular imports
         from litellm.proxy.management_endpoints.ui_sso import SSOAuthenticationHandler
-        
+
         return await SSOAuthenticationHandler.get_redirect_response_from_openid(
             result=openid_response,
             request=request,
             received_response=None,
             generic_client_id=None,
             ui_access_mode=None,
-        ) 
\ No newline at end of file
+        )
diff --git a/litellm/integrations/custom_sso_handler.py b/litellm/integrations/custom_sso_handler.py
index 7f60decabc..202e488e0e 100644
--- a/litellm/integrations/custom_sso_handler.py
+++ b/litellm/integrations/custom_sso_handler.py
@@ -18,6 +18,17 @@ class CustomSSOLoginHandler(CustomLogger):
         self,
         request: Request,
     ) -> OpenID:
+        from litellm.proxy.auth.trusted_proxy_utils import (
+            require_trusted_proxy_request,
+        )
+        from litellm.proxy.proxy_server import general_settings
+
+        require_trusted_proxy_request(
+            request=request,
+            general_settings=general_settings,
+            feature_name="Custom UI SSO",
+        )
+
         request_headers_dict = dict(request.headers)
         return OpenID(
             id=request_headers_dict.get("x-litellm-user-id"),
diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py
index 92c920ca59..5165c7fd50 100644
--- a/litellm/proxy/_types.py
+++ b/litellm/proxy/_types.py
@@ -2374,6 +2374,10 @@ class ConfigGeneralSettings(LiteLLMPydanticObjectBase):
         None,
         description="CIDR ranges of trusted reverse proxies. When set, X-Forwarded-For headers are only trusted from these IPs.",
     )
+    trusted_proxy_ranges: Optional[List[str]] = Field(
+        None,
+        description="CIDR ranges of trusted reverse proxies allowed to provide identity headers for header-based auth paths such as enable_oauth2_proxy_auth and custom_ui_sso_sign_in_handler.",
+    )
     store_model_in_db: Optional[bool] = Field(
         None,
         description="If True, models and config are stored in and loaded from the database. Default is False.",
diff --git a/litellm/proxy/auth/oauth2_proxy_hook.py b/litellm/proxy/auth/oauth2_proxy_hook.py
index 389a5b2b9e..9fc4c4fb53 100644
--- a/litellm/proxy/auth/oauth2_proxy_hook.py
+++ b/litellm/proxy/auth/oauth2_proxy_hook.py
@@ -4,6 +4,7 @@ from fastapi import Request
 
 from litellm._logging import verbose_proxy_logger
 from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy.auth.trusted_proxy_utils import require_trusted_proxy_request
 
 # OAuth2-proxy header trust is for **identity assertion** from a trusted
 # upstream auth proxy (oauth2-proxy, Authelia, etc.). The allowlist below
@@ -57,6 +58,12 @@ async def handle_oauth2_proxy_request(request: Request) -> UserAPIKeyAuth:
     from litellm.proxy.proxy_server import general_settings
 
     verbose_proxy_logger.debug("Handling oauth2 proxy request")
+    require_trusted_proxy_request(
+        request=request,
+        general_settings=general_settings,
+        feature_name="OAuth2 proxy auth",
+    )
+
     oauth2_config_mappings: Dict[str, str] = (
         general_settings.get("oauth2_config_mappings") or {}
     )
diff --git a/litellm/proxy/auth/trusted_proxy_utils.py b/litellm/proxy/auth/trusted_proxy_utils.py
new file mode 100644
index 0000000000..df7b3080f2
--- /dev/null
+++ b/litellm/proxy/auth/trusted_proxy_utils.py
@@ -0,0 +1,118 @@
+import ipaddress
+from typing import Any, Dict, List, Optional, Union
+
+from fastapi import Request
+
+from litellm._logging import verbose_proxy_logger
+
+TRUSTED_PROXY_RANGES_KEY = "trusted_proxy_ranges"
+TrustedProxyNetwork = Union[ipaddress.IPv4Network, ipaddress.IPv6Network]
+
+
+def _get_proxy_general_settings() -> Dict[str, Any]:
+    try:
+        from litellm.proxy.proxy_server import general_settings
+
+        return general_settings or {}
+    except ImportError:
+        return {}
+
+
+def _normalize_cidr_ranges(configured_ranges: Any, *, setting_name: str) -> List[str]:
+    if not configured_ranges:
+        return []
+    if isinstance(configured_ranges, str):
+        return [
+            raw_range.strip()
+            for raw_range in configured_ranges.split(",")
+            if raw_range.strip()
+        ]
+    if isinstance(configured_ranges, (list, tuple, set)):
+        return [
+            str(raw_range).strip()
+            for raw_range in configured_ranges
+            if str(raw_range).strip()
+        ]
+    verbose_proxy_logger.warning(
+        "Invalid %s value: expected a list of CIDR ranges, got %s",
+        setting_name,
+        type(configured_ranges).__name__,
+    )
+    return []
+
+
+def parse_trusted_proxy_ranges(
+    configured_ranges: Any,
+    *,
+    setting_name: str = TRUSTED_PROXY_RANGES_KEY,
+) -> List[TrustedProxyNetwork]:
+    networks: List[TrustedProxyNetwork] = []
+    for cidr in _normalize_cidr_ranges(configured_ranges, setting_name=setting_name):
+        try:
+            networks.append(ipaddress.ip_network(cidr, strict=False))
+        except ValueError:
+            verbose_proxy_logger.warning(
+                "Invalid CIDR in %s: %s, skipping", setting_name, cidr
+            )
+    return networks
+
+
+def _get_direct_client_ip(request: Request) -> Optional[str]:
+    client = getattr(request, "client", None)
+    client_host = getattr(client, "host", None)
+    if isinstance(client_host, str):
+        return client_host
+    return None
+
+
+def _is_ip_in_networks(
+    client_ip: Optional[str], networks: List[TrustedProxyNetwork]
+) -> bool:
+    if not client_ip or not networks:
+        return False
+    try:
+        addr = ipaddress.ip_address(client_ip.strip())
+    except ValueError:
+        return False
+    return any(addr in network for network in networks)
+
+
+def require_trusted_proxy_request(
+    *,
+    request: Request,
+    general_settings: Optional[Dict[str, Any]] = None,
+    feature_name: str,
+    setting_name: str = TRUSTED_PROXY_RANGES_KEY,
+) -> None:
+    """
+    Fail closed unless the direct TCP peer is one of the configured
+    trusted reverse proxies.
+
+    Header-based auth paths must validate the direct peer, not
+    X-Forwarded-For, because the direct peer is the actor supplying the
+    identity headers.
+    """
+    if general_settings is None:
+        general_settings = _get_proxy_general_settings()
+
+    trusted_networks = parse_trusted_proxy_ranges(
+        general_settings.get(setting_name), setting_name=setting_name
+    )
+    if not trusted_networks:
+        raise ValueError(
+            f"{feature_name} requires general_settings.{setting_name} before "
+            "trusting identity headers from an upstream proxy."
+        )
+
+    direct_client_ip = _get_direct_client_ip(request)
+    if not _is_ip_in_networks(direct_client_ip, trusted_networks):
+        verbose_proxy_logger.warning(
+            "%s rejected identity headers from untrusted direct client IP %r",
+            feature_name,
+            direct_client_ip,
+        )
+        raise ValueError(
+            f"{feature_name} only accepts identity headers from configured "
+            f"trusted proxy ranges. Direct client IP {direct_client_ip!r} "
+            "is not trusted."
+        )
diff --git a/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py b/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py
index 9d0bdcf351..dcbfd281e0 100644
--- a/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py
+++ b/tests/test_litellm/proxy/auth/test_oauth2_proxy_hook.py
@@ -3,12 +3,14 @@ Regression tests for the OAuth2-proxy header-forgery fix
 (GHSA-5c3m-qffq-4r9m).
 
 The hook reads HTTP request headers per ``oauth2_config_mappings`` and
-constructs a ``UserAPIKeyAuth`` from them. Without the
-identity-only allowlist any field could be mapped — including
-``user_role``, which Pydantic coerces from the string
-``"proxy_admin"`` into ``LitellmUserRoles.PROXY_ADMIN``. An attacker
-who reaches the proxy directly (or via a misconfigured reverse
-proxy) sets the mapped header and gains full admin privileges.
+constructs a ``UserAPIKeyAuth`` from them. The fix has two parts:
+
+1. Only requests from configured trusted proxy CIDR ranges may provide
+   identity headers.
+2. Only identity fields may be mapped from those headers. Without the
+   identity-only allowlist any field could be mapped — including
+   ``user_role``, which Pydantic coerces from the string
+   ``"proxy_admin"`` into ``LitellmUserRoles.PROXY_ADMIN``.
 """
 
 import os
@@ -27,9 +29,10 @@ from litellm.proxy.auth.oauth2_proxy_hook import (
 )
 
 
-def _request_with_headers(headers: dict) -> Request:
+def _request_with_headers(headers: dict, *, client_host: str = "127.0.0.1") -> Request:
     scope = {
         "type": "http",
+        "client": (client_host, 12345),
         "headers": [(k.lower().encode(), v.encode()) for k, v in headers.items()],
     }
     request = Request(scope=scope)
@@ -40,19 +43,24 @@ def _request_with_headers(headers: dict) -> Request:
 @pytest.fixture
 def configure_proxy(monkeypatch):
     """
-    Yields a callable that sets ``oauth2_config_mappings`` on the
-    proxy_server module for the duration of one test. Default mapping
-    is a single ``user_id -> x-user-id`` (identity-only).
+    Yields a callable that sets ``oauth2_config_mappings`` and
+    ``trusted_proxy_ranges`` on the proxy_server module for the duration
+    of one test. Defaults to a single identity mapping and localhost as
+    a trusted proxy.
     """
     import litellm.proxy.proxy_server as proxy_server
 
-    def _configure(*, mappings=None):
+    def _configure(*, mappings=None, trusted_proxy_ranges=("127.0.0.1/32",)):
         if mappings is None:
             mappings = {"user_id": "x-user-id"}
+        settings = {
+            "oauth2_config_mappings": mappings,
+            "trusted_proxy_ranges": trusted_proxy_ranges,
+        }
         monkeypatch.setattr(
             proxy_server,
             "general_settings",
-            {"oauth2_config_mappings": mappings},
+            settings,
             raising=False,
         )
 
@@ -70,6 +78,24 @@ async def test_returns_auth_for_simple_user_id_mapping(configure_proxy):
     assert auth.user_role is None
 
 
+@pytest.mark.asyncio
+async def test_rejects_identity_headers_without_trusted_proxy_ranges(configure_proxy):
+    configure_proxy(trusted_proxy_ranges=None)
+    request = _request_with_headers({"x-user-id": "alice"})
+
+    with pytest.raises(ValueError, match="trusted_proxy_ranges"):
+        await handle_oauth2_proxy_request(request)
+
+
+@pytest.mark.asyncio
+async def test_rejects_identity_headers_from_untrusted_direct_client(configure_proxy):
+    configure_proxy(trusted_proxy_ranges=["10.0.0.0/24"])
+    request = _request_with_headers({"x-user-id": "alice"}, client_host="203.0.113.10")
+
+    with pytest.raises(ValueError, match="not trusted"):
+        await handle_oauth2_proxy_request(request)
+
+
 @pytest.mark.parametrize(
     "privileged_field",
     [
diff --git a/tests/test_litellm/proxy/management_endpoints/test_ui_sso.py b/tests/test_litellm/proxy/management_endpoints/test_ui_sso.py
index eecfcaa035..92759c56cb 100644
--- a/tests/test_litellm/proxy/management_endpoints/test_ui_sso.py
+++ b/tests/test_litellm/proxy/management_endpoints/test_ui_sso.py
@@ -4,7 +4,6 @@ import os
 import sys
 from unittest.mock import AsyncMock, MagicMock, patch
 
-import httpx
 import pytest
 from fastapi import HTTPException, Request
 
@@ -25,7 +24,6 @@ from litellm.proxy.management_endpoints.ui_sso import (
     SSOAuthenticationHandler,
     _setup_team_mappings,
     _sync_user_role_from_jwt_role_map,
-    determine_role_from_groups,
     normalize_email,
     process_sso_jwt_access_token,
 )
@@ -1849,6 +1847,7 @@ class TestCustomUISSO:
             "x-forwarded-for": "192.168.1.1",
         }
         mock_request.base_url = "https://test.litellm.ai/"
+        mock_request.client.host = "10.0.0.10"
 
         # Mock the custom handler
         mock_custom_handler = MagicMock(spec=CustomSSOLoginHandler)
@@ -1874,36 +1873,73 @@ class TestCustomUISSO:
                 "litellm.proxy.proxy_server.user_custom_ui_sso_sign_in_handler",
                 mock_custom_handler,
             ):
-                with patch.object(
-                    SSOAuthenticationHandler,
-                    "get_redirect_response_from_openid",
-                    return_value=mock_redirect_response,
-                ) as mock_get_redirect:
-                    # Act
-                    result = (
+                with patch(
+                    "litellm.proxy.proxy_server.general_settings",
+                    {"trusted_proxy_ranges": ["10.0.0.0/24"]},
+                ):
+                    with patch.object(
+                        SSOAuthenticationHandler,
+                        "get_redirect_response_from_openid",
+                        return_value=mock_redirect_response,
+                    ) as mock_get_redirect:
+                        # Act
+                        result = await EnterpriseCustomSSOHandler.handle_custom_ui_sso_sign_in(
+                            request=mock_request
+                        )
+
+                        # Assert
+                        # Verify the custom handler was called with the request
+                        mock_custom_handler.handle_custom_ui_sso_sign_in.assert_called_once_with(
+                            request=mock_request
+                        )
+
+                        # Verify the redirect response was generated with correct OpenID
+                        mock_get_redirect.assert_called_once_with(
+                            result=expected_openid,
+                            request=mock_request,
+                            received_response=None,
+                            generic_client_id=None,
+                            ui_access_mode=None,
+                        )
+
+                        # Verify the result is the redirect response
+                        assert result == mock_redirect_response
+                        assert result.status_code == 303
+
+    @pytest.mark.asyncio
+    async def test_handle_custom_ui_sso_sign_in_rejects_untrusted_proxy(self):
+        """Custom UI SSO rejects spoofed identity headers from direct clients."""
+        from enterprise.litellm_enterprise.proxy.auth.custom_sso_handler import (
+            EnterpriseCustomSSOHandler,
+        )
+        from litellm.integrations.custom_sso_handler import CustomSSOLoginHandler
+
+        mock_request = MagicMock(spec=Request)
+        mock_request.headers = {
+            "x-litellm-user-id": "admin",
+            "x-litellm-user-email": "admin@example.com",
+        }
+        mock_request.base_url = "https://test.litellm.ai/"
+        mock_request.client.host = "203.0.113.10"
+
+        mock_custom_handler = MagicMock(spec=CustomSSOLoginHandler)
+        mock_custom_handler.handle_custom_ui_sso_sign_in = AsyncMock()
+
+        with patch("litellm.proxy.proxy_server.premium_user", True):
+            with patch(
+                "litellm.proxy.proxy_server.user_custom_ui_sso_sign_in_handler",
+                mock_custom_handler,
+            ):
+                with patch(
+                    "litellm.proxy.proxy_server.general_settings",
+                    {"trusted_proxy_ranges": ["10.0.0.0/24"]},
+                ):
+                    with pytest.raises(ValueError, match="not trusted"):
                         await EnterpriseCustomSSOHandler.handle_custom_ui_sso_sign_in(
                             request=mock_request
                         )
-                    )
 
-                    # Assert
-                    # Verify the custom handler was called with the request
-                    mock_custom_handler.handle_custom_ui_sso_sign_in.assert_called_once_with(
-                        request=mock_request
-                    )
-
-                    # Verify the redirect response was generated with correct OpenID
-                    mock_get_redirect.assert_called_once_with(
-                        result=expected_openid,
-                        request=mock_request,
-                        received_response=None,
-                        generic_client_id=None,
-                        ui_access_mode=None,
-                    )
-
-                    # Verify the result is the redirect response
-                    assert result == mock_redirect_response
-                    assert result.status_code == 303
+        mock_custom_handler.handle_custom_ui_sso_sign_in.assert_not_called()
 
     @pytest.mark.asyncio
     async def test_custom_ui_sso_handler_execution_with_real_class(self):
@@ -1954,6 +1990,7 @@ class TestCustomUISSO:
             "x-forwarded-for": "10.0.0.1",
         }
         mock_request.base_url = "https://custom.litellm.ai/"
+        mock_request.client.host = "10.0.0.20"
 
         # Mock the redirect response method
         mock_redirect_response = MagicMock()
@@ -1964,34 +2001,36 @@ class TestCustomUISSO:
                 "litellm.proxy.proxy_server.user_custom_ui_sso_sign_in_handler",
                 test_handler_instance,
             ):
-                with patch.object(
-                    SSOAuthenticationHandler,
-                    "get_redirect_response_from_openid",
-                    return_value=mock_redirect_response,
-                ) as mock_get_redirect:
-                    # Act
-                    result = (
-                        await EnterpriseCustomSSOHandler.handle_custom_ui_sso_sign_in(
+                with patch(
+                    "litellm.proxy.proxy_server.general_settings",
+                    {"trusted_proxy_ranges": ["10.0.0.0/24"]},
+                ):
+                    with patch.object(
+                        SSOAuthenticationHandler,
+                        "get_redirect_response_from_openid",
+                        return_value=mock_redirect_response,
+                    ) as mock_get_redirect:
+                        # Act
+                        result = await EnterpriseCustomSSOHandler.handle_custom_ui_sso_sign_in(
                             request=mock_request
                         )
-                    )
 
-                    # Assert that our custom handler was executed
-                    assert test_handler_instance.method_called is True
-                    assert test_handler_instance.received_request == mock_request
+                        # Assert that our custom handler was executed
+                        assert test_handler_instance.method_called is True
+                        assert test_handler_instance.received_request == mock_request
 
-                    # Verify the redirect response was called with the OpenID from our custom handler
-                    mock_get_redirect.assert_called_once()
-                    call_args = mock_get_redirect.call_args.kwargs
+                        # Verify the redirect response was called with the OpenID from our custom handler
+                        mock_get_redirect.assert_called_once()
+                        call_args = mock_get_redirect.call_args.kwargs
 
-                    # Verify the OpenID object has the expected values from our custom handler
-                    openid_result = call_args["result"]
-                    assert openid_result.id == "custom_test_user_456"
-                    assert openid_result.email == "custom@example.com"
-                    assert openid_result.first_name == "Custom"
-                    assert openid_result.last_name == "Handler"
-                    assert openid_result.display_name == "Custom Handler Test"
-                    assert openid_result.provider == "custom"
+                        # Verify the OpenID object has the expected values from our custom handler
+                        openid_result = call_args["result"]
+                        assert openid_result.id == "custom_test_user_456"
+                        assert openid_result.email == "custom@example.com"
+                        assert openid_result.first_name == "Custom"
+                        assert openid_result.last_name == "Handler"
+                        assert openid_result.display_name == "Custom Handler Test"
+                        assert openid_result.provider == "custom"
 
                     # Verify the request and other parameters were passed correctly
                     assert call_args["request"] == mock_request