feat(auth_v2): add Casbin ABAC authorizer

Add ABACEngine alongside RBACEngine for attribute-based decisions over subject
attributes (roles, teams, org, scopes, claims) and resource attributes
(endpoint, method, model, mcp_server, mcp_tool). Policies are operator-supplied
YAML loaded via add_policy.

The Casbin CSV FileAdapter is avoided on purpose: it retains the quotes around a
comma-bearing expression, turning an eval'd rule into a truthy string literal and
silently allowing inputs that should deny. Claims access yields None for missing
keys so a single policy row referencing an absent claim cannot poison the whole
decision, and rule-evaluation errors fail closed.

Engine only; not yet wired into the live request path

litellm/proxy/auth_v2/authorization/__init__.py

@@ -1,12 +1,13 @@
+from litellm.proxy.auth_v2.authorization.abac import ABACEngine, ProtectedResource
 from litellm.proxy.auth_v2.authorization.base import Authorizer
 from litellm.proxy.auth_v2.authorization.rbac import RBACEngine
 from litellm.proxy.auth_v2.authorization.roles import Role, filter_claim_roles
-from litellm.proxy.auth_v2.authorization.scopes import has_required_scopes
 
 __all__ = [
+    "ABACEngine",
     "Authorizer",
+    "ProtectedResource",
     "RBACEngine",
     "Role",
     "filter_claim_roles",
-    "has_required_scopes",
 ]

litellm/proxy/auth_v2/authorization/abac.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
+
+import casbin
+import yaml
+from pydantic import BaseModel
+
+from litellm.proxy.auth_v2.authorization.base import Authorizer
+from litellm.proxy.auth_v2.authorization.roles import Role
+
+if TYPE_CHECKING:
+    from litellm.proxy.auth_v2.models import Principal
+
+_MODEL_TEXT = """
+[request_definition]
+r = sub, obj, act
+
+[policy_definition]
+p = sub_rule, obj_rule, act
+
+[policy_effect]
+e = some(where (p.eft == allow))
+
+[matchers]
+m = eval(p_sub_rule) && eval(p_obj_rule) && regexMatch(r_act, "^(" + p_act + ")$")
+"""
+
+
+class ProtectedResource(BaseModel):
+    """The resource a principal is acting on, exposed to policy as ``r_obj``."""
+
+    endpoint: Optional[str] = None
+    method: Optional[str] = None
+    model: Optional[str] = None
+    mcp_server: Optional[str] = None
+    mcp_tool: Optional[str] = None
+
+
+class _SafeClaims(dict):
+    """Claims map that yields None for absent keys.
+
+    Casbin evaluates every policy row in a single matcher, so a row referencing
+    ``r_sub.claims['x']`` would raise KeyError for any principal lacking that
+    claim and abort the whole decision. Returning None keeps such a row simply
+    non-matching instead of poisoning unrelated rows.
+    """
+
+    def __missing__(self, key: str) -> None:
+        return None
+
+
+class _Subject:
+    """Principal attributes exposed to policy as ``r_sub``."""
+
+    def __init__(
+        self,
+        roles: List[str],
+        teams: List[str],
+        org: Optional[str],
+        scopes: List[str],
+        claims: Dict[str, Any],
+        user_id: Optional[str],
+        email: Optional[str],
+    ) -> None:
+        self.roles = roles
+        self.teams = teams
+        self.org = org
+        self.scopes = scopes
+        self.claims = _SafeClaims(claims)
+        self.user_id = user_id
+        self.email = email
+
+
+def _subject_view(principal: "Principal") -> _Subject:
+    roles = [role.value for role in principal.roles]
+    claim_roles = principal.claims.get("roles", [])
+    if isinstance(claim_roles, list):
+        roles += [r for r in claim_roles if isinstance(r, str) and r not in roles]
+    return _Subject(
+        roles=roles,
+        teams=[team.name for team in principal.teams if team.name],
+        org=principal.organization.id if principal.organization else None,
+        scopes=list(principal.scopes),
+        claims=principal.claims,
+        user_id=principal.user.id if principal.user else None,
+        email=principal.user.email if principal.user else None,
+    )
+
+
+def _load_policies(policy_path: str) -> List[Tuple[str, str, str]]:
+    with open(policy_path, "r") as handle:
+        document = yaml.safe_load(handle) or {}
+    entries = document.get("policies") or []
+    rules: List[Tuple[str, str, str]] = []
+    for index, entry in enumerate(entries):
+        try:
+            rules.append((entry["sub_rule"], entry["obj_rule"], entry["act"]))
+        except (TypeError, KeyError) as exc:
+            raise ValueError(
+                f"abac policy entry {index} must define sub_rule, obj_rule and act"
+            ) from exc
+    return rules
+
+
+class ABACEngine(Authorizer):
+    """Attribute-based authorizer: subject x resource policies via Casbin.
+
+    Policies are operator-supplied YAML (``{policies: [{sub_rule, obj_rule,
+    act}]}``) loaded into an in-memory enforcer. The CSV FileAdapter is avoided
+    on purpose: it retains the quotes around comma-bearing expressions, turning
+    an eval'd condition into a truthy string literal and silently allowing.
+    """
+
+    def __init__(self, policy_path: Optional[str] = None) -> None:
+        model = casbin.Model()
+        model.load_model_from_text(_MODEL_TEXT)
+        self._enforcer = casbin.Enforcer(model)
+        if policy_path:
+            for rule in _load_policies(policy_path):
+                self._enforcer.add_policy(*rule)
+
+    def decide(self, principal: "Principal", resource: ProtectedResource) -> bool:
+        try:
+            return self._enforcer.enforce(
+                _subject_view(principal), resource, resource.method or ""
+            )
+        except Exception:
+            return False
+
+    def enforce(self, principal: "Principal", obj: str, act: str) -> bool:
+        return self.decide(principal, ProtectedResource(endpoint=obj, method=act))
+
+    def has_any_role(self, principal: "Principal", allowed: Tuple[Role, ...]) -> bool:
+        allowed_values = {role.value for role in allowed}
+        return any(role.value in allowed_values for role in principal.roles)

