Preserve lazy configured-route classification in native auth

litellm-rust/crates/python-bridge/src/lib.rs

@@ -65,6 +65,7 @@ mod _native {
         request_auth_managed_model_grants,
         request_auth_model_projection,
         request_auth_route_enable,
+        request_auth_llm_route,
     };
     #[pymodule_export]
     use crate::router::{RouterCall, RouterCoreHandle};

litellm-rust/crates/python-bridge/src/request_auth.rs

@@ -348,3 +348,17 @@ pub(crate) fn request_auth_route_enable(py: Python<'_>, facts: &Bound<'_, PyAny>
     let decision = py.detach(move || core::route_enable::decide(&facts)).map_err(map_error)?;
     to_py(py, &decision)
 }
+
+#[pyfunction]
+#[pyo3(signature = (context, configured=None))]
+pub(crate) fn request_auth_llm_route(
+    py: Python<'_>,
+    context: &Bound<'_, PyAny>,
+    configured: Option<bool>,
+) -> PyResult<Py<PyAny>> {
+    let context: core::route_access::RouteContext = from_py(context)?;
+    let decision = py
+        .detach(move || core::route_enable::llm_route(&context, configured))
+        .map_err(map_error)?;
+    to_py(py, &decision)
+}

litellm-rust/crates/request-auth-core/src/route_enable.rs

@@ -25,6 +25,7 @@ pub struct Facts {
     pub admin: Option<Setting>,
     pub llm: Option<Setting>,
     pub exemptions: Option<Vec<String>>,
+    pub eu_jev_configured: Option<bool>,
 }
 
 #[derive(Clone, Copy, Deserialize, strum::VariantArray)]
@@ -46,11 +47,42 @@ pub enum Decision {
     PropagateHttp,
     ReadImport,
     ReadExemptions,
+    ReadEuJev,
     ReadSettings { feature: Feature },
     ReadSecret { feature: Feature },
     Deny { feature: Feature, reason: Reason },
 }
 
