feat(proxy): auth_v2 slice 5 - data-plane model access on the inference path

Brings inference under casbin too, via the hybrid design: control-plane access
stays RBAC policy rows, data-plane access (can this principal call this model)
is a casbin ABAC matcher over an attribute carried on the already-loaded key, so
the hot path reads no policy store and writes no per-key policy rows.

Inference routes (chat/completions, completions, embeddings, responses) now
authenticate and check the requested model against the principal's allowed-model
attribute. Semantics match v1: an empty models list, "*", "all-proxy-models" and
"all-team-models" are unrestricted; otherwise the model must be in the list.

Tests cover the allow/deny/unrestricted matrix and inference-route detection.

litellm/proxy/auth/v2/data_plane.conf

@@ -0,0 +1,11 @@
+[request_definition]
+r = sub, obj
+
+[policy_definition]
+p = eft
+
+[policy_effect]
+e = some(where (p.eft == allow))
+
+[matchers]
+m = r.sub.unrestricted || r.obj in r.sub.allowed_models

litellm/proxy/auth/v2/data_plane.py

@@ -0,0 +1,47 @@
+import os
+from dataclasses import dataclass
+from typing import List, Optional
+
+import casbin
+
+_MODEL_PATH = os.path.join(os.path.dirname(__file__), "data_plane.conf")
+
+# Sentinels that mean "any model" in the existing key/team model lists.
+_UNRESTRICTED_SENTINELS = {"*", "all-proxy-models", "all-team-models"}
+
+
+@dataclass
+class ModelAccessSubject:
+    """Carries the principal's model entitlement as a casbin ABAC attribute.
+
+    The data plane runs on the inference hot path, so access is decided from
+    attributes already on the loaded key/team (no per-key policy rows, no policy
+    store read). An empty list means unrestricted, matching existing key
+    semantics where ``models == []`` allows every model.
+    """
+
+    allowed_models: List[str]
+
+    @property
+    def unrestricted(self) -> bool:
+        if not self.allowed_models:
+            return True
+        return any(model in _UNRESTRICTED_SENTINELS for model in self.allowed_models)
+
+
+_enforcer: Optional[casbin.Enforcer] = None
+
+
+def _get_enforcer() -> casbin.Enforcer:
+    global _enforcer
+    if _enforcer is None:
+        enforcer = casbin.Enforcer(_MODEL_PATH)
+        enforcer.add_policy("allow")  # single gate; the matcher does the deciding
+        _enforcer = enforcer
+    return _enforcer
+
+
+def can_call_model(allowed_models: Optional[List[str]], requested_model: str) -> bool:
+    """Decide whether a principal with ``allowed_models`` may call ``requested_model``."""
+    subject = ModelAccessSubject(allowed_models=list(allowed_models or []))
+    return _get_enforcer().enforce(subject, requested_model)

litellm/proxy/auth/v2/entry.py

@@ -4,10 +4,11 @@ from fastapi import HTTPException, Request, status
 
 from .authenticators import AuthContext, authenticate
 from .authorizer import AuthorizationDenied, authorize
+from .data_plane import can_call_model
 from .enforcer import CasbinEnforcer
 from .policy_store import load_policy_snapshot
 from .principal import build_principal
-from .route_map import match_route
+from .route_map import is_inference_route, match_route
 
 
 async def _anonymous_identity(api_key: Optional[str]) -> Any:
@@ -51,25 +52,45 @@ async def user_api_key_auth_v2(
     )
 
     rule = match_route(route)
-    if rule is None:
-        # Loud-open handled inside authorize(); no identity required here.
-        identity = await _best_effort_identity(token, ctx)
-        authorize(build_principal(identity), route, None, _DENY_ALL)
+    if rule is not None:
+        # Control plane: RBAC policy rows.
+        identity = await authenticate(token, ctx)
+        request_data = await _read_request_body(request=request)
+        principal = build_principal(identity)
+
+        policies, groupings = await load_policy_snapshot(prisma_client)
+        enforcer = CasbinEnforcer(policies, groupings + principal.groupings)
+
+        try:
+            authorize(principal, route, request_data, enforcer)
+        except AuthorizationDenied as e:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN, detail=str(e)
+            )
+
         identity.request_route = route
         return identity
 