litellm/proxy/auth_v2/config.py

@@ -44,7 +44,9 @@ class OIDCProviderConfig(BaseModel):
     require_at_jwt: bool = False
     client_id: Optional[str] = None
     client_secret: Optional[SecretStr] = None
-    login_scopes: List[str] = Field(default_factory=lambda: ["openid", "email", "profile"])
+    login_scopes: List[str] = Field(
+        default_factory=lambda: ["openid", "email", "profile"]
+    )
     allowed_roles: List[str] = Field(default_factory=list)
     allow_platform_roles: bool = False
 
@@ -70,14 +72,18 @@ class SAMLConfig(BaseModel):
     sp_cert_file: Optional[str] = None
     allow_unsolicited: bool = False
     xmlsec_binary: Optional[str] = None
-    attribute_map: Dict[str, str] = Field(default_factory=lambda: dict(DEFAULT_SAML_ATTRIBUTE_MAP))
+    attribute_map: Dict[str, str] = Field(
+        default_factory=lambda: dict(DEFAULT_SAML_ATTRIBUTE_MAP)
+    )
     allowed_roles: List[str] = Field(default_factory=list)
     allow_platform_roles: bool = False
 
     @model_validator(mode="after")
     def _require_idp_metadata(self) -> "SAMLConfig":
         if self.enabled and not self.idp_metadata.strip():
-            raise ValueError("SAML enabled but idp_metadata is empty (inline XML, local path, or URL)")
+            raise ValueError(
+                "SAML enabled but idp_metadata is empty (inline XML, local path, or URL)"
+            )
         return self
 
 
@@ -134,3 +140,4 @@ class AuthConfig(BaseModel):
     session: SessionConfig = Field(default_factory=SessionConfig)
     saml: Optional[SAMLConfig] = None
     casbin_policy_path: Optional[str] = None
+    abac_policy_path: Optional[str] = None

tests/test_litellm/proxy/auth_v2/test_abac.py