+#[derive(Debug, PartialEq, Eq, Serialize, strum::VariantArray)]
+pub enum LlmDecision {
+    Allow,
+    Deny,
+    ReadEuJev,
+}
+
+pub fn llm_route(context: &RouteContext, configured: Option<bool>) -> Result<LlmDecision, Error> {
+    let static_context = RouteContext {
+        configured_eu_jev: true,
+        ..context.clone()
+    };
+    if route_access::decide(route_access::Request::Classify {
+        context: static_context,
+        kind: Classification::Llm,
+    })?
+    .allowed
+    {
+        return Ok(LlmDecision::Allow);
+    }
+    if context.route.trim_end_matches('/') != "/eu_jev/v1/systemone" {
+        return Ok(LlmDecision::Deny);
+    }
+    Ok(match configured {
+        None => LlmDecision::ReadEuJev,
+        Some(true) => LlmDecision::Deny,
+        Some(false) => LlmDecision::Allow,
+    })
+}
+
 fn setting(setting: Option<&Setting>, feature: Feature) -> Option<Decision> {
     let Some(setting) = setting else {
         return Some(Decision::ReadSettings { feature });
@@ -93,13 +125,10 @@ pub fn decide(facts: &Facts) -> Result<Decision, Error> {
     {
         return Ok(decision);
     }
-    if !route_access::decide(route_access::Request::Classify {
-        context: facts.context.clone(),
-        kind: Classification::Llm,
-    })?
-    .allowed
-    {
-        return Ok(Decision::Allow);
+    match llm_route(&facts.context, facts.eu_jev_configured)? {
+        LlmDecision::Deny => return Ok(Decision::Allow),
+        LlmDecision::ReadEuJev => return Ok(Decision::ReadEuJev),
+        LlmDecision::Allow => (),
     }
     let Some(exemptions) = &facts.exemptions else {
         return Ok(Decision::ReadExemptions);
@@ -115,6 +144,18 @@ mod tests {
     use super::*;
     use rstest::rstest;
 
+    #[rstest]
+    #[case::missing(None, LlmDecision::ReadEuJev)]
+    #[case::builtin(Some(false), LlmDecision::Allow)]
+    #[case::configured(Some(true), LlmDecision::Deny)]
+    fn builtin_configuration(#[case] configured: Option<bool>, #[case] expected: LlmDecision) {
+        let context = RouteContext {
+            route: "/eu_jev/v1/systemone/".into(),
+            ..RouteContext::default()
+        };
+        assert_eq!(llm_route(&context, configured).unwrap(), expected);
+    }
+
     #[rstest]
     #[case::license(true, false, Decision::Deny { feature: Feature::Admin, reason: Reason::License })]
     #[case::absent(false, false, Decision::ReadSecret { feature: Feature::Admin })]

litellm/_auth/route_enable.py

@@ -8,6 +8,7 @@ from litellm._auth.routes import RouteContext
 from litellm.rust_bridge import _native
 
 RouteFeature: TypeAlias = Literal["Admin", "Llm"]
+LlmRouteDecision: TypeAlias = Literal["Allow", "Deny", "ReadEuJev"]
 
 
 @dataclass(frozen=True, slots=True)
@@ -25,18 +26,27 @@ class RouteEnableFacts:
     admin: RouteSetting | None = None
     llm: RouteSetting | None = None
     exemptions: tuple[str, ...] | None = None
+    eu_jev_configured: bool | None = None
 
 
 @dataclass(frozen=True, slots=True)
 class RouteEnableDecision:
-    kind: Literal["Allow", "PropagateHttp", "ReadImport", "ReadExemptions", "ReadSettings", "ReadSecret", "Deny"]
+    kind: Literal[
+        "Allow", "PropagateHttp", "ReadImport", "ReadExemptions", "ReadEuJev", "ReadSettings", "ReadSecret", "Deny"
+    ]
     feature: RouteFeature | None = None
     reason: Literal["License", "Disabled"] | None = None
 
 
 _DECISION: Final = TypeAdapter(RouteEnableDecision)
+_LLM: Final = TypeAdapter[LlmRouteDecision](LlmRouteDecision)
 
 
 @native_boundary
 def route_enable(facts: RouteEnableFacts) -> RouteEnableDecision:
     return _DECISION.validate_python(_native.request_auth_route_enable(asdict(facts)))
+
+
+@native_boundary
+def llm_route(context: RouteContext, configured: bool | None = None) -> LlmRouteDecision:
+    return _LLM.validate_python(_native.request_auth_llm_route(asdict(context), configured))

litellm/proxy/auth/route_enable_adapter.py

@@ -9,7 +9,7 @@ from pydantic import BaseModel
 from litellm._auth.failure import NativeAuthFailure, NativeAuthRuntimeError, native_boundary
 from litellm._auth.route_enable import RouteEnableDecision, RouteEnableFacts, RouteFeature, RouteSetting, route_enable
 from litellm.proxy._types import CommonProxyErrors
-from litellm.proxy.auth.route_policy_adapter import route_context
+from litellm.proxy.auth.route_policy_adapter import read_eu_jev_configuration, route_context
 
 _KEYS: Final = {"Admin": "DISABLE_ADMIN_ENDPOINTS", "Llm": "DISABLE_LLM_API_ENDPOINTS"}
 
@@ -50,6 +50,7 @@ class RouteEnableReaders:
     setting: Callable[[RouteFeature], RouteSetting] = read_setting
     secret: Callable[[RouteFeature], bool] = read_secret
     exemptions: Callable[[object], tuple[str, ...]] = read_exemptions
+    eu_jev: Callable[[], bool] = read_eu_jev_configuration
 
 
 _READERS: Final = RouteEnableReaders()
@@ -63,6 +64,8 @@ def _answer(
             return replace(facts, imported=True), readers.enterprise()
         case "ReadExemptions":
             return replace(facts, exemptions=readers.exemptions(enterprise)), enterprise
+        case "ReadEuJev":
+            return replace(facts, eu_jev_configured=readers.eu_jev()), enterprise
         case "ReadSettings" | "ReadSecret":
             if decision.feature is None:
                 raise NativeAuthRuntimeError("route setting read without feature")

litellm/proxy/auth/route_policy_adapter.py

@@ -219,7 +219,21 @@ def classify(
     kind: Literal["Llm", "Unbilled", "Discovery", "SelfCredential", "Azure", "Managed"],
     method: str | None = None,
 ) -> bool:
-    return routes.classify_route(route_context(route, method, registration=kind == "Llm"), kind)
+    from litellm._auth.route_enable import llm_route
+
+    context: Final = route_context(route, method)
+    if kind != "Llm":
+        return routes.classify_route(context, kind)
+    decision: Final = llm_route(context)
+    if decision == "ReadEuJev":
+        return llm_route(context, read_eu_jev_configuration()) == "Allow"
+    return decision == "Allow"
+
+
+def read_eu_jev_configuration() -> bool:
+    from litellm.proxy.pass_through_endpoints.pass_through_endpoints import InitPassThroughEndpointHelpers
+
+    return InitPassThroughEndpointHelpers.has_configured_route_ignoring_trailing_slash("/eu_jev/v1/systemone")
 
 
 def denied_passthrough(route: str, metadata_sources: Iterable[Mapping[str, object] | None]) -> str | None:

litellm/rust_bridge/_native.pyi

@@ -54,6 +54,7 @@ def request_auth_agent_ceiling_source(facts: object) -> object: ...
 def request_auth_managed_model_grants(facts: object) -> object: ...
 def request_auth_model_projection(facts: object, reads: object) -> object: ...
 def request_auth_route_enable(facts: object) -> object: ...
+def request_auth_llm_route(context: object, configured: bool | None = None) -> object: ...
 
 @final
 class NativeClickHouseSpendConfig:

tests/unit/_auth/support/route_enable_reference.py

@@ -25,8 +25,13 @@ def route_enable(facts: RouteEnableFacts) -> RouteEnableDecision:
         admin: Final = _setting(facts.admin, "Admin")
         if admin is not None:
             return admin
-    if not RouteChecks.is_llm_api_route(facts.context.route):
-        return RouteEnableDecision("Allow")
+    if not RouteChecks.is_llm_api_route(facts.context.route, configured_eu_jev=True):
+        if facts.context.route.rstrip("/") != "/eu_jev/v1/systemone":
+            return RouteEnableDecision("Allow")
+        if facts.eu_jev_configured is None:
+            return RouteEnableDecision("ReadEuJev")
+        if facts.eu_jev_configured:
+            return RouteEnableDecision("Allow")
     if facts.exemptions is None:
         return RouteEnableDecision("ReadExemptions")
     if facts.context.route in facts.exemptions:

tests/unit/_auth/support/route_reference.py

@@ -553,7 +553,7 @@ class RouteChecks:
         return route in _UNBILLED_ROUTES or resource in _TOKEN_COUNTING_SEGMENTS or action == _TOKEN_COUNTING_ACTION
 
     @staticmethod
-    def is_llm_api_route(route: str) -> bool:
+    def is_llm_api_route(route: str, configured_eu_jev: bool | None = None) -> bool:
         """
         Helper to checks if provided route is an OpenAI route
 
@@ -615,12 +615,14 @@ class RouteChecks:
         for _llm_passthrough_route in LiteLLMRoutes.mapped_pass_through_routes.value:
             if route == _llm_passthrough_route or route.startswith(_llm_passthrough_route + "/"):
                 return True
-        return RouteChecks.is_builtin_eu_jev_decision_route(route=route)
+        return RouteChecks.is_builtin_eu_jev_decision_route(route=route, configured_eu_jev=configured_eu_jev)
 
     @staticmethod
-    def is_builtin_eu_jev_decision_route(route: str) -> bool:
+    def is_builtin_eu_jev_decision_route(route: str, configured_eu_jev: bool | None = None) -> bool:
         if route.rstrip("/") != _EU_JEV_DECISION_ROUTE:
             return False
+        if configured_eu_jev is not None:
+            return not configured_eu_jev
         from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (  # noqa: PLC0415  # pass_through_endpoints imports the auth stack
             InitPassThroughEndpointHelpers,
         )

tests/unit/_auth/test_route_enable.py

@@ -3,7 +3,14 @@ from typing import Final, Literal
 
 import pytest
 
-from litellm._auth.route_enable import RouteEnableDecision, RouteEnableFacts, RouteSetting, route_enable
+from litellm._auth.route_enable import (
+    LlmRouteDecision,
+    RouteEnableDecision,
+    RouteEnableFacts,
+    RouteSetting,
+    llm_route,
+    route_enable,
+)
 from litellm.proxy.auth.route_policy_adapter import route_context
 from tests.unit._auth.support.route_enable_reference import route_enable as reference
 
@@ -36,3 +43,33 @@ def test_optional_route_enable_decision_table(
 ) -> None:
     facts: Final = RouteEnableFacts(route_context(route), imported, failed, setting, setting, exemptions)
     assert policy(facts) == reference(facts)
+
+
+@pytest.mark.parametrize(
+    ("route", "configured", "expected"),
+    (
+        ("/chat/completions", None, "Allow"),
+        ("/unknown", None, "Deny"),
+        ("/eu_jev/v1/systemone", None, "ReadEuJev"),
+        ("/eu_jev/v1/systemone", False, "Allow"),
+        ("/eu_jev/v1/systemone", True, "Deny"),
+        ("/eu_jev/v1/systemone/", True, "Deny"),
+    ),
+)
+def test_llm_route_configuration_read_is_selected_only_at_builtin_fallback(
+    route: str, configured: bool | None, expected: LlmRouteDecision
+) -> None:
+    assert llm_route(route_context(route), configured) == expected
+
+
+@pytest.mark.parametrize("configured", (None, False, True))
+@pytest.mark.parametrize("route", ("/eu_jev/v1/systemone", "/eu_jev/v1/systemone/"))
+def test_optional_route_builtin_configuration_matches_reference(configured: bool | None, route: str) -> None:
+    facts: Final = RouteEnableFacts(
+        route_context(route),
+        imported=True,
+        llm=RouteSetting(True, True, True),
+        exemptions=(),
+        eu_jev_configured=configured,
+    )
+    assert route_enable(facts) == reference(facts)

tests/unit/proxy/auth/test_route_enable_adapter.py

@@ -86,3 +86,29 @@ def test_optional_import_failure_policy(error: Exception) -> None:
         assert caught.value is error
     else:
         assert check_route_enabled("/key/generate", readers=readers) is None
+
+
+@pytest.mark.parametrize("configured", (False, True))
+@pytest.mark.parametrize("route", ("/eu_jev/v1/systemone", "/eu_jev/v1/systemone/"))
+def test_configured_eu_jev_route_skips_llm_disable_setting(configured: bool, route: str) -> None:
+    events: Final[SimpleQueue[str]] = SimpleQueue()
+
+    def eu_jev() -> bool:
+        events.put("configuration")
+        return configured
+
+    def setting(feature: RouteFeature) -> RouteSetting:
+        events.put(feature)
+        return RouteSetting(True, True)
+
+    readers: Final = RouteEnableReaders(
+        enterprise=lambda: object(), setting=setting, secret=lambda _: True, exemptions=lambda _: (), eu_jev=eu_jev
+    )
+    if configured:
+        assert check_route_enabled(route, readers=readers) is None
+        assert tuple(events.get() for _ in range(events.qsize())) == ("configuration",)
+    else:
+        with pytest.raises(HTTPException) as error:
+            check_route_enabled(route, readers=readers)
+        assert (error.value.status_code, error.value.detail) == (403, "LLM API routes are disabled for this instance.")
+        assert tuple(events.get() for _ in range(events.qsize())) == ("configuration", "Llm")