-    identity = await authenticate(token, ctx)
-    request_data = await _read_request_body(request=request)
-    principal = build_principal(identity)
-
-    policies, groupings = await load_policy_snapshot(prisma_client)
-    enforcer = CasbinEnforcer(policies, groupings + principal.groupings)
-
-    try:
-        authorize(principal, route, request_data, enforcer)
-    except AuthorizationDenied as e:
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(e))
+    if is_inference_route(route):
+        # Data plane: casbin ABAC over the principal's allowed-model attribute.
+        identity = await authenticate(token, ctx)
+        request_data = await _read_request_body(request=request)
+        requested_model = (
+            request_data.get("model") if isinstance(request_data, dict) else None
+        )
+        if requested_model and not can_call_model(
+            getattr(identity, "models", None), requested_model
+        ):
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail=f"auth_v2: not permitted to call model '{requested_model}'",
+            )
+        identity.request_route = route
+        return identity
 
+    # Loud-open: route v2 doesn't yet govern. No identity required.
+    identity = await _best_effort_identity(token, ctx)
+    authorize(build_principal(identity), route, None, _DENY_ALL)
     identity.request_route = route
     return identity
 

litellm/proxy/auth/v2/route_map.py

@@ -38,7 +38,26 @@ _GOVERNED: Dict[str, GovernedRoute] = {
 }
 
 
+# Inference routes carry a `model` in the body and are authorized on the data
+# plane (casbin ABAC over the principal's allowed-model attribute), not the
+# control-plane RBAC map.
+_INFERENCE_ROUTES = {
+    "/chat/completions",
+    "/v1/chat/completions",
+    "/completions",
+    "/v1/completions",
+    "/embeddings",
+    "/v1/embeddings",
+    "/responses",
+    "/v1/responses",
+}
+
+
 def match_route(route: str) -> Optional[GovernedRoute]:
     """Return the governance rule for ``route``, or None if v2 doesn't yet own it."""
     normalized = route.rstrip("/") or "/"
     return _GOVERNED.get(normalized)
+
+
+def is_inference_route(route: str) -> bool:
+    return (route.rstrip("/") or "/") in _INFERENCE_ROUTES

tests/test_litellm/proxy/auth/v2/test_data_plane.py

@@ -0,0 +1,29 @@
+from litellm.proxy.auth.v2.data_plane import can_call_model
+
+
+def test_model_in_allowed_list_is_permitted():
+    assert can_call_model(["gpt-4o", "claude-3"], "gpt-4o") is True
+
+
+def test_model_not_in_allowed_list_is_denied():
+    assert can_call_model(["gpt-4o"], "claude-3") is False
+
+
+def test_empty_list_is_unrestricted_matching_v1():
+    # In litellm an empty models list means "no restriction", not "no access".
+    assert can_call_model([], "anything") is True
+    assert can_call_model(None, "anything") is True
+
+
+def test_wildcards_are_unrestricted():
+    assert can_call_model(["*"], "anything") is True
+    assert can_call_model(["all-proxy-models"], "anything") is True
+    assert can_call_model(["all-team-models"], "anything") is True
+
+
+def test_specific_list_denies_unlisted_model():
+    assert can_call_model(["gpt-4o", "gpt-4o-mini"], "o1") is False
+
+
+def test_wildcard_mixed_with_specific_still_unrestricted():
+    assert can_call_model(["gpt-4o", "all-proxy-models"], "o1") is True

tests/test_litellm/proxy/auth/v2/test_route_map.py

@@ -1,4 +1,4 @@
-from litellm.proxy.auth.v2.route_map import match_route
+from litellm.proxy.auth.v2.route_map import is_inference_route, match_route
 
 
 def test_model_routes_map_to_resource_and_action():
@@ -31,6 +31,28 @@ def test_trailing_slash_is_normalized():
     assert match_route("/model/info/").resource == "model"
 
 
+def test_inference_routes_are_detected():
+    for route in (
+        "/chat/completions",
+        "/v1/chat/completions",
+        "/embeddings",
+        "/v1/embeddings",
+        "/completions",
+        "/responses",
+    ):
+        assert is_inference_route(route) is True
+
+
+def test_non_inference_routes_are_not_inference():
+    for route in ("/model/new", "/team/info", "/key/generate", "/"):
+        assert is_inference_route(route) is False
+
+
+def test_inference_routes_are_not_control_plane_governed():
+    # Inference is data-plane (model attribute), not in the RBAC route map.
+    assert match_route("/chat/completions") is None
+
+
 def test_ungoverned_routes_return_none():
     # These are loud-open in this slice and must not be governed yet.
     # /team/member_add is deferred with the recursive `manage` action.