@@ -0,0 +1,237 @@
+from __future__ import annotations
+
+import pytest
+
+from litellm.proxy.auth_v2.authorization import ABACEngine, ProtectedResource, Role
+from litellm.proxy.auth_v2.models import (
+    AuthMethod,
+    Principal,
+    PrincipalType,
+    TeamIdentity,
+)
+
+# The two policies from the design: a role-restricted model allowlist on
+# /v1/messages, and a team-restricted MCP tool allowlist on the github server.
+POLICIES = """
+policies:
+  - sub_rule: "'manager' in r_sub.roles"
+    obj_rule: "r_obj.endpoint == '/v1/messages' and r_obj.model in ['claude-sonnet-4-6','gpt-4o']"
+    act: "POST"
+  - sub_rule: "'eng' in r_sub.teams"
+    obj_rule: "r_obj.mcp_server == 'github' and r_obj.mcp_tool in ['search','read_file']"
+    act: "POST|GET"
+"""
+
+
+def _principal(*, roles=None, teams=None, claims=None, scopes=None) -> Principal:
+    return Principal(
+        principal_type=PrincipalType.HUMAN,
+        subject="u1",
+        auth_method=AuthMethod.OIDC,
+        roles=roles or [],
+        teams=teams or [],
+        claims=claims or {},
+        scopes=scopes or [],
+    )
+
+
+def _manager() -> Principal:
+    return _principal(claims={"roles": ["manager"]})
+
+
+def _eng() -> Principal:
+    return _principal(teams=[TeamIdentity(id="t1", name="eng")])
+
+
+def _engine_from(tmp_path, policy_text: str) -> ABACEngine:
+    path = tmp_path / "abac.yaml"
+    path.write_text(policy_text)
+    return ABACEngine(policy_path=str(path))
+
+
+@pytest.fixture
+def engine(tmp_path) -> ABACEngine:
+    return _engine_from(tmp_path, POLICIES)
+
+
+# --------------------------------------------------------------------------- #
+# Model allowlist policy: each attribute is an independent gate
+# --------------------------------------------------------------------------- #
+
+
+def test_manager_allowed_model_is_permitted(engine):
+    assert engine.decide(
+        _manager(),
+        ProtectedResource(endpoint="/v1/messages", model="gpt-4o", method="POST"),
+    )
+
+
+def test_manager_unlisted_model_is_denied(engine):
+    # set membership, not just "manager is allowed"; also the exact case that the
+    # CSV FileAdapter silently *allowed* (quotes leaking into the eval'd rule)
+    assert not engine.decide(
+        _manager(),
+        ProtectedResource(
+            endpoint="/v1/messages", model="claude-opus-4-8", method="POST"
+        ),
+    )
+
+
+def test_non_manager_is_denied_allowed_model(engine):
+    assert not engine.decide(
+        _principal(),
+        ProtectedResource(endpoint="/v1/messages", model="gpt-4o", method="POST"),
+    )
+
+
+def test_manager_wrong_action_is_denied(engine):
+    assert not engine.decide(
+        _manager(),
+        ProtectedResource(endpoint="/v1/messages", model="gpt-4o", method="GET"),
+    )
+
+
+def test_manager_wrong_endpoint_is_denied(engine):
+    assert not engine.decide(
+        _manager(),
+        ProtectedResource(
+            endpoint="/v1/chat/completions", model="gpt-4o", method="POST"
+        ),
+    )
+
+
+def test_manager_without_model_is_denied(engine):
+    # a model-restricted rule must not match a request that carries no model
+    assert not engine.decide(
+        _manager(), ProtectedResource(endpoint="/v1/messages", method="POST")
+    )
+
+
+# --------------------------------------------------------------------------- #
+# MCP tool allowlist policy: team membership x server x tool x action
+# --------------------------------------------------------------------------- #
+
+
+def test_eng_listed_tool_is_permitted(engine):
+    assert engine.decide(
+        _eng(),
+        ProtectedResource(mcp_server="github", mcp_tool="search", method="POST"),
+    )
+    assert engine.decide(
+        _eng(),
+        ProtectedResource(mcp_server="github", mcp_tool="read_file", method="GET"),
+    )
+
+
+def test_eng_unlisted_tool_is_denied(engine):
+    assert not engine.decide(
+        _eng(),
+        ProtectedResource(mcp_server="github", mcp_tool="delete_repo", method="POST"),
+    )
+
+
+def test_eng_wrong_server_is_denied(engine):
+    assert not engine.decide(
+        _eng(),
+        ProtectedResource(mcp_server="gitlab", mcp_tool="search", method="POST"),
+    )
+
+
+def test_non_eng_team_is_denied(engine):
+    sales = _principal(teams=[TeamIdentity(id="t2", name="sales")])
+    assert not engine.decide(
+        sales,
+        ProtectedResource(mcp_server="github", mcp_tool="search", method="POST"),
+    )
+
+
+def test_unrelated_policy_row_does_not_poison_decision(engine):
+    # the manager request must be allowed even though the eng (team) policy row
+    # is also evaluated against a principal with no teams
+    assert engine.decide(
+        _manager(),
+        ProtectedResource(endpoint="/v1/messages", model="gpt-4o", method="POST"),
+    )
+
+
+# --------------------------------------------------------------------------- #
+# Default deny and claim-access safety
+# --------------------------------------------------------------------------- #
+
+
+def test_no_policy_denies_everything():
+    engine = ABACEngine()
+    assert not engine.decide(
+        _manager(),
+        ProtectedResource(endpoint="/v1/messages", model="gpt-4o", method="POST"),
+    )
+
+
+def test_missing_claim_does_not_raise_and_denies(tmp_path):
+    policy = """
+policies:
+  - sub_rule: "r_sub.claims['department'] == 'eng'"
+    obj_rule: "r_obj.endpoint == '/reports'"
+    act: "GET"
+"""
+    engine = _engine_from(tmp_path, policy)
+    assert engine.decide(
+        _principal(claims={"department": "eng"}),
+        ProtectedResource(endpoint="/reports", method="GET"),
+    )
+    # principal lacking the claim: must deny, not raise
+    assert not engine.decide(
+        _principal(), ProtectedResource(endpoint="/reports", method="GET")
+    )
+
+
+def test_malformed_policy_entry_fails_fast(tmp_path):
+    bad = tmp_path / "abac.yaml"
+    bad.write_text('policies:\n  - sub_rule: "1 == 1"\n    act: GET\n')
+    with pytest.raises(ValueError):
+        ABACEngine(policy_path=str(bad))
+
+
+def test_rule_evaluation_error_fails_closed(tmp_path):
+    # an operator referencing an attribute the resource does not carry must deny,
+    # not surface the exception
+    policy = """
+policies:
+  - sub_rule: "True"
+    obj_rule: "r_obj.nonexistent == 1"
+    act: "GET"
+"""
+    engine = _engine_from(tmp_path, policy)
+    assert not engine.decide(
+        _principal(), ProtectedResource(endpoint="/x", method="GET")
+    )
+
+
+# --------------------------------------------------------------------------- #
+# Authorizer protocol compatibility (path/method enforce, flat has_any_role)
+# --------------------------------------------------------------------------- #
+
+
+def test_enforce_path_method_policy(tmp_path):
+    policy = """
+policies:
+  - sub_rule: "'admin' in r_sub.roles"
+    obj_rule: "r_obj.endpoint == '/health'"
+    act: "GET"
+"""
+    engine = _engine_from(tmp_path, policy)
+    admin = _principal(claims={"roles": ["admin"]})
+    assert engine.enforce(admin, "/health", "GET")
+    assert not engine.enforce(admin, "/health", "POST")
+    assert not engine.enforce(admin, "/secrets", "GET")
+    assert not engine.enforce(_principal(), "/health", "GET")
+
+
+def test_has_any_role_is_flat_membership(tmp_path):
+    engine = _engine_from(tmp_path, POLICIES)
+    assert engine.has_any_role(_principal(roles=[Role.ORG_ADMIN]), (Role.ORG_ADMIN,))
+    # no hierarchy: platform_admin does not imply org_admin here (use RBACEngine)
+    assert not engine.has_any_role(
+        _principal(roles=[Role.PLATFORM_ADMIN]), (Role.ORG_ADMIN,)
+    )
+    assert not engine.has_any_role(_principal(), (Role.TEAM_MEMBER,))