"""
Unit tests for auth_utils functions related to rate limiting and customer ID extraction.
"""

import base64
from typing import Optional
from unittest.mock import MagicMock, patch

import pytest
from fastapi import Request

from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.auth.auth_utils import (
    _get_customer_id_from_standard_headers,
    abbreviate_api_key,
    check_complete_credentials,
    custom_auth_common_checks_warning,
    warn_once_if_custom_auth_skips_common_checks,
    get_end_user_id_from_request_body,
    get_key_mcp_rpm_limit,
    get_key_model_rpm_limit,
    get_key_model_tpm_limit,
    get_key_tag_rpm_limit,
    get_model_from_request,
    get_project_model_rpm_limit,
    get_project_model_tpm_limit,
    get_request_route_template,
    is_request_body_safe,
)


class TestCustomAuthCommonChecksWarning:
    """custom_auth_common_checks_warning only warns when custom auth is configured
    and the common-checks opt-in is off, since that is the only state where
    project/team enforcement silently does nothing."""

    def test_warns_when_custom_auth_configured_and_checks_off(self):
        warning = custom_auth_common_checks_warning(
            custom_auth_configured=True,
            run_common_checks=False,
        )
        assert warning is not None
        assert "custom_auth_run_common_checks: true" in warning
        assert "https://docs.litellm.ai/docs/proxy/custom_auth" in warning

    def test_no_warning_when_common_checks_enabled(self):
        assert (
            custom_auth_common_checks_warning(
                custom_auth_configured=True,
                run_common_checks=True,
            )
            is None
        )

    def test_no_warning_when_custom_auth_not_configured(self):
        assert (
            custom_auth_common_checks_warning(
                custom_auth_configured=False,
                run_common_checks=False,
            )
            is None
        )
        assert (
            custom_auth_common_checks_warning(
                custom_auth_configured=False,
                run_common_checks=True,
            )
            is None
        )


class TestWarnOnceIfCustomAuthSkipsCommonChecks:
    """The startup warning must fire at most once per process, since load_config
    re-runs on hot-reload / config refresh and would otherwise spam the log."""

    @pytest.fixture(autouse=True)
    def _reset_sentinel(self, monkeypatch):
        monkeypatch.setattr(
            "litellm.proxy.auth.auth_utils._custom_auth_common_checks_warning_emitted",
            False,
        )

    def test_warns_only_once_across_repeated_calls(self):
        logger = MagicMock()
        for _ in range(3):
            warn_once_if_custom_auth_skips_common_checks(
                custom_auth_configured=True,
                run_common_checks=False,
                logger=logger,
            )
        assert logger.warning.call_count == 1
        assert "custom_auth_run_common_checks" in logger.warning.call_args[0][0]

    def test_does_not_warn_when_common_checks_enabled(self):
        logger = MagicMock()
        warn_once_if_custom_auth_skips_common_checks(
            custom_auth_configured=True,
            run_common_checks=True,
            logger=logger,
        )
        assert logger.warning.call_count == 0


class TestGetKeyModelRpmLimit:
    """Tests for get_key_model_rpm_limit function."""

    def test_returns_key_metadata_when_present(self):
        """Key metadata takes priority over team metadata."""
        user_api_key_dict = UserAPIKeyAuth(
            api_key="sk-123",
            metadata={"model_rpm_limit": {"gpt-4": 100}},
            team_metadata={"model_rpm_limit": {"gpt-4": 50}},
        )
        result = get_key_model_rpm_limit(user_api_key_dict)
        assert result == {"gpt-4": 100}

    def test_falls_back_to_team_metadata_when_key_has_other_metadata(self):
        """Should fall back to team metadata when key metadata exists but has no model_rpm_limit."""
        user_api_key_dict = UserAPIKeyAuth(
            api_key="sk-123",
            metadata={
                "some_other_key": "value"
            },  # Has metadata, but not model_rpm_limit
            team_metadata={"model_rpm_limit": {"gpt-4": 50}},
        )
        result = get_key_model_rpm_limit(user_api_key_dict)
        assert result == {"gpt-4": 50}

    def test_extracts_from_model_max_budget(self):
        """Should extract rpm_limit from model_max_budget when metadata is empty."""
        user_api_key_dict = UserAPIKeyAuth(
            api_key="sk-123",
            model_max_budget={
                "gpt-4": {"rpm_limit": 100, "tpm_limit": 1000},
                "gpt-3.5-turbo": {"rpm_limit": 200},
            },
        )
        result = get_key_model_rpm_limit(user_api_key_dict)
        assert result == {"gpt-4": 100, "gpt-3.5-turbo": 200}

    def test_skips_models_without_rpm_limit(self):
        """Should skip models that don't have rpm_limit in model_max_budget."""
        user_api_key_dict = UserAPIKeyAuth(
            api_key="sk-123",
            model_max_budget={
                "gpt-4": {"rpm_limit": 100},
                "gpt-3.5-turbo": {"tpm_limit": 1000},  # No rpm_limit
            },
        )
        result = get_key_model_rpm_limit(user_api_key_dict)
        assert result == {"gpt-4": 100}

    def test_returns_none_when_no_limits_configured(self):
        """Should return None when no rate limits are configured."""
        user_api_key_dict = UserAPIKeyAuth(api_key="sk-123")
        result = get_key_model_rpm_limit(user_api_key_dict)
        assert result is None

    def test_team_metadata_empty_rpm_dict_falls_through_to_deployment_default(self):
        """Explicitly empty team model_rpm_limit ({}) should be returned as-is, not fallen through."""
        # An empty dict is a valid team limit map (no per-model limits configured).
        # It should be returned directly rather than falling through to deployment defaults,
        # so a team with an empty map is treated as unconstrained at the team level.
        user_api_key_dict = UserAPIKeyAuth(
            api_key="sk-123",
            team_metadata={"model_rpm_limit": {}},
        )
        result = get_key_model_rpm_limit(user_api_key_dict)
        assert result == {}


class TestGetKeyMcpRpmLimit:
    def test_empty_dict_limits_are_returned(self):
        key_override = UserAPIKeyAuth(
            api_key="sk-123",
            metadata={"mcp_rpm_limit": {}},
            team_metadata={"mcp_rpm_limit": {"github": 50}},
        )
        assert get_key_mcp_rpm_limit(key_override) == {}

        team_empty = UserAPIKeyAuth(
            api_key="sk-123",
            team_metadata={"mcp_rpm_limit": {}},
        )
        assert get_key_mcp_rpm_limit(team_empty) == {}


class TestGetKeyModelTpmLimit:
    """Tests for get_key_model_tpm_limit function."""

    def test_returns_key_metadata_when_present(self):
        """Key metadata takes priority over team metadata."""
        user_api_key_dict = UserAPIKeyAuth(
            api_key="sk-123",
            metadata={"model_tpm_limit": {"gpt-4": 10000}},
            team_metadata={"model_tpm_limit": {"gpt-4": 5000}},
        )
        result = get_key_model_tpm_limit(user_api_key_dict)
        assert result == {"gpt-4": 10000}

    def test_falls_back_to_team_metadata_when_key_has_other_metadata(self):
        """Should fall back to team metadata when key metadata exists but has no model_tpm_limit."""
        user_api_key_dict = UserAPIKeyAuth(
            api_key="sk-123",
            metadata={
                "some_other_key": "value"
            },  # Has metadata, but not model_tpm_limit
            team_metadata={"model_tpm_limit": {"gpt-4": 5000}},
        )
        result = get_key_model_tpm_limit(user_api_key_dict)
        assert result == {"gpt-4": 5000}

    def test_extracts_from_model_max_budget(self):
        """Should extract tpm_limit from model_max_budget when metadata is empty."""
        user_api_key_dict = UserAPIKeyAuth(
            api_key="sk-123",
            model_max_budget={
                "gpt-4": {"tpm_limit": 10000, "rpm_limit": 100},
                "gpt-3.5-turbo": {"tpm_limit": 20000},
            },
        )
        result = get_key_model_tpm_limit(user_api_key_dict)
        assert result == {"gpt-4": 10000, "gpt-3.5-turbo": 20000}

    def test_skips_models_without_tpm_limit(self):
        """Should skip models that don't have tpm_limit in model_max_budget."""
        user_api_key_dict = UserAPIKeyAuth(
            api_key="sk-123",
            model_max_budget={
                "gpt-4": {"tpm_limit": 10000},
                "gpt-3.5-turbo": {"rpm_limit": 100},  # No tpm_limit
            },
        )
        result = get_key_model_tpm_limit(user_api_key_dict)
        assert result == {"gpt-4": 10000}

    def test_returns_none_when_no_limits_configured(self):
        """Should return None when no rate limits are configured."""
        user_api_key_dict = UserAPIKeyAuth(api_key="sk-123")
        result = get_key_model_tpm_limit(user_api_key_dict)
        assert result is None

    def test_model_max_budget_priority_over_team(self):
        """model_max_budget should take priority over team_metadata."""
        user_api_key_dict = UserAPIKeyAuth(
            api_key="sk-123",
            model_max_budget={"gpt-4": {"tpm_limit": 10000}},
            team_metadata={"model_tpm_limit": {"gpt-4": 5000}},
        )
        result = get_key_model_tpm_limit(user_api_key_dict)
        assert result == {"gpt-4": 10000}

    def test_team_metadata_empty_tpm_dict_falls_through_to_deployment_default(self):
        """Explicitly empty team model_tpm_limit ({}) should be returned as-is, not fallen through."""
        # An empty dict is a valid team limit map (no per-model limits configured).
        # It should be returned directly rather than falling through to deployment defaults,
        # so a team with an empty map is treated as unconstrained at the team level.
        user_api_key_dict = UserAPIKeyAuth(
            api_key="sk-123",
            team_metadata={"model_tpm_limit": {}},
        )
        result = get_key_model_tpm_limit(user_api_key_dict)
        assert result == {}

    def test_skips_deployments_with_malformed_limit_value(self):
        """Deployments with non-integer-parseable limit values are skipped without raising."""
        user_api_key_dict = UserAPIKeyAuth(api_key="sk-123")
        mock_router = MagicMock()
        mock_router.get_model_list.return_value = [
            {
                "model_name": "model1",
                "litellm_params": {"default_api_key_tpm_limit": "not-a-number"},
            },
            _make_deployment_dict("model1", tpm=500),
        ]
        with patch(_ROUTER_PATCH, mock_router):
            result = get_key_model_tpm_limit(user_api_key_dict, model_name="model1")
        # The malformed deployment is skipped; the valid one provides 500
        assert result == {"model1": 500}


class TestGetCustomerIdFromStandardHeaders:
    """Tests for _get_customer_id_from_standard_headers helper function."""

    def test_should_return_customer_id_from_x_litellm_customer_id_header(self):
        """Should extract customer ID from x-litellm-customer-id header."""
        headers = {"x-litellm-customer-id": "customer-123"}
        result = _get_customer_id_from_standard_headers(request_headers=headers)
        assert result == "customer-123"

    def test_should_return_customer_id_from_x_litellm_end_user_id_header(self):
        """Should extract customer ID from x-litellm-end-user-id header."""
        headers = {"x-litellm-end-user-id": "end-user-456"}
        result = _get_customer_id_from_standard_headers(request_headers=headers)
        assert result == "end-user-456"

    def test_should_return_none_when_headers_is_none(self):
        """Should return None when headers is None."""
        result = _get_customer_id_from_standard_headers(request_headers=None)
        assert result is None

    def test_should_return_none_when_no_standard_headers_present(self):
        """Should return None when no standard customer ID headers are present."""
        headers = {"x-other-header": "some-value"}
        result = _get_customer_id_from_standard_headers(request_headers=headers)
        assert result is None


class TestGetEndUserIdFromRequestBodyWithStandardHeaders:
    """Tests for get_end_user_id_from_request_body with standard customer ID headers."""

    def test_should_prioritize_standard_header_over_body_user(self):
        """Standard customer ID header should take precedence over body user field."""
        headers = {"x-litellm-customer-id": "header-customer"}
        request_body = {"user": "body-user"}

        with patch("litellm.proxy.proxy_server.general_settings", {}):
            result = get_end_user_id_from_request_body(
                request_body=request_body, request_headers=headers
            )
        assert result == "header-customer"

    def test_should_fall_back_to_body_when_no_standard_header(self):
        """Should fall back to body user when no standard headers are present."""
        headers = {"x-other-header": "value"}
        request_body = {"user": "body-user"}

        with patch("litellm.proxy.proxy_server.general_settings", {}):
            result = get_end_user_id_from_request_body(
                request_body=request_body, request_headers=headers
            )
        assert result == "body-user"


def _request_dispatched_to(endpoint) -> Request:
    """Build a minimal Request whose FastAPI-resolved endpoint is ``endpoint``,
    mirroring what Starlette sets in ``scope`` once routing has matched."""
    return Request(scope={"type": "http", "headers": [], "endpoint": endpoint})


def _pass_through_endpoint():
    from litellm.types.passthrough_endpoints.pass_through_endpoints import (
        LITELLM_PASS_THROUGH_ENDPOINT_MARKER,
    )

    def endpoint():  # stand-in for create_pass_through_route's handler
        ...

    setattr(endpoint, LITELLM_PASS_THROUGH_ENDPOINT_MARKER, True)
    return endpoint


def test_get_model_from_request_skips_pass_through_dispatched_request():
    """When FastAPI dispatched the request to a user-defined pass-through handler,
    the body `model` names an upstream model and must not be treated as a LiteLLM
    model for allowlist/budget enforcement."""
    assert (
        get_model_from_request(
            request_data={"model": "upstream-special-model"},
            route="/my-custom-endpoint",
            request=_request_dispatched_to(_pass_through_endpoint()),
        )
        is None
    )


def test_get_model_from_request_enforces_when_builtin_handler_dispatched():
    """A custom pass-through path that collides with a built-in route resolves to the
    built-in handler (no marker), so the body `model` must still be extracted and
    enforced. Same request path as above, but dispatched to a non-pass-through
    endpoint: the model must NOT be suppressed."""

    def builtin_chat_completions():
        ...

    assert (
        get_model_from_request(
            request_data={"model": "gpt-4o"},
            route="/v1/chat/completions",
            request=_request_dispatched_to(builtin_chat_completions),
        )
        == "gpt-4o"
    )


def test_get_model_from_request_no_request_extracts_model():
    """Callers without a request object (e.g. budget reservation) still extract the
    model; the pass-through suppression only applies to a dispatched pass-through
    handler."""
    assert (
        get_model_from_request(
            request_data={"model": "gpt-4o"},
            route="/v1/chat/completions",
        )
        == "gpt-4o"
    )


def test_get_model_from_request_supports_google_model_names_with_slashes():
    assert (
        get_model_from_request(
            request_data={},
            route="/v1beta/models/bedrock/claude-sonnet-3.7:generateContent",
        )
        == "bedrock/claude-sonnet-3.7"
    )
    assert (
        get_model_from_request(
            request_data={},
            route="/models/hosted_vllm/gpt-oss-20b:generateContent",
        )
        == "hosted_vllm/gpt-oss-20b"
    )


def test_get_model_from_request_vertex_passthrough_still_works():
    route = "/vertex_ai/v1/projects/p/locations/l/publishers/google/models/gemini-1.5-pro:generateContent"
    assert get_model_from_request(request_data={}, route=route) == "gemini-1.5-pro"


def test_get_model_from_request_openai_deployment_route_still_works():
    assert (
        get_model_from_request(
            request_data={},
            route="/openai/deployments/my-azure-deployment/chat/completions",
        )
        == "my-azure-deployment"
    )


def test_get_model_from_request_includes_file_endpoint_header_model():
    assert (
        get_model_from_request(
            request_data={},
            route="/v1/files",
            request_headers={"X-LiteLLM-Model": "restricted-model"},
        )
        == "restricted-model"
    )


def test_get_model_from_request_ignores_routing_header_on_standard_llm_routes():
    assert (
        get_model_from_request(
            request_data={"model": "allowed-model"},
            route="/v1/chat/completions",
            request_headers={"x-litellm-model": "restricted-model"},
        )
        == "allowed-model"
    )


def test_get_model_from_request_authorizes_all_file_routing_model_sources():
    models = get_model_from_request(
        request_data={"model": "body-model"},
        route="/v1/files",
        request_headers={"x-litellm-model": "header-model"},
        request_query_params={"target_model_names": "query-model-a,query-model-b"},
    )
    assert isinstance(models, list)
    assert set(models) == {
        "body-model",
        "query-model-a",
        "query-model-b",
        "header-model",
    }


def test_get_model_from_request_extracts_simple_encoded_file_id_model():
    from litellm.proxy.openai_files_endpoints.common_utils import (
        encode_file_id_with_model,
    )

    file_id = encode_file_id_with_model(
        file_id="file-provider-id",
        model="restricted-model",
    )

    assert (
        get_model_from_request(
            request_data={"file_id": file_id},
            route="/v1/files/{file_id}",
        )
        == "restricted-model"
    )


def test_get_model_from_request_extracts_unified_file_id_models():
    raw_unified_file_id = (
        "litellm_proxy:application/octet-stream;unified_id,test-id;"
        "target_model_names,model-a,model-b;llm_output_file_id,file-provider-id"
    )
    encoded_unified_file_id = (
        base64.urlsafe_b64encode(raw_unified_file_id.encode()).decode().rstrip("=")
    )

    assert get_model_from_request(
        request_data={"file_id": encoded_unified_file_id},
        route="/v1/files/{file_id}",
    ) == ["model-a", "model-b"]


def test_get_model_from_request_extracts_eval_completion_model():
    assert (
        get_model_from_request(
            request_data={"completion": {"model": "judge-model"}},
            route="/v1/evals/{eval_id}/runs",
        )
        == "judge-model"
    )


def test_get_model_from_request_includes_fine_tuning_target_model_query():
    assert (
        get_model_from_request(
            request_data={},
            route="/v1/fine_tuning/jobs",
            request_query_params={"target_model_names": "fine-tune-model"},
        )
        == "fine-tune-model"
    )


def test_get_model_from_request_extracts_video_id_model():
    from litellm.types.videos.utils import encode_video_id_with_provider

    video_id = encode_video_id_with_provider(
        video_id="video-provider-id",
        provider="openai",
        model_id="video-model",
    )

    assert (
        get_model_from_request(
            request_data={"video_id": video_id},
            route="/v1/videos/{video_id}",
        )
        == "video-model"
    )


def test_get_model_from_request_resolves_video_id_model_with_router():
    from litellm.types.videos.utils import encode_video_id_with_provider

    provider_video_id = (
        "projects/test-project/locations/us-central1/publishers/google/models/"
        "veo-3.1-generate-001/operations/operation-id"
    )
    video_id = encode_video_id_with_provider(
        video_id=provider_video_id,
        provider="vertex_ai",
        model_id="veo-3.1-generate-001",
    )
    llm_router = MagicMock()
    llm_router.resolve_model_name_from_model_id.return_value = (
        "gcp/google/veo-3.1-generate-001"
    )

    assert (
        get_model_from_request(
            request_data={"video_id": video_id},
            route="/v1/videos/{video_id}",
            llm_router=llm_router,
        )
        == "gcp/google/veo-3.1-generate-001"
    )
    llm_router.resolve_model_name_from_model_id.assert_called_once_with(
        "veo-3.1-generate-001"
    )


_BATCH_DEPLOYMENT_ID = "8d0eaa7e6c6f54a425dfd0062cb6b0dc"


def _managed_batch_router():
    from litellm.router import Router

    return Router(
        model_list=[
            {
                "model_name": "bedrock-batch-model",
                "litellm_params": {
                    "model": "bedrock/global.anthropic.claude-haiku-4-5-20251001-v1:0",
                },
                "model_info": {"id": _BATCH_DEPLOYMENT_ID},
            },
            {
                "model_name": "some-other-model",
                "litellm_params": {"model": "openai/gpt-4o-mini", "api_key": "test-key"},
                "model_info": {"id": "a-different-deployment-id"},
            },
        ]
    )


def _encode_managed_id(decoded: str) -> str:
    return base64.urlsafe_b64encode(decoded.encode()).decode().rstrip("=")


_MANAGED_BATCH_ID = _encode_managed_id(
    f"litellm_proxy;model_id:{_BATCH_DEPLOYMENT_ID};llm_batch_id:provider-batch-123"
)
_MANAGED_BATCH_OUTPUT_FILE_ID = _encode_managed_id(
    f"litellm_proxy;model_id:{_BATCH_DEPLOYMENT_ID};llm_batch_id:provider-batch-123;"
    "llm_output_file_id:provider-file-456"
)


@pytest.mark.parametrize(
    "route, request_data",
    [
        ("/v1/batches/{batch_id}", {"batch_id": _MANAGED_BATCH_ID}),
        ("/v1/batches/{batch_id}/cancel", {"batch_id": _MANAGED_BATCH_ID}),
        ("/v1/files/{file_id}", {"file_id": _MANAGED_BATCH_OUTPUT_FILE_ID}),
        ("/v1/files/{file_id}/content", {"file_id": _MANAGED_BATCH_OUTPUT_FILE_ID}),
    ],
)
def test_get_model_from_request_resolves_batch_id_deployment_to_model_name(route, request_data):
    """Regression for #32580: managed batch retrieve/cancel and managed batch output
    file reads encode the deployment model_id into the resource id. The auth layer must
    resolve that id back to the public model group name so model-access checks compare
    against the model group, not the raw deployment id."""
    assert (
        get_model_from_request(
            request_data=request_data,
            route=route,
            llm_router=_managed_batch_router(),
        )
        == "bedrock-batch-model"
    )


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "route, request_data",
    [
        ("/v1/batches/{batch_id}", {"batch_id": _MANAGED_BATCH_ID}),
        ("/v1/batches/{batch_id}/cancel", {"batch_id": _MANAGED_BATCH_ID}),
        ("/v1/files/{file_id}/content", {"file_id": _MANAGED_BATCH_OUTPUT_FILE_ID}),
    ],
)
async def test_managed_batch_routes_pass_team_model_access_check(route, request_data):
    """End-to-end regression for #32580: a team scoped to the batch model group got
    ``team_model_access_denied`` on retrieve/cancel because the deployment id, not the
    model group, was authorized. Fails pre-fix with the deployment id in the message."""
    from litellm.proxy._types import LiteLLM_TeamTable
    from litellm.proxy.auth.auth_checks import can_team_access_model

    llm_router = _managed_batch_router()
    model = get_model_from_request(request_data=request_data, route=route, llm_router=llm_router)

    assert (
        await can_team_access_model(
            model=model,
            team_object=LiteLLM_TeamTable(team_id="team-batch", models=["bedrock-batch-model"]),
            llm_router=llm_router,
        )
        is True
    )

    with pytest.raises(Exception, match="team not allowed to access model"):
        await can_team_access_model(
            model=model,
            team_object=LiteLLM_TeamTable(team_id="team-other", models=["some-other-model"]),
            llm_router=llm_router,
        )


def test_get_model_from_request_resolves_character_id_model_with_router():
    from litellm.types.videos.utils import encode_character_id_with_provider

    character_id = encode_character_id_with_provider(
        character_id="character-provider-id",
        provider="vertex_ai",
        model_id="veo-3.1-generate-001",
    )
    llm_router = MagicMock()
    llm_router.resolve_model_name_from_model_id.return_value = (
        "gcp/google/veo-3.1-generate-001"
    )

    assert (
        get_model_from_request(
            request_data={"character_id": character_id},
            route="/v1/videos/characters/{character_id}",
            llm_router=llm_router,
        )
        == "gcp/google/veo-3.1-generate-001"
    )
    llm_router.resolve_model_name_from_model_id.assert_called_once_with(
        "veo-3.1-generate-001"
    )


def test_get_model_from_request_only_runs_media_decoders_for_matching_fields():
    with (
        patch(
            "litellm.types.videos.utils.decode_video_id_with_provider",
            return_value={"model_id": "video-model"},
        ) as video_decoder,
        patch(
            "litellm.types.videos.utils.decode_character_id_with_provider",
            return_value={"model_id": "character-model"},
        ) as character_decoder,
    ):
        assert (
            get_model_from_request(
                request_data={"file_id": "file-provider-id"},
                route="/v1/files/{file_id}",
            )
            is None
        )
        video_decoder.assert_not_called()
        character_decoder.assert_not_called()

        assert (
            get_model_from_request(
                request_data={"video_id": "video-provider-id"},
                route="/v1/videos/{video_id}",
            )
            == "video-model"
        )
        video_decoder.assert_called_once_with("video-provider-id")
        character_decoder.assert_not_called()

        video_decoder.reset_mock()
        character_decoder.reset_mock()
        assert (
            get_model_from_request(
                request_data={"character_id": "character-provider-id"},
                route="/v1/videos/{character_id}",
            )
            == "character-model"
        )
        video_decoder.assert_not_called()
        character_decoder.assert_called_once_with("character-provider-id")


def test_get_model_from_request_handles_managed_id_decoder_failures():
    with (
        patch(
            "litellm.proxy.openai_files_endpoints.common_utils.decode_model_from_file_id",
            side_effect=Exception("decode failed"),
        ),
        patch(
            "litellm.llms.base_llm.managed_resources.utils.parse_unified_id",
            side_effect=Exception("parse failed"),
        ),
        patch(
            "litellm.types.videos.utils.decode_video_id_with_provider",
            side_effect=Exception("video decode failed"),
        ),
    ):
        assert (
            get_model_from_request(
                request_data={"file_id": "not-a-managed-resource-id"},
                route="/v1/files/{file_id}",
            )
            is None
        )
        assert (
            get_model_from_request(
                request_data={"video_id": "not-a-managed-resource-id"},
                route="/v1/videos/{video_id}",
            )
            is None
        )


@pytest.mark.parametrize(
    "route",
    [
        "/realtime/client_secrets",
        "/v1/realtime/client_secrets",
        "/openai/v1/realtime/client_secrets",
        "/realtime/calls",
        "/v1/realtime/calls",
        "/openai/v1/realtime/calls",
    ],
)
def test_get_model_from_request_extracts_realtime_session_model(route):
    """The effective realtime model lives in ``session.model`` (not the
    top-level ``model``). It must be surfaced so can_key_call_model() can
    validate the model a restricted key is actually requesting.

    Regression test for the model-access bypass on the GA Realtime WebRTC
    HTTP routes (https://github.com/BerriAI/litellm/issues/29923).
    """
    assert (
        get_model_from_request(
            request_data={"session": {"type": "realtime", "model": "gpt-realtime"}},
            route=route,
        )
        == "gpt-realtime"
    )


def test_get_model_from_request_realtime_includes_top_level_and_session_model():
    """When both top-level and session model are present, both are returned so
    neither path can smuggle a disallowed model past the model-access check."""
    models = get_model_from_request(
        request_data={
            "model": "gpt-4o-realtime-preview",
            "session": {"type": "realtime", "model": "gpt-realtime"},
        },
        route="/v1/realtime/client_secrets",
    )
    assert models == ["gpt-4o-realtime-preview", "gpt-realtime"]


def test_get_model_from_request_ignores_session_model_on_non_realtime_routes():
    """A nested ``session.model`` must not leak into model resolution for
    unrelated routes."""
    assert (
        get_model_from_request(
            request_data={"session": {"type": "realtime", "model": "gpt-realtime"}},
            route="/v1/chat/completions",
        )
        is None
    )


def test_abbreviate_api_key():
    assert abbreviate_api_key("sk-test-1234-abcdefgh") == "sk-...efgh"
    assert abbreviate_api_key("sk-abcdefghijklm") == "sk-...jklm"


def test_abbreviate_api_key_short_key_is_fully_masked():
    """Regression test for LIT-4355: for keys shorter than the enforced minimum,
    showing the last 4 characters can reveal the entire key (sk-1234 -> sk-...1234)."""
    assert abbreviate_api_key("sk-1234") == "sk-..."
    assert abbreviate_api_key("sk-test-1234") == "sk-..."
    assert abbreviate_api_key("") == "sk-..."


def test_get_customer_user_header_returns_none_when_no_customer_role():
    from litellm.proxy.auth.auth_utils import get_customer_user_header_from_mapping

    mappings = [
        {"header_name": "X-OpenWebUI-User-Id", "litellm_user_role": "internal_user"}
    ]
    result = get_customer_user_header_from_mapping(mappings)
    assert result is None


def test_get_customer_user_header_returns_none_for_single_non_customer_mapping():
    from litellm.proxy.auth.auth_utils import get_customer_user_header_from_mapping

    mapping = {"header_name": "X-Only-Internal", "litellm_user_role": "internal_user"}
    result = get_customer_user_header_from_mapping(mapping)
    assert result is None


def test_get_customer_user_header_from_mapping_returns_customer_header():
    from litellm.proxy.auth.auth_utils import get_customer_user_header_from_mapping

    mappings = [
        {"header_name": "X-OpenWebUI-User-Id", "litellm_user_role": "internal_user"},
        {"header_name": "X-OpenWebUI-User-Email", "litellm_user_role": "customer"},
    ]
    result = get_customer_user_header_from_mapping(mappings)
    assert result == ["x-openwebui-user-email"]


def test_get_customer_user_header_returns_customers_header_in_config_order_when_multiple_exist():
    from litellm.proxy.auth.auth_utils import get_customer_user_header_from_mapping

    mappings = [
        {"header_name": "X-OpenWebUI-User-Id", "litellm_user_role": "internal_user"},
        {"header_name": "X-OpenWebUI-User-Email", "litellm_user_role": "customer"},
        {"header_name": "X-User-Id", "litellm_user_role": "customer"},
    ]
    result = get_customer_user_header_from_mapping(mappings)
    assert result == ["x-openwebui-user-email", "x-user-id"]


def test_get_end_user_id_returns_id_from_user_header_mappings():
    from litellm.proxy.auth.auth_utils import get_end_user_id_from_request_body

    mappings = [
        {"header_name": "x-openwebui-user-id", "litellm_user_role": "internal_user"},
        {"header_name": "x-openwebui-user-email", "litellm_user_role": "customer"},
    ]
    general_settings = {"user_header_mappings": mappings}
    headers = {"x-openwebui-user-email": "1234"}

    with (
        patch(
            "litellm.proxy.auth.auth_utils._get_customer_id_from_standard_headers",
            return_value=None,
        ),
        patch("litellm.proxy.proxy_server.general_settings", general_settings),
    ):
        result = get_end_user_id_from_request_body(
            request_body={}, request_headers=headers
        )

    assert result == "1234"


def test_get_end_user_id_returns_first_customer_header_when_multiple_mappings_exist():
    from litellm.proxy.auth.auth_utils import get_end_user_id_from_request_body

    mappings = [
        {"header_name": "x-openwebui-user-id", "litellm_user_role": "internal_user"},
        {"header_name": "x-user-id", "litellm_user_role": "customer"},
        {"header_name": "x-openwebui-user-email", "litellm_user_role": "customer"},
    ]
    general_settings = {"user_header_mappings": mappings}
    headers = {
        "x-user-id": "user-456",
        "x-openwebui-user-email": "user@example.com",
    }

    with (
        patch(
            "litellm.proxy.auth.auth_utils._get_customer_id_from_standard_headers",
            return_value=None,
        ),
        patch("litellm.proxy.proxy_server.general_settings", general_settings),
    ):
        result = get_end_user_id_from_request_body(
            request_body={}, request_headers=headers
        )

    assert result == "user-456"


def test_get_end_user_id_returns_none_when_no_customer_role_in_mappings():
    from litellm.proxy.auth.auth_utils import get_end_user_id_from_request_body

    mappings = [
        {"header_name": "x-openwebui-user-id", "litellm_user_role": "internal_user"},
    ]
    general_settings = {"user_header_mappings": mappings}
    headers = {"x-openwebui-user-id": "user-789"}

    with (
        patch(
            "litellm.proxy.auth.auth_utils._get_customer_id_from_standard_headers",
            return_value=None,
        ),
        patch("litellm.proxy.proxy_server.general_settings", general_settings),
    ):
        result = get_end_user_id_from_request_body(
            request_body={}, request_headers=headers
        )

    assert result is None


def test_get_end_user_id_falls_back_to_deprecated_user_header_name():
    from litellm.proxy.auth.auth_utils import get_end_user_id_from_request_body

    general_settings = {"user_header_name": "x-custom-user-id"}
    headers = {"x-custom-user-id": "user-legacy"}

    with (
        patch(
            "litellm.proxy.auth.auth_utils._get_customer_id_from_standard_headers",
            return_value=None,
        ),
        patch("litellm.proxy.proxy_server.general_settings", general_settings),
    ):
        result = get_end_user_id_from_request_body(
            request_body={}, request_headers=headers
        )

    assert result == "user-legacy"


class TestCoerceUserIdToStr:
    """Unit tests for the _coerce_user_id_to_str helper."""

    def test_plain_string_is_returned_verbatim(self):
        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str

        assert _coerce_user_id_to_str("alice@example.com") == "alice@example.com"

    def test_string_is_stripped(self):
        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str

        assert _coerce_user_id_to_str("  bob  ") == "bob"

    def test_codex_opaque_identifier_is_preserved(self):
        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str

        codex_id = (
            "user_8a4a360c36621665b341e06fb76041d9b6def732bb183eea148d4abc9d97c1de"
            "_account__session_a2bce4a5-8887-44ef-b491-fbf0a55c6569"
        )
        assert _coerce_user_id_to_str(codex_id) == codex_id

    def test_none_returns_none(self):
        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str

        assert _coerce_user_id_to_str(None) is None

    def test_empty_string_returns_none(self):
        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str

        assert _coerce_user_id_to_str("") is None
        assert _coerce_user_id_to_str("   ") is None

    def test_dict_returns_none(self):
        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str

        payload = {
            "device_id": "abc",
            "account_uuid": "",
            "session_id": "c284b8cb",
        }
        assert _coerce_user_id_to_str(payload) is None

    def test_list_returns_none(self):
        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str

        assert _coerce_user_id_to_str(["a", "b"]) is None

    def test_json_encoded_dict_string_passes_through_by_default(self):
        """JSON-encoded dict strings are preserved unless opt-in flag is on.

        This preserves backwards compatibility: existing deployments that
        intentionally pass JSON-encoded user identifiers keep working.
        """
        import litellm
        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str

        blob = (
            '{"device_id":"d5abe9199ee7759a0558974e9371e78c7b38d7621aae26d6609c1de61af6afb0",'
            '"account_uuid":"","session_id":"c284b8cb-a050-4278-8599-cc4e016a10ab"}'
        )
        original = litellm.validate_end_user_id_in_db
        litellm.validate_end_user_id_in_db = False
        try:
            assert _coerce_user_id_to_str(blob) == blob
        finally:
            litellm.validate_end_user_id_in_db = original

    def test_json_encoded_dict_string_returns_none_when_validation_enabled(self):
        import litellm
        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str

        # Same broken shape we saw in spend logs, but pre-stringified to JSON.
        blob = (
            '{"device_id":"d5abe9199ee7759a0558974e9371e78c7b38d7621aae26d6609c1de61af6afb0",'
            '"account_uuid":"","session_id":"c284b8cb-a050-4278-8599-cc4e016a10ab"}'
        )
        original = litellm.validate_end_user_id_in_db
        litellm.validate_end_user_id_in_db = True
        try:
            assert _coerce_user_id_to_str(blob) is None
        finally:
            litellm.validate_end_user_id_in_db = original

    def test_json_encoded_list_string_passes_through_by_default(self):
        import litellm
        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str

        original = litellm.validate_end_user_id_in_db
        litellm.validate_end_user_id_in_db = False
        try:
            assert _coerce_user_id_to_str('["a","b"]') == '["a","b"]'
        finally:
            litellm.validate_end_user_id_in_db = original

    def test_json_encoded_list_string_returns_none_when_validation_enabled(self):
        import litellm
        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str

        original = litellm.validate_end_user_id_in_db
        litellm.validate_end_user_id_in_db = True
        try:
            assert _coerce_user_id_to_str('["a","b"]') is None
        finally:
            litellm.validate_end_user_id_in_db = original

    def test_int_returns_str(self):
        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str

        assert _coerce_user_id_to_str(12345) == "12345"

    def test_bool_returns_none(self):
        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str

        # bool is an int subclass — reject explicitly, never produce "True"/"False".
        assert _coerce_user_id_to_str(True) is None
        assert _coerce_user_id_to_str(False) is None

    def test_brace_string_that_isnt_json_is_kept(self):
        """A string starting with `{` but failing to parse stays as-is."""
        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str

        assert _coerce_user_id_to_str("{not json") == "{not json"


class TestGetEndUserIdDropsMalformedBodyValues:
    """Tests that get_end_user_id_from_request_body drops dict-shaped values
    rather than stringifying them into spend logs."""

    def test_dict_user_falls_through_to_litellm_metadata(self):
        request_body = {
            "user": {
                "device_id": "abc",
                "session_id": "c284b8cb",
            },
            "litellm_metadata": {"user": "alice@example.com"},
        }

        with patch("litellm.proxy.proxy_server.general_settings", {}):
            result = get_end_user_id_from_request_body(
                request_body=request_body, request_headers={}
            )

        assert result == "alice@example.com"

    def test_dict_user_with_no_other_sources_returns_none(self):
        request_body = {
            "user": {"device_id": "abc", "session_id": "xyz"},
        }

        with patch("litellm.proxy.proxy_server.general_settings", {}):
            result = get_end_user_id_from_request_body(
                request_body=request_body, request_headers={}
            )

        assert result is None

    def test_json_encoded_user_string_passes_through_by_default(self):
        """JSON-encoded user strings pass through unless validation is opted in.

        Gating behind ``litellm.validate_end_user_id_in_db`` keeps existing
        deployments that send JSON-encoded identifiers working until they
        explicitly opt into the stricter extraction.
        """
        import litellm

        blob = (
            '{"device_id":"d5abe9199ee7759a","account_uuid":"",'
            '"session_id":"c284b8cb-a050-4278-8599-cc4e016a10ab"}'
        )
        request_body = {"user": blob}

        original = litellm.validate_end_user_id_in_db
        litellm.validate_end_user_id_in_db = False
        try:
            with patch("litellm.proxy.proxy_server.general_settings", {}):
                result = get_end_user_id_from_request_body(
                    request_body=request_body, request_headers={}
                )
        finally:
            litellm.validate_end_user_id_in_db = original

        assert result == blob

    def test_json_encoded_user_string_returns_none_when_validation_enabled(self):
        import litellm

        request_body = {
            "user": (
                '{"device_id":"d5abe9199ee7759a","account_uuid":"",'
                '"session_id":"c284b8cb-a050-4278-8599-cc4e016a10ab"}'
            ),
        }

        original = litellm.validate_end_user_id_in_db
        litellm.validate_end_user_id_in_db = True
        try:
            with patch("litellm.proxy.proxy_server.general_settings", {}):
                result = get_end_user_id_from_request_body(
                    request_body=request_body, request_headers={}
                )
        finally:
            litellm.validate_end_user_id_in_db = original

        assert result is None

    def test_plain_string_user_is_preserved(self):
        request_body = {"user": "alice@example.com"}

        with patch("litellm.proxy.proxy_server.general_settings", {}):
            result = get_end_user_id_from_request_body(
                request_body=request_body, request_headers={}
            )

        assert result == "alice@example.com"

    def test_codex_opaque_user_is_preserved(self):
        codex_id = (
            "user_8a4a360c36621665b341e06fb76041d9b6def732bb183eea148d4abc9d97c1de"
            "_account__session_a2bce4a5-8887-44ef-b491-fbf0a55c6569"
        )
        request_body = {"user": codex_id}

        with patch("litellm.proxy.proxy_server.general_settings", {}):
            result = get_end_user_id_from_request_body(
                request_body=request_body, request_headers={}
            )

        assert result == codex_id

    def test_int_user_is_coerced_to_string(self):
        request_body = {"user": 12345}

        with patch("litellm.proxy.proxy_server.general_settings", {}):
            result = get_end_user_id_from_request_body(
                request_body=request_body, request_headers={}
            )

        assert result == "12345"

    def test_list_user_falls_through(self):
        request_body = {
            "user": ["a", "b"],
            "safety_identifier": "alice@example.com",
        }

        with patch("litellm.proxy.proxy_server.general_settings", {}):
            result = get_end_user_id_from_request_body(
                request_body=request_body, request_headers={}
            )

        assert result == "alice@example.com"

    def test_dict_safety_identifier_returns_none(self):
        request_body = {
            "safety_identifier": {"device_id": "abc"},
        }

        with patch("litellm.proxy.proxy_server.general_settings", {}):
            result = get_end_user_id_from_request_body(
                request_body=request_body, request_headers={}
            )

        assert result is None

    def test_dict_metadata_user_id_returns_none(self):
        request_body = {
            "metadata": {"user_id": {"device_id": "abc"}},
        }

        with patch("litellm.proxy.proxy_server.general_settings", {}):
            result = get_end_user_id_from_request_body(
                request_body=request_body, request_headers={}
            )

        assert result is None

    def test_whitespace_user_falls_through(self):
        request_body = {"user": "   ", "safety_identifier": "alice@example.com"}

        with patch("litellm.proxy.proxy_server.general_settings", {}):
            result = get_end_user_id_from_request_body(
                request_body=request_body, request_headers={}
            )

        assert result == "alice@example.com"

    def test_dict_user_header_falls_through_to_body(self):
        """A dict-shaped value in a configured user-id header is dropped, not stringified."""
        general_settings = {"user_header_name": "x-custom-user-id"}
        # A header value will normally be a str, but be defensive: the coercion
        # must drop anything that isn't a usable identifier.
        headers = {"x-custom-user-id": {"device_id": "abc"}}
        request_body = {"user": "alice@example.com"}

        with (
            patch(
                "litellm.proxy.auth.auth_utils._get_customer_id_from_standard_headers",
                return_value=None,
            ),
            patch("litellm.proxy.proxy_server.general_settings", general_settings),
        ):
            result = get_end_user_id_from_request_body(
                request_body=request_body, request_headers=headers
            )

        assert result == "alice@example.com"


def _make_deployment_dict(
    model_name: str, tpm: Optional[int] = None, rpm: Optional[int] = None
) -> dict:
    """Helper to build a minimal deployment dict as returned by router.get_model_list."""
    litellm_params: dict = {"model": model_name}
    if tpm is not None:
        litellm_params["default_api_key_tpm_limit"] = tpm
    if rpm is not None:
        litellm_params["default_api_key_rpm_limit"] = rpm
    return {"model_name": model_name, "litellm_params": litellm_params}


_ROUTER_PATCH = "litellm.proxy.proxy_server.llm_router"


class TestDeploymentDefaultRpmLimit:
    """Tests for deployment default_api_key_rpm_limit fallback in get_key_model_rpm_limit."""

    def test_returns_deployment_default_when_key_has_no_limits(self):
        """Case 2 from spec: key has no model-specific limits, falls back to deployment default."""
        user_api_key_dict = UserAPIKeyAuth(api_key="sk-123")
        mock_router = MagicMock()
        mock_router.get_model_list.return_value = [
            _make_deployment_dict("model1", rpm=200)
        ]
        with patch(_ROUTER_PATCH, mock_router):
            result = get_key_model_rpm_limit(user_api_key_dict, model_name="model1")
        assert result == {"model1": 200}

    def test_key_model_limit_takes_priority_over_deployment_default(self):
        """Case 1 from spec: key model-specific limit wins over deployment default."""
        user_api_key_dict = UserAPIKeyAuth(
            api_key="sk-123",
            metadata={"model_rpm_limit": {"model1": 10}},
        )
        mock_router = MagicMock()
        mock_router.get_model_list.return_value = [
            _make_deployment_dict("model1", rpm=200)
        ]
        with patch(_ROUTER_PATCH, mock_router):
            result = get_key_model_rpm_limit(user_api_key_dict, model_name="model1")
        assert result == {"model1": 10}

    def test_returns_none_when_no_deployment_default_and_no_key_limits(self):
        """Returns None when neither the key nor the deployment has any rpm limit."""
        user_api_key_dict = UserAPIKeyAuth(api_key="sk-123")
        mock_router = MagicMock()
        mock_router.get_model_list.return_value = [
            _make_deployment_dict("model1")  # no rpm default
        ]
        with patch(_ROUTER_PATCH, mock_router):
            result = get_key_model_rpm_limit(user_api_key_dict, model_name="model1")
        assert result is None

    def test_returns_none_without_model_name_even_when_deployment_has_default(self):
        """No model_name means deployment fallback is skipped."""
        user_api_key_dict = UserAPIKeyAuth(api_key="sk-123")
        mock_router = MagicMock()
        mock_router.get_model_list.return_value = [
            _make_deployment_dict("model1", rpm=200)
        ]
        with patch(_ROUTER_PATCH, mock_router):
            result = get_key_model_rpm_limit(user_api_key_dict)
        assert result is None

    def test_returns_none_when_llm_router_is_none(self):
        """No router means deployment fallback returns None gracefully."""
        user_api_key_dict = UserAPIKeyAuth(api_key="sk-123")
        with patch(_ROUTER_PATCH, None):
            result = get_key_model_rpm_limit(user_api_key_dict, model_name="model1")
        assert result is None

    def test_returns_minimum_across_multiple_deployments(self):
        """When multiple deployments share a model name, the minimum rpm limit is used."""
        user_api_key_dict = UserAPIKeyAuth(api_key="sk-123")
        mock_router = MagicMock()
        mock_router.get_model_list.return_value = [
            _make_deployment_dict("model1", rpm=200),
            _make_deployment_dict("model1", rpm=50),
            _make_deployment_dict("model1", rpm=150),
        ]
        with patch(_ROUTER_PATCH, mock_router):
            result = get_key_model_rpm_limit(user_api_key_dict, model_name="model1")
        assert result == {"model1": 50}

    def test_ignores_deployments_without_default_when_others_have_it(self):
        """Deployments missing the field are skipped; min is taken over those that have it."""
        user_api_key_dict = UserAPIKeyAuth(api_key="sk-123")
        mock_router = MagicMock()
        mock_router.get_model_list.return_value = [
            _make_deployment_dict("model1"),  # no rpm default
            _make_deployment_dict("model1", rpm=75),
        ]
        with patch(_ROUTER_PATCH, mock_router):
            result = get_key_model_rpm_limit(user_api_key_dict, model_name="model1")
        assert result == {"model1": 75}

    def test_skips_deployments_with_malformed_limit_value(self):
        """Deployments with non-integer-parseable limit values are skipped without raising."""
        user_api_key_dict = UserAPIKeyAuth(api_key="sk-123")
        mock_router = MagicMock()
        mock_router.get_model_list.return_value = [
            {
                "model_name": "model1",
                "litellm_params": {"default_api_key_rpm_limit": "not-a-number"},
            },
            _make_deployment_dict("model1", rpm=100),
        ]
        with patch(_ROUTER_PATCH, mock_router):
            result = get_key_model_rpm_limit(user_api_key_dict, model_name="model1")
        # The malformed deployment is skipped; the valid one provides 100
        assert result == {"model1": 100}


class TestDeploymentDefaultTpmLimit:
    """Tests for deployment default_api_key_tpm_limit fallback in get_key_model_tpm_limit."""

    def test_returns_deployment_default_when_key_has_no_limits(self):
        """Case 2 from spec: key has no model-specific limits, falls back to deployment default."""
        user_api_key_dict = UserAPIKeyAuth(api_key="sk-123")
        mock_router = MagicMock()
        mock_router.get_model_list.return_value = [
            _make_deployment_dict("model1", tpm=100)
        ]
        with patch(_ROUTER_PATCH, mock_router):
            result = get_key_model_tpm_limit(user_api_key_dict, model_name="model1")
        assert result == {"model1": 100}

    def test_key_model_limit_takes_priority_over_deployment_default(self):
        """Case 1 from spec: key model-specific limit wins over deployment default."""
        user_api_key_dict = UserAPIKeyAuth(
            api_key="sk-123",
            metadata={"model_tpm_limit": {"model1": 20}},
        )
        mock_router = MagicMock()
        mock_router.get_model_list.return_value = [
            _make_deployment_dict("model1", tpm=100)
        ]
        with patch(_ROUTER_PATCH, mock_router):
            result = get_key_model_tpm_limit(user_api_key_dict, model_name="model1")
        assert result == {"model1": 20}

    def test_returns_none_when_no_deployment_default_and_no_key_limits(self):
        """Returns None when neither the key nor the deployment has any tpm limit."""
        user_api_key_dict = UserAPIKeyAuth(api_key="sk-123")
        mock_router = MagicMock()
        mock_router.get_model_list.return_value = [
            _make_deployment_dict("model1")  # no tpm default
        ]
        with patch(_ROUTER_PATCH, mock_router):
            result = get_key_model_tpm_limit(user_api_key_dict, model_name="model1")
        assert result is None

    def test_returns_none_without_model_name_even_when_deployment_has_default(self):
        """No model_name means deployment fallback is skipped."""
        user_api_key_dict = UserAPIKeyAuth(api_key="sk-123")
        mock_router = MagicMock()
        mock_router.get_model_list.return_value = [
            _make_deployment_dict("model1", tpm=100)
        ]
        with patch(_ROUTER_PATCH, mock_router):
            result = get_key_model_tpm_limit(user_api_key_dict)
        assert result is None

    def test_returns_none_when_llm_router_is_none(self):
        """No router means deployment fallback returns None gracefully."""
        user_api_key_dict = UserAPIKeyAuth(api_key="sk-123")
        with patch(_ROUTER_PATCH, None):
            result = get_key_model_tpm_limit(user_api_key_dict, model_name="model1")
        assert result is None

    def test_returns_minimum_across_multiple_deployments(self):
        """When multiple deployments share a model name, the minimum tpm limit is used."""
        user_api_key_dict = UserAPIKeyAuth(api_key="sk-123")
        mock_router = MagicMock()
        mock_router.get_model_list.return_value = [
            _make_deployment_dict("model1", tpm=1000),
            _make_deployment_dict("model1", tpm=300),
            _make_deployment_dict("model1", tpm=700),
        ]
        with patch(_ROUTER_PATCH, mock_router):
            result = get_key_model_tpm_limit(user_api_key_dict, model_name="model1")
        assert result == {"model1": 300}

    def test_ignores_deployments_without_default_when_others_have_it(self):
        """Deployments missing the field are skipped; min is taken over those that have it."""
        user_api_key_dict = UserAPIKeyAuth(api_key="sk-123")
        mock_router = MagicMock()
        mock_router.get_model_list.return_value = [
            _make_deployment_dict("model1"),  # no tpm default
            _make_deployment_dict("model1", tpm=400),
        ]
        with patch(_ROUTER_PATCH, mock_router):
            result = get_key_model_tpm_limit(user_api_key_dict, model_name="model1")
        assert result == {"model1": 400}


class TestGetProjectModelRpmLimit:
    """Tests for get_project_model_rpm_limit function."""

    def test_returns_project_metadata_rpm_limit(self):
        user_api_key_dict = UserAPIKeyAuth(
            api_key="sk-123",
            project_metadata={"model_rpm_limit": {"gpt-4": 200}},
        )
        result = get_project_model_rpm_limit(user_api_key_dict)
        assert result == {"gpt-4": 200}

    def test_returns_none_when_no_project_metadata(self):
        user_api_key_dict = UserAPIKeyAuth(api_key="sk-123")
        result = get_project_model_rpm_limit(user_api_key_dict)
        assert result is None

    def test_returns_none_when_project_metadata_missing_key(self):
        user_api_key_dict = UserAPIKeyAuth(
            api_key="sk-123",
            project_metadata={"other_key": "value"},
        )
        result = get_project_model_rpm_limit(user_api_key_dict)
        assert result is None


class TestGetProjectModelTpmLimit:
    """Tests for get_project_model_tpm_limit function."""

    def test_returns_project_metadata_tpm_limit(self):
        user_api_key_dict = UserAPIKeyAuth(
            api_key="sk-123",
            project_metadata={"model_tpm_limit": {"gpt-4": 50000}},
        )
        result = get_project_model_tpm_limit(user_api_key_dict)
        assert result == {"gpt-4": 50000}

    def test_returns_none_when_no_project_metadata(self):
        user_api_key_dict = UserAPIKeyAuth(api_key="sk-123")
        result = get_project_model_tpm_limit(user_api_key_dict)
        assert result is None

    def test_returns_none_when_project_metadata_missing_key(self):
        user_api_key_dict = UserAPIKeyAuth(
            api_key="sk-123",
            project_metadata={"other_key": "value"},
        )
        result = get_project_model_tpm_limit(user_api_key_dict)
        assert result is None


class TestCheckCompleteCredentials:
    """Tests for the api_key validation in check_complete_credentials."""

    def test_returns_false_when_api_key_missing(self):
        result = check_complete_credentials({"model": "gpt-4"})
        assert result is False

    def test_returns_false_when_api_key_is_none(self):
        result = check_complete_credentials({"model": "gpt-4", "api_key": None})
        assert result is False

    def test_returns_false_when_api_key_is_empty_string(self):
        result = check_complete_credentials({"model": "gpt-4", "api_key": ""})
        assert result is False

    def test_returns_false_when_api_key_is_whitespace(self):
        result = check_complete_credentials({"model": "gpt-4", "api_key": "   "})
        assert result is False

    def test_returns_true_when_api_key_is_valid(self):
        result = check_complete_credentials({"model": "gpt-4", "api_key": "sk-valid"})
        assert result is True


class TestCheckCompleteCredentialsBlocksSSRF:
    """
    Even with credentials supplied, ``api_base`` / ``base_url`` must not
    point at private / internal / cloud-metadata addresses. Without this
    the gate accepts ``api_key=anything`` plus a malicious target and the
    proxy is used as an SSRF pivot.

    The check only runs when ``litellm.user_url_validation`` is True, so
    every test in this class flips the toggle. Tests stay mock-only — no
    real DNS is performed.
    """

    @pytest.fixture(autouse=True)
    def _enable_url_validation(self, monkeypatch):
        import litellm

        monkeypatch.setattr(litellm, "user_url_validation", True, raising=False)

    @pytest.mark.parametrize(
        "url_field",
        ["api_base", "base_url"],
    )
    @pytest.mark.parametrize(
        "blocked_url",
        [
            "http://169.254.169.254/latest/meta-data/iam/security-credentials/",
            "http://metadata.google.internal/computeMetadata/v1/",
            "http://127.0.0.1:8080/admin",
            "http://10.0.0.1/",
            "http://192.168.1.1/",
        ],
    )
    def test_rejects_private_or_metadata_targets(self, url_field, blocked_url):
        from litellm.litellm_core_utils.url_utils import SSRFError

        with patch(
            "litellm.proxy.auth.auth_utils.validate_url",
            side_effect=SSRFError(f"blocked: {blocked_url}"),
        ):
            with pytest.raises(ValueError, match='is rejected by the SSRF guard') as exc_info:
                check_complete_credentials(
                    {
                        "model": "gpt-4",
                        "api_key": "sk-some-clientside-key",
                        url_field: blocked_url,
                    }
                )
        assert url_field in str(exc_info.value)
        assert "SSRF" in str(exc_info.value)

    def test_allows_public_target_when_validate_url_passes(self):
        # ``validate_url`` is mocked so no real DNS is performed.
        with patch(
            "litellm.proxy.auth.auth_utils.validate_url",
            return_value=("https://api.openai.com/v1", "api.openai.com"),
        ):
            result = check_complete_credentials(
                {
                    "model": "gpt-4",
                    "api_key": "sk-some-clientside-key",
                    "api_base": "https://api.openai.com/v1",
                }
            )
        assert result is True

    def test_skips_url_validation_when_toggle_is_off(self, monkeypatch):
        # Admins who disable ``user_url_validation`` (default) should not
        # have requests rejected at the proxy boundary even if the URL
        # would fail the SSRF guard.
        import litellm

        monkeypatch.setattr(litellm, "user_url_validation", False, raising=False)
        with patch(
            "litellm.proxy.auth.auth_utils.validate_url",
        ) as mocked:
            result = check_complete_credentials(
                {
                    "model": "gpt-4",
                    "api_key": "sk-some-clientside-key",
                    "api_base": "http://127.0.0.1:8080/admin",
                }
            )
        assert result is True
        mocked.assert_not_called()


class TestGetDynamicLitellmParamsClearsAdminConfigOnBaseOverride:
    """
    When the caller redirects ``api_base`` / ``base_url`` to their own
    server, admin-set fields like ``OpenAI-Organization``, ``extra_body``,
    AWS / Vertex / Azure tokens, and per-deployment ``api_version`` must
    NOT flow through to that destination.
    """

    def test_clears_admin_organization_and_extra_body_on_base_override(self):
        from litellm.router_utils.clientside_credential_handler import (
            get_dynamic_litellm_params,
        )

        admin_params = {
            "model": "gpt-4",
            "api_key": "sk-admin-key",
            "api_base": "https://admin.upstream/v1",
            "organization": "org-admin-corp",
            "extra_body": {"x-admin-secret": "super-secret"},
            "api_version": "2026-04-01",
        }
        out = get_dynamic_litellm_params(
            litellm_params=dict(admin_params),
            request_kwargs={
                "api_key": "sk-attacker",
                "api_base": "https://attacker.example",
            },
        )
        assert out["api_base"] == "https://attacker.example"
        assert out["api_key"] == "sk-attacker"
        assert "organization" not in out
        assert "extra_body" not in out
        assert "api_version" not in out

    def test_clears_aws_and_vertex_secrets_on_base_override(self):
        from litellm.router_utils.clientside_credential_handler import (
            get_dynamic_litellm_params,
        )

        admin_params = {
            "model": "bedrock/claude-3",
            "aws_access_key_id": "AKIA-EXAMPLE",
            "aws_secret_access_key": "secret-example",
            "aws_session_token": "session-example",
            "vertex_credentials": '{"private_key":"-----BEGIN..."}',
            "vertex_project": "admin-gcp-project",
        }
        out = get_dynamic_litellm_params(
            litellm_params=dict(admin_params),
            request_kwargs={"base_url": "https://attacker.example", "api_key": "sk-caller"},
        )
        assert "aws_access_key_id" not in out
        assert "aws_secret_access_key" not in out
        assert "aws_session_token" not in out
        assert "vertex_credentials" not in out
        assert "vertex_project" not in out

    def test_clears_nvcf_function_id_on_base_override(self):
        from litellm.router_utils.clientside_credential_handler import (
            get_dynamic_litellm_params,
        )

        admin_params = {
            "model": "nvidia_riva/parakeet",
            "api_base": "grpc.nvcf.nvidia.com:443",
            "api_key": "nvapi-admin",
            "nvcf_function_id": "admin-pinned-function",
        }
        out = get_dynamic_litellm_params(
            litellm_params=dict(admin_params),
            request_kwargs={"api_base": "self-hosted.example.com:50051", "api_key": "sk-caller"},
        )
        assert out["api_base"] == "self-hosted.example.com:50051"
        assert "nvcf_function_id" not in out

    def test_clears_use_ssl_on_base_override(self):
        from litellm.router_utils.clientside_credential_handler import (
            get_dynamic_litellm_params,
        )

        admin_params = {
            "model": "nvidia_riva/parakeet",
            "api_base": "grpc.nvcf.nvidia.com:443",
            "api_key": "nvapi-admin",
            "use_ssl": True,
        }
        out = get_dynamic_litellm_params(
            litellm_params=dict(admin_params),
            request_kwargs={"api_base": "self-hosted.example.com:50051", "api_key": "sk-caller"},
        )
        assert out["api_base"] == "self-hosted.example.com:50051"
        assert "use_ssl" not in out

    def test_caller_resupplied_value_overrides_admin_value_on_base_override(self):
        # When the caller redirects ``api_base`` and *also* supplies their
        # own value for one of the admin fields (e.g. ``organization``),
        # the caller's value must win — never the admin's. The naive
        # ``if field not in request_kwargs: pop`` shape lets a caller echo
        # the field name with any value (or empty string) to keep the
        # admin's value forwarded, which is the exfiltration vector this
        # test guards against.
        from litellm.router_utils.clientside_credential_handler import (
            get_dynamic_litellm_params,
        )

        out = get_dynamic_litellm_params(
            litellm_params={
                "api_base": "https://admin.upstream/v1",
                "organization": "org-admin",
                "extra_body": {"admin": "value"},
            },
            request_kwargs={
                "api_base": "https://attacker.example",
                "api_key": "sk-caller",
                "organization": "org-attacker",
                "extra_body": {"attacker": "value"},
            },
        )
        assert out["organization"] == "org-attacker"
        assert out["extra_body"] == {"attacker": "value"}

    def test_field_echo_does_not_preserve_admin_value(self):
        # Regression: a caller that echoes an admin-config field name with
        # an *empty* value (or any value) must not be able to keep the
        # admin's value in ``litellm_params``.
        from litellm.router_utils.clientside_credential_handler import (
            get_dynamic_litellm_params,
        )

        out = get_dynamic_litellm_params(
            litellm_params={
                "api_base": "https://admin.upstream/v1",
                "organization": "org-admin-secret",
                "extra_body": {"x-admin-only": "secret"},
            },
            request_kwargs={
                "api_base": "https://attacker.example",
                "api_key": "sk-caller",
                "organization": "",
                "extra_body": "",
            },
        )
        assert out["organization"] == ""
        assert out["extra_body"] == ""
        assert "org-admin-secret" not in str(out)

    def test_no_clearing_when_only_api_key_overridden(self):
        from litellm.router_utils.clientside_credential_handler import (
            get_dynamic_litellm_params,
        )

        # Caller only overrides api_key (BYOK pattern); admin's organization /
        # extra_body / region still apply because the destination is unchanged.
        out = get_dynamic_litellm_params(
            litellm_params={
                "api_base": "https://admin.upstream/v1",
                "organization": "org-admin",
                "api_version": "2026-04-01",
            },
            request_kwargs={"api_key": "sk-byok"},
        )
        assert out["organization"] == "org-admin"
        assert out["api_version"] == "2026-04-01"
        assert out["api_base"] == "https://admin.upstream/v1"

    def test_client_api_key_used_when_supplied_with_base_override(self):
        from litellm.router_utils.clientside_credential_handler import (
            get_dynamic_litellm_params,
        )

        out = get_dynamic_litellm_params(
            litellm_params={
                "model": "gpt-4",
                "api_key": "sk-admin-secret",
                "api_base": "https://admin.upstream/v1",
            },
            request_kwargs={
                "api_base": "https://attacker.example",
                "api_key": "sk-client-byok",
            },
        )
        assert out["api_key"] == "sk-client-byok"
        assert "sk-admin-secret" not in str(out)


_OPENAI_CHAT_RESPONSE = {
    "id": "chatcmpl-x",
    "object": "chat.completion",
    "created": 1,
    "model": "gpt-4",
    "choices": [{"index": 0, "message": {"role": "assistant", "content": "hi"}, "finish_reason": "stop"}],
    "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
}


class TestClientsideBaseOverrideOutboundKey:
    """Drive a completion through the router and assert on the outbound request
    when the caller overrides ``api_base``."""

    def _router(self):
        from litellm import Router

        return Router(
            model_list=[
                {
                    "model_name": "gpt-4",
                    "litellm_params": {
                        "model": "openai/gpt-4",
                        "api_key": "sk-SERVER-CONFIG",
                        "api_base": "https://admin.upstream/v1",
                    },
                }
            ]
        )

    @pytest.fixture(autouse=True)
    def _ambient_server_key(self, monkeypatch):
        import litellm

        monkeypatch.setenv("OPENAI_API_KEY", "sk-SERVER-ENV")
        monkeypatch.setattr(litellm, "api_key", None, raising=False)

    def test_caller_key_override_sends_caller_key_never_server_key(self):
        import httpx
        import respx

        with respx.mock:
            route = respx.post("https://caller.example/v1/chat/completions").mock(
                return_value=httpx.Response(200, json=_OPENAI_CHAT_RESPONSE)
            )
            self._router().completion(
                model="gpt-4",
                messages=[{"role": "user", "content": "hi"}],
                api_base="https://caller.example/v1",
                api_key="sk-CALLER",
            )
            authorization = route.calls.last.request.headers.get("authorization")
            assert authorization == "Bearer sk-CALLER"
            assert "SERVER" not in (authorization or "")


def _rounds_deep_api_base_payload(rounds, field):
    """Build a fallbacks payload with ``api_base`` on a target nested ``rounds``
    fallback-rounds deep, each round wrapped in its own grouping dict."""
    node = {"model": "leaf", "api_base": "https://attacker.example"}
    for i in range(rounds):
        node = {"model": f"m{i}", field: [{"grp": [node]}]}
    return {"model": "gpt-4", field: [{"grp": [node]}]}


class TestIsRequestBodySafeBlocksFallbackSmuggle:
    """``is_request_body_safe`` runs the banned-param check on every dict target
    inside the fallback lists."""

    @pytest.fixture(autouse=True)
    def _disable_url_validation(self, monkeypatch):
        import litellm

        monkeypatch.setattr(litellm, "user_url_validation", False, raising=False)

    @pytest.mark.parametrize(
        "fallback_key",
        ["fallbacks", "context_window_fallbacks", "content_policy_fallbacks"],
    )
    def test_api_base_smuggled_via_nested_fallback_is_rejected(self, fallback_key):
        with pytest.raises(ValueError, match="api_base"):
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    fallback_key: [
                        {
                            "gpt-4": [
                                {"model": "evil", "api_base": "https://attacker.example"},
                            ]
                        }
                    ],
                },
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )

    def test_string_only_fallbacks_are_accepted(self):
        assert (
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    "fallbacks": [{"gpt-4": ["gpt-3.5-turbo", "claude-3-haiku"]}],
                },
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )
            is True
        )

    def test_benign_dict_fallback_entry_is_accepted(self):
        assert (
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    "fallbacks": [{"gpt-4": [{"model": "gpt-3.5-turbo"}]}],
                },
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )
            is True
        )

    def test_smuggled_fallback_allowed_under_proxy_wide_opt_in(self):
        assert (
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    "fallbacks": [
                        {"gpt-4": [{"model": "byok", "api_base": "https://my-byok.example"}]}
                    ],
                },
                general_settings={"allow_client_side_credentials": True},
                llm_router=None,
                model="gpt-4",
            )
            is True
        )

    @pytest.mark.parametrize(
        "fallback_field",
        ["fallbacks", "context_window_fallbacks", "content_policy_fallbacks"],
    )
    @pytest.mark.parametrize("surface", ["top_level", "router_settings_override"])
    def test_deeply_nested_api_base_smuggle_rejected_on_both_surfaces(self, fallback_field, surface):
        nested = [
            {
                "always-fail": [
                    {
                        "model": "x",
                        fallback_field: [
                            {"x": [{"model": "deepseek-chat", "api_base": "http://attacker"}]}
                        ],
                    }
                ]
            }
        ]
        request_body = {"model": "gpt-4"}
        if surface == "top_level":
            request_body[fallback_field] = nested
        else:
            request_body["router_settings_override"] = {fallback_field: nested}
        with pytest.raises(ValueError, match="api_base"):
            is_request_body_safe(
                request_body=request_body,
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )

    def test_router_settings_override_single_level_api_base_rejected(self):
        with pytest.raises(ValueError, match="api_base"):
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    "router_settings_override": {
                        "fallbacks": [{"gpt-4": [{"model": "x", "api_base": "http://attacker"}]}]
                    },
                },
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )

    def test_model_less_config_dict_api_base_rejected(self):
        with pytest.raises(ValueError, match="api_base"):
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    "fallbacks": [{"gpt-4": [{"api_base": "http://attacker"}]}],
                },
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )

    def test_nested_api_base_caught_across_router_fallback_rounds(self):
        """An ``api_base`` target nested ``ROUTER_MAX_FALLBACKS - 1`` rounds deep
        is still reached and rejected."""
        import litellm

        with pytest.raises(ValueError, match="api_base"):
            is_request_body_safe(
                request_body=_rounds_deep_api_base_payload(litellm.ROUTER_MAX_FALLBACKS - 1, "fallbacks"),
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )

    def test_grouping_only_deep_chain_is_rejected_at_depth_limit(self):
        """A deep grouping-only chain (``{"g": [{"g": [...]}]}``) is rejected at the
        validation-depth limit rather than accepted or raising RecursionError."""
        node: object = ["safe-model"]
        for _ in range(5000):
            node = [{"grp": node}]
        with pytest.raises(ValueError, match="depth"):
            is_request_body_safe(
                request_body={"model": "gpt-4", "fallbacks": node},
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )

    def test_pathologically_deep_model_nesting_is_rejected(self):
        with pytest.raises(ValueError, match="depth"):
            is_request_body_safe(
                request_body=_rounds_deep_api_base_payload(5000, "fallbacks"),
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )


class TestIsRequestBodySafeRejectsUrlValuedFallback:
    @pytest.mark.parametrize("fallback_field", ["fallbacks", "context_window_fallbacks", "content_policy_fallbacks"])
    def test_url_valued_string_fallback_is_rejected(self, fallback_field):
        with pytest.raises(ValueError, match="URL-valued fallback"):
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    fallback_field: [{"gpt-4": ["huggingface/http://attacker.example/path"]}],
                },
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )

    @pytest.mark.parametrize("fallback_field", ["fallbacks", "context_window_fallbacks", "content_policy_fallbacks"])
    def test_url_valued_dict_model_fallback_is_rejected(self, fallback_field):
        with pytest.raises(ValueError, match="URL-valued fallback"):
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    fallback_field: [{"gpt-4": [{"model": "huggingface/http://attacker.example/path"}]}],
                },
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )

    def test_ordinary_string_fallback_is_allowed(self):
        assert (
            is_request_body_safe(
                request_body={"model": "gpt-4", "fallbacks": [{"gpt-4": ["gpt-4-backup"]}]},
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )
            is True
        )

    def test_ordinary_dict_model_fallback_is_allowed(self):
        assert (
            is_request_body_safe(
                request_body={"model": "gpt-4", "fallbacks": [{"gpt-4": [{"model": "gpt-4-backup"}]}]},
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )
            is True
        )


class TestIsRequestBodySafeBlocksEndpointTargetingFields:
    """
    ``is_request_body_safe`` rejects request-body fields that retarget the
    outbound request to a caller-controlled host. Beyond the original
    ``api_base`` / ``base_url``, the same protection must apply to:

    * ``aws_bedrock_runtime_endpoint`` — Bedrock endpoint redirect; an
      attacker-controlled value coerces the proxy to authenticate against
      their host with the admin's AWS creds.
    * ``langsmith_base_url`` — Langsmith callback host; attacker-controlled
      values exfiltrate the entire request payload (incl. message content)
      via the observability hook.
    * ``langfuse_host`` — same exfil vector via the Langfuse hook.
    """

    @pytest.fixture(autouse=True)
    def _disable_url_validation(self, monkeypatch):
        # The new banned-params entries should be rejected even when
        # ``user_url_validation`` is off — the gate isn't the URL guard,
        # it's the banned-params list.
        import litellm

        monkeypatch.setattr(litellm, "user_url_validation", False, raising=False)

    @pytest.mark.parametrize(
        "field",
        [
            "aws_bedrock_runtime_endpoint",
            "langsmith_base_url",
            "langfuse_host",
            "posthog_host",
            "braintrust_host",
            "slack_webhook_url",
            "s3_endpoint_url",
            "sagemaker_base_url",
            "deployment_url",
        ],
    )
    def test_endpoint_targeting_field_in_request_body_is_rejected(self, field):
        with pytest.raises(ValueError, match='Rejected Request') as exc:
            is_request_body_safe(
                request_body={"model": "gpt-4", field: "https://attacker.example"},
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )
        # The function lists the offending param name in the error.
        assert field in str(exc.value)

    @pytest.mark.parametrize(
        "field",
        ["api_base", "base_url", "user_config", "langfuse_host", "slack_webhook_url"],
    )
    def test_api_key_does_not_bypass_blocklist(self, field):
        # Regression: the historical ``check_complete_credentials`` clause
        # made the entire blocklist a no-op for any caller that supplied
        # a non-empty ``api_key``. That bypass turned every missing entry
        # on the blocklist into an SSRF / credential-exfil hole. Verify
        # that supplying an api_key (alongside the banned param) does NOT
        # bypass the gate — it can only be opened by an admin opt-in.
        with pytest.raises(ValueError, match='Rejected Request') as exc:
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    "api_key": "sk-anything",
                    field: "https://attacker.example",
                },
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )
        assert field in str(exc.value)

    def test_admin_opt_in_proxy_wide_still_allows(self):
        # ``general_settings.allow_client_side_credentials = True`` remains
        # the documented proxy-wide BYOK opt-in.
        assert (
            is_request_body_safe(
                request_body={"model": "gpt-4", "api_base": "https://my-byok.example"},
                general_settings={"allow_client_side_credentials": True},
                llm_router=None,
                model="gpt-4",
            )
            is True
        )


class TestIsRequestBodySafeBlocksBedrockProjectOverride:
    """``aws_bedrock_project_id`` pins a deployment to a Bedrock project so
    that project's data-retention policy applies to its requests. A
    caller-supplied value would run the request under any project reachable
    with the deployment's shared AWS credentials, bypassing the configured
    retention/accounting association."""

    def test_project_id_in_request_body_is_rejected(self):
        with pytest.raises(ValueError, match="aws_bedrock_project_id"):
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    "aws_bedrock_project_id": "proj_attacker000000",
                },
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )

    def test_admin_opt_in_proxy_wide_allows_project_id(self):
        assert (
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    "aws_bedrock_project_id": "proj_byok000000",
                },
                general_settings={"allow_client_side_credentials": True},
                llm_router=None,
                model="gpt-4",
            )
            is True
        )


class TestIsRequestBodySafeBlocksRustOptIn:
    """``rust`` hands the whole call to the Rust core, which signs and sends
    with its own HTTP client rather than the one the deployment configured, and
    reports no ``post_call``. The proxy splats the request body straight into
    the router, and ``rust`` is a litellm param, so it lands in
    ``litellm_params`` and the gate honours it: without this entry any
    authenticated caller picks a transport and a callback surface the admin
    never chose. It stays a deployment decision, liftable only by the same
    admin opt-in as the rest of the list."""

    def test_rust_in_request_body_is_rejected(self):
        with pytest.raises(ValueError, match="rust"):
            is_request_body_safe(
                request_body={"model": "gpt-4", "rust": True},
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )

    def test_rust_under_extra_body_is_rejected(self):
        with pytest.raises(ValueError, match="not allowed in request body"):
            is_request_body_safe(
                request_body={"model": "gpt-4", "extra_body": {"rust": True}},
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )

    def test_api_key_does_not_bypass_the_rust_block(self):
        with pytest.raises(ValueError, match="rust"):
            is_request_body_safe(
                request_body={"model": "gpt-4", "api_key": "sk-anything", "rust": True},
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )

    def test_admin_opt_in_proxy_wide_allows_rust(self):
        assert (
            is_request_body_safe(
                request_body={"model": "gpt-4", "rust": True},
                general_settings={"allow_client_side_credentials": True},
                llm_router=None,
                model="gpt-4",
            )
            is True
        )

    def test_body_without_rust_is_still_allowed(self):
        assert (
            is_request_body_safe(
                request_body={"model": "gpt-4", "temperature": 0.7},
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )
            is True
        )


class TestIsRequestBodySafeBlocksVertexCredentialAlias:
    @pytest.mark.parametrize("field", ["vertex_ai_credentials"])
    def test_field_in_request_body_is_rejected(self, field):
        with pytest.raises(ValueError, match=field):
            is_request_body_safe(
                request_body={"model": "gpt-4", field: "attacker-supplied"},
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )

    @pytest.mark.parametrize("field", ["vertex_ai_credentials"])
    def test_admin_opt_in_proxy_wide_allows(self, field):
        assert (
            is_request_body_safe(
                request_body={"model": "gpt-4", field: "byok-supplied"},
                general_settings={"allow_client_side_credentials": True},
                llm_router=None,
                model="gpt-4",
            )
            is True
        )

    def test_legitimate_request_body_param_still_allowed(self):
        assert (
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    "temperature": 0.7,
                    "max_tokens": 128,
                    "user": "end-user-123",
                },
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )
            is True
        )


class TestIsRequestBodySafeBlocksNVCFFunctionOverride:
    """``nvcf_function_id`` is rejected as a request-body param unless the
    admin opted in proxy-wide or per-deployment."""

    def test_nvcf_function_id_in_request_body_is_rejected(self):
        with pytest.raises(ValueError, match="nvcf_function_id"):
            is_request_body_safe(
                request_body={
                    "model": "nvidia_riva/parakeet",
                    "nvcf_function_id": "caller-supplied",
                },
                general_settings={},
                llm_router=None,
                model="nvidia_riva/parakeet",
            )

    def test_nvcf_function_id_with_api_key_still_rejected(self):
        with pytest.raises(ValueError, match="nvcf_function_id"):
            is_request_body_safe(
                request_body={
                    "model": "nvidia_riva/parakeet",
                    "api_key": "sk-anything",
                    "nvcf_function_id": "caller-supplied",
                },
                general_settings={},
                llm_router=None,
                model="nvidia_riva/parakeet",
            )

    def test_admin_opt_in_proxy_wide_allows_nvcf_function_id(self):
        assert (
            is_request_body_safe(
                request_body={
                    "model": "nvidia_riva/parakeet",
                    "nvcf_function_id": "byok-function-id",
                },
                general_settings={"allow_client_side_credentials": True},
                llm_router=None,
                model="nvidia_riva/parakeet",
            )
            is True
        )

    def test_admin_opt_in_per_deployment_allows_nvcf_function_id(self, monkeypatch):
        """The error message lists per-deployment ``configurable_clientside_auth_params``
        as a second opt-in. Cover that path too so it can't silently regress."""
        from litellm.proxy.auth import auth_utils

        monkeypatch.setattr(
            auth_utils,
            "_allow_model_level_clientside_configurable_parameters",
            lambda model, param, request_body_value, llm_router: param == "nvcf_function_id",
        )

        assert (
            is_request_body_safe(
                request_body={
                    "model": "nvidia_riva/parakeet",
                    "nvcf_function_id": "byok-function-id",
                },
                general_settings={},
                llm_router=None,
                model="nvidia_riva/parakeet",
            )
            is True
        )


class TestIsRequestBodySafeBlocksRivaUseSsl:
    """``use_ssl`` is rejected as a request-body param unless the admin
    opted in proxy-wide or per-deployment."""

    def test_use_ssl_in_request_body_is_rejected(self):
        with pytest.raises(ValueError, match="use_ssl"):
            is_request_body_safe(
                request_body={
                    "model": "nvidia_riva/parakeet",
                    "use_ssl": False,
                },
                general_settings={},
                llm_router=None,
                model="nvidia_riva/parakeet",
            )

    def test_admin_opt_in_proxy_wide_allows_use_ssl(self):
        assert (
            is_request_body_safe(
                request_body={
                    "model": "nvidia_riva/parakeet",
                    "use_ssl": True,
                },
                general_settings={"allow_client_side_credentials": True},
                llm_router=None,
                model="nvidia_riva/parakeet",
            )
            is True
        )

    def test_admin_opt_in_per_deployment_allows_use_ssl(self, monkeypatch):
        from litellm.proxy.auth import auth_utils

        monkeypatch.setattr(
            auth_utils,
            "_allow_model_level_clientside_configurable_parameters",
            lambda model, param, request_body_value, llm_router: param == "use_ssl",
        )

        assert (
            is_request_body_safe(
                request_body={
                    "model": "nvidia_riva/parakeet",
                    "use_ssl": True,
                },
                general_settings={},
                llm_router=None,
                model="nvidia_riva/parakeet",
            )
            is True
        )


class TestIsRequestBodySafeBlocksBedrockTags:
    """``bedrock_tags`` lands as AWS resource tags on Bedrock batch jobs
    created with the proxy's AWS identity, so a caller-supplied value can
    forge ownership or cost-allocation labels; like
    ``aws_bedrock_project_id`` it is blocked without an admin opt-in."""

    def test_bedrock_tags_in_request_body_is_rejected(self):
        with pytest.raises(ValueError, match="bedrock_tags"):
            is_request_body_safe(
                request_body={
                    "model": "bedrock-batch-opus",
                    "bedrock_tags": [{"key": "application", "value": "genai-proxy"}],
                },
                general_settings={},
                llm_router=None,
                model="bedrock-batch-opus",
            )

    def test_admin_opt_in_proxy_wide_allows_bedrock_tags(self):
        assert (
            is_request_body_safe(
                request_body={
                    "model": "bedrock-batch-opus",
                    "bedrock_tags": [{"key": "application", "value": "genai-proxy"}],
                },
                general_settings={"allow_client_side_credentials": True},
                llm_router=None,
                model="bedrock-batch-opus",
            )
            is True
        )

    def test_admin_opt_in_per_deployment_allows_bedrock_tags(self):
        from litellm import Router

        router = Router(
            model_list=[
                {
                    "model_name": "bedrock-batch-opus",
                    "litellm_params": {
                        "model": "bedrock/us.anthropic.claude-opus-4-7",
                        "configurable_clientside_auth_params": ["bedrock_tags"],
                    },
                }
            ]
        )
        assert (
            is_request_body_safe(
                request_body={
                    "model": "bedrock-batch-opus",
                    "bedrock_tags": [{"key": "application", "value": "genai-proxy"}],
                },
                general_settings={},
                llm_router=router,
                model="bedrock-batch-opus",
            )
            is True
        )

    def test_per_deployment_opt_in_for_other_param_still_rejects_bedrock_tags(self):
        from litellm import Router

        router = Router(
            model_list=[
                {
                    "model_name": "bedrock-batch-opus",
                    "litellm_params": {
                        "model": "bedrock/us.anthropic.claude-opus-4-7",
                        "configurable_clientside_auth_params": ["api_base"],
                    },
                }
            ]
        )
        with pytest.raises(ValueError, match="bedrock_tags"):
            is_request_body_safe(
                request_body={
                    "model": "bedrock-batch-opus",
                    "bedrock_tags": [{"key": "application", "value": "genai-proxy"}],
                },
                general_settings={},
                llm_router=router,
                model="bedrock-batch-opus",
            )


# ── is_request_body_safe nested-config recursion (VERIA-6) ────────────────────


class TestIsRequestBodySafeNestedConfig:
    """The Milvus vector store transformer unpacks
    ``litellm_embedding_config`` as ``**kwargs`` into ``litellm.embedding(...)``
    — same SSRF / credential-exfil surface as a top-level ``api_base`` in
    the request body. ``is_request_body_safe`` must recurse into this
    nested dict so a banned param can't be smuggled in via nesting."""

    def test_root_level_api_base_blocked_when_no_opt_in(self):
        """Sanity check: pre-existing root-level enforcement still works."""
        with pytest.raises(ValueError, match="api_base"):
            is_request_body_safe(
                request_body={"api_base": "https://attacker.example.com"},
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )

    def test_nested_api_base_in_embedding_config_blocked(self):
        """Smuggling ``api_base`` inside ``litellm_embedding_config`` is
        the VERIA-6 bypass — must be blocked by the recursive check."""
        with pytest.raises(ValueError, match="api_base"):
            is_request_body_safe(
                request_body={
                    "litellm_embedding_config": {
                        "api_base": "https://attacker.example.com",
                        "api_key": "leaked-key",
                    }
                },
                general_settings={},
                llm_router=None,
                model="milvus-store",
            )

    def test_nested_nvcf_function_id_in_metadata_blocked(self):
        """Smuggling ``nvcf_function_id`` via ``metadata`` / ``extra_body``
        is the same shape as the VERIA-6 ``api_base`` bypass — must be
        rejected by the recursive walk so the NVCF override gate cannot
        be sidestepped with nesting."""
        with pytest.raises(ValueError, match="nvcf_function_id"):
            is_request_body_safe(
                request_body={
                    "model": "nvidia_riva/parakeet",
                    "litellm_metadata": {"nvcf_function_id": "attacker-via-metadata"},
                },
                general_settings={},
                llm_router=None,
                model="nvidia_riva/parakeet",
            )

    def test_nested_langfuse_host_in_embedding_config_blocked(self):
        """The recursion uses the *full* banned-param list, not a special
        subset — so any flag that's banned at the root is also banned
        when nested."""
        with pytest.raises(ValueError, match="langfuse_host"):
            is_request_body_safe(
                request_body={
                    "litellm_embedding_config": {
                        "langfuse_host": "https://attacker.example.com"
                    }
                },
                general_settings={},
                llm_router=None,
                model="milvus-store",
            )

    def test_nested_api_base_allowed_when_admin_opts_in(self):
        """Admins who explicitly enable client-side credential passthrough
        keep the existing escape hatch — same UX as for root-level."""
        assert (
            is_request_body_safe(
                request_body={
                    "litellm_embedding_config": {
                        "api_base": "https://my-azure.example.com"
                    }
                },
                general_settings={"allow_client_side_credentials": True},
                llm_router=None,
                model="milvus-store",
            )
            is True
        )

    def test_safe_nested_config_accepted(self):
        """A nested config without any banned params passes — there's no
        false-positive on legitimate ``api_version`` / model params."""
        assert (
            is_request_body_safe(
                request_body={
                    "litellm_embedding_config": {
                        "api_version": "2024-02-15-preview",
                    }
                },
                general_settings={},
                llm_router=None,
                model="milvus-store",
            )
            is True
        )

    def test_non_dict_nested_config_does_not_break_check(self):
        """A bogus type for ``litellm_embedding_config`` (string, list,
        None) must not crash the validator — it should just fall through."""
        assert (
            is_request_body_safe(
                request_body={"litellm_embedding_config": "not-a-dict"},
                general_settings={},
                llm_router=None,
                model="x",
            )
            is True
        )

    def test_deeply_nested_config_does_not_recurse(self):
        """Greptile P1: ``is_request_body_safe`` is iterative single-level —
        a deeply-nested ``litellm_embedding_config`` cannot exhaust the
        Python call stack to trigger a 500 ``RecursionError``. Build a
        body 1000 levels deep; the validator must complete in O(1)
        descent."""
        body = {"litellm_embedding_config": {}}
        cur = body["litellm_embedding_config"]
        for _ in range(1000):
            cur["litellm_embedding_config"] = {}
            cur = cur["litellm_embedding_config"]
        # Banned param at the deepest level shouldn't be reached — single
        # level only.
        cur["api_base"] = "https://attacker.example.com"

        # No exception raised: deeper levels aren't checked.
        assert (
            is_request_body_safe(
                request_body=body,
                general_settings={},
                llm_router=None,
                model="x",
            )
            is True
        )


# ── observability-callback ban (root + metadata) ───────────────────────────


class TestObservabilityCallbackBans:
    """The proxy must reject observability credentials, hosts, and project
    identifiers regardless of whether they arrive at the request body root,
    in ``metadata`` / ``litellm_metadata``, or in a JSON-string-encoded
    metadata blob (multipart/``extra_body`` path).

    The ban list is derived from
    ``litellm.litellm_core_utils.initialize_dynamic_callback_params._supported_callback_params``
    minus a small ``_SAFE_CLIENT_CALLBACK_PARAMS`` allow-list, plus
    ``_EXTRA_BANNED_OBSERVABILITY_PARAMS`` for fields integrations read but
    that are not yet in the canonical allow-list. The derivation keeps the
    proxy in sync as new integrations are added.
    """

    @pytest.fixture(autouse=True)
    def _disable_url_validation(self, monkeypatch):
        import litellm

        monkeypatch.setattr(litellm, "user_url_validation", False, raising=False)

    @pytest.mark.parametrize(
        "field",
        [
            "langfuse_public_key",
            "langfuse_secret",
            "langfuse_secret_key",
            "langsmith_api_key",
            "langsmith_project",
            "langsmith_tenant_id",
            "arize_api_key",
            "arize_space_key",
            "arize_space_id",
            "posthog_api_key",
            "posthog_api_url",
            "braintrust_api_key",
            "braintrust_project",
            "wandb_api_key",
            "weave_project_id",
            "gcs_bucket_name",
            "gcs_path_service_account",
            "humanloop_api_key",
            "lunary_public_key",
        ],
    )
    def test_observability_field_in_request_body_root_is_rejected(self, field):
        with pytest.raises(ValueError, match='Rejected Request') as exc:
            is_request_body_safe(
                request_body={"model": "gpt-4", field: "attacker-value"},
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )
        assert field in str(exc.value)

    @pytest.mark.parametrize(
        "metadata_key",
        ["metadata", "litellm_metadata"],
    )
    @pytest.mark.parametrize(
        "field",
        [
            "langfuse_host",
            "langfuse_secret_key",
            "langsmith_api_key",
            "posthog_api_url",
            "braintrust_project",
            "user_api_key_auth_metadata",
        ],
    )
    def test_observability_field_in_metadata_dict_is_rejected(
        self, metadata_key, field
    ):
        # Verifies the metadata walk: a value smuggled inside ``metadata``
        # or ``litellm_metadata`` is just as dangerous as the same field
        # at the body root, and must hit the same gate.
        with pytest.raises(ValueError, match='Rejected Request') as exc:
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    metadata_key: {field: "attacker-value"},
                },
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )
        assert field in str(exc.value)

    @pytest.mark.parametrize("metadata_key", ["metadata", "litellm_metadata"])
    @pytest.mark.parametrize(
        "field",
        ["phoenix_project_name", "phoenix_project_name_override"],
    )
    def test_phoenix_project_fields_in_metadata_are_accepted(self, metadata_key, field):
        # The Phoenix integrations only honor the project from
        # ``user_api_key_auth_metadata`` on the proxy, so the bare metadata
        # fields are inert and must not 400 SDK-style callers that send them.
        assert (
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    metadata_key: {field: "client-project"},
                },
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )
            is True
        )

    def test_observability_field_in_litellm_params_metadata_is_rejected(self):
        with pytest.raises(ValueError, match='Rejected Request: turn_off_message_logging is not allowed') as exc:
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    "litellm_params": {
                        "metadata": {"turn_off_message_logging": False}
                    },
                },
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )
        assert "turn_off_message_logging" in str(exc.value)

    @pytest.mark.parametrize(
        "metadata_key",
        ["metadata", "litellm_metadata"],
    )
    def test_observability_field_in_json_string_metadata_is_rejected(
        self, metadata_key
    ):
        # Multipart/form-data and ``extra_body`` callers send metadata as a
        # JSON-encoded string. The bouncer parses it before applying the
        # banned-params check so the JSON-string path can't smuggle past
        # the ``isinstance(dict)`` guard.
        import json

        with pytest.raises(ValueError, match='Rejected Request: langfuse_host is not allowed in request') as exc:
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    metadata_key: json.dumps(
                        {"langfuse_host": "https://attacker.example"}
                    ),
                },
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )
        assert "langfuse_host" in str(exc.value)

    def test_admin_opt_in_allows_metadata_credential_passthrough(self):
        # The opt-in gate covers the metadata path the same way it covers
        # the root path — operators running BYO observability with
        # clientside creds flip a single flag and both paths work.
        assert (
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    "metadata": {
                        "langfuse_host": "https://my-langfuse.example",
                        "langfuse_public_key": "pk-mine",
                        "langfuse_secret_key": "sk-mine",
                    },
                },
                general_settings={"allow_client_side_credentials": True},
                llm_router=None,
                model="gpt-4",
            )
            is True
        )

    def test_safe_per_request_observability_metadata_is_allowed(self):
        # Informational fields (sampling rate, prompt version) describe
        # the request being logged — they don't choose the destination or
        # credentials, so they must remain accepted from clients without
        # the opt-in flag.
        assert (
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    "metadata": {
                        "langfuse_prompt_version": "v2",
                        "langsmith_sampling_rate": 0.1,
                    },
                },
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )
            is True
        )


def test_model_level_allow_does_not_skip_subsequent_banned_params(monkeypatch):
    """Greptile P1: ``_check_banned_params`` previously ``return``-ed when a
    deployment's ``configurable_clientside_auth_params`` permitted one
    banned field, exiting before any later banned field in the same body
    was checked. The metadata walk this PR adds multiplies the surface
    where that bypass matters: a body pairing a model-level-allowed
    ``api_base`` with an observability credential like ``langfuse_host``
    must still reject on the second field, not silently pass."""
    from litellm.proxy.auth import auth_utils

    monkeypatch.setattr(
        auth_utils,
        "_allow_model_level_clientside_configurable_parameters",
        lambda model, param, request_body_value, llm_router: param == "api_base",
    )

    with pytest.raises(ValueError, match='Rejected Request: langfuse_host is not allowed in request') as exc:
        is_request_body_safe(
            request_body={
                "model": "gpt-4",
                "api_base": "https://allowed-by-deployment.example",
                "langfuse_host": "https://attacker.example",
            },
            general_settings={},
            llm_router=None,
            model="gpt-4",
        )
    assert "langfuse_host" in str(exc.value)


def test_observability_ban_covers_canonical_supported_callback_params():
    """Guard test: every entry in the canonical
    ``_supported_callback_params`` allow-list must end up either banned by
    the proxy or explicitly safe-listed. New integrations added to that
    list are banned by default (the safe failure mode); flagging them as
    safe is an explicit decision recorded in
    ``_SAFE_CLIENT_CALLBACK_PARAMS``."""
    from litellm.litellm_core_utils.initialize_dynamic_callback_params import (
        _request_blocked_callback_params,
        _supported_callback_params,
    )
    from litellm.proxy.auth.auth_utils import (
        _BANNED_REQUEST_BODY_PARAMS,
        _SAFE_CLIENT_CALLBACK_PARAMS,
    )

    banned = set(_BANNED_REQUEST_BODY_PARAMS)
    for param in _supported_callback_params:
        assert param in banned or param in _SAFE_CLIENT_CALLBACK_PARAMS, (
            f"{param} is in _supported_callback_params but neither banned nor "
            f"safe-listed. Add it to _SAFE_CLIENT_CALLBACK_PARAMS if it is an "
            f"informational per-request field; otherwise the derivation will "
            f"ban it automatically."
        )
    for param in _request_blocked_callback_params:
        assert param in banned, (
            f"{param} is in _request_blocked_callback_params but is not banned "
            "at the proxy request-body boundary."
        )


# ── pricing injection (global model cost registry poisoning) ──────────────────


class TestPricingInjectionBlocked:
    """Authenticated clients must not be able to mutate the global
    litellm.model_cost registry by supplying pricing fields in the request
    body.  Any CustomPricingLiteLLMParams field (input_cost_per_token etc.)
    passed to completion() is forwarded to register_model(), which overwrites
    the shared global dict for ALL users on the instance.

    Fix: all CustomPricingLiteLLMParams fields are in _BANNED_REQUEST_BODY_PARAMS,
    so is_request_body_safe() rejects them before they reach completion().
    """

    @pytest.mark.parametrize(
        "field,value",
        [
            ("input_cost_per_token", -0.01),
            ("output_cost_per_token", 0.0),
            ("input_cost_per_second", 999.0),
            ("output_cost_per_second", -1.0),
            ("cache_read_input_token_cost", 0.0),
            ("cache_creation_input_token_cost", -0.05),
        ],
    )
    def test_pricing_field_rejected_by_default(self, field, value):
        with pytest.raises(ValueError, match='Rejected Request') as exc:
            is_request_body_safe(
                request_body={"model": "gpt-4", field: value},
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )
        assert field in str(exc.value)

    def test_all_custom_pricing_fields_are_banned(self):
        from litellm.proxy.auth.auth_utils import _BANNED_REQUEST_BODY_PARAMS
        from litellm.types.utils import CustomPricingLiteLLMParams

        banned = set(_BANNED_REQUEST_BODY_PARAMS)
        for field in CustomPricingLiteLLMParams.model_fields:
            assert field in banned, (
                f"CustomPricingLiteLLMParams.{field} is not in "
                "_BANNED_REQUEST_BODY_PARAMS — clients can poison the global "
                "model cost registry by supplying it in the request body."
            )

    def test_pricing_field_allowed_with_admin_opt_in(self):
        assert (
            is_request_body_safe(
                request_body={"model": "gpt-4", "input_cost_per_token": 0.00001},
                general_settings={"allow_client_side_credentials": True},
                llm_router=None,
                model="gpt-4",
            )
            is True
        )


class TestGetRequestRouteTemplate:
    """get_request_route_template returns the low-cardinality FastAPI route
    template (e.g. /v1/threads/{thread_id}/runs) for http.route, distinct
    from the literal url.path. None when unavailable."""

    def _request(self, scope):
        req = MagicMock()
        req.scope = scope
        return req

    def test_returns_route_template(self):
        route = MagicMock()
        route.path = "/v1/threads/{thread_id}/runs"
        req = self._request({"route": route, "path": "/v1/threads/abc123/runs"})
        # template, not the literal path — two thread IDs share this value
        assert get_request_route_template(req) == "/v1/threads/{thread_id}/runs"

    def test_scope_not_dict_returns_none(self):
        assert get_request_route_template(self._request("not-a-dict")) is None

    def test_no_route_in_scope_returns_none(self):
        assert get_request_route_template(self._request({"path": "/x"})) is None

    def test_route_without_str_path_returns_none(self):
        route = MagicMock()
        route.path = 12345  # not a str
        assert get_request_route_template(self._request({"route": route})) is None

    def test_route_with_empty_path_returns_none(self):
        route = MagicMock()
        route.path = ""
        assert get_request_route_template(self._request({"route": route})) is None

    def test_exception_returns_none(self):
        req = MagicMock()
        type(req).scope = property(
            lambda self: (_ for _ in ()).throw(RuntimeError("boom"))
        )
        assert get_request_route_template(req) is None


class TestIsRequestBodySafeBlocksModelList:
    """model_list is an SDK-only field with no proxy API meaning; it must
    be rejected from the request body regardless of any opt-in."""

    def test_model_list_rejected_with_no_opt_in(self):
        with pytest.raises(ValueError, match="model_list is not allowed"):
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    "messages": [{"role": "user", "content": "hi"}],
                    "model_list": [{"model_name": "x", "litellm_params": {}}],
                },
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )

    def test_model_list_rejected_even_with_proxy_wide_opt_in(self):
        with pytest.raises(ValueError, match="model_list is not allowed"):
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    "messages": [{"role": "user", "content": "hi"}],
                    "model_list": [],
                },
                general_settings={"allow_client_side_credentials": True},
                llm_router=None,
                model="gpt-4",
            )

    def test_normal_body_still_passes(self):
        assert (
            is_request_body_safe(
                request_body={
                    "model": "gpt-4",
                    "messages": [{"role": "user", "content": "hi"}],
                },
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )
            is True
        )


class TestGetKeyTagRateLimits:
    """Tests for get_key_tag_rpm_limit."""

    def test_reads_tag_rpm_limit_from_metadata(self):
        key = UserAPIKeyAuth(
            api_key="sk-123", metadata={"tag_rpm_limit": {"cell-1": 5}}
        )
        assert get_key_tag_rpm_limit(key) == {"cell-1": 5}

    def test_returns_none_when_unset(self):
        key = UserAPIKeyAuth(api_key="sk-123")
        assert get_key_tag_rpm_limit(key) is None


class TestIsRequestBodySafeChecksBracketNotationMetadata:
    """Bracket notation is how multipart callers express nested metadata; it is
    validated the same way the dict form is."""

    @pytest.mark.parametrize("metadata_key", ["metadata", "litellm_metadata"])
    def test_bracket_notation_banned_param_is_rejected(self, metadata_key):
        with pytest.raises(ValueError, match="langfuse_host"):
            is_request_body_safe(
                request_body={
                    "purpose": "assistants",
                    f"{metadata_key}[langfuse_host]": "https://example.invalid",
                },
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )

    def test_bracket_notation_api_base_is_rejected(self):
        with pytest.raises(ValueError, match="api_base"):
            is_request_body_safe(
                request_body={"litellm_metadata[api_base]": "https://example.invalid"},
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )

    def test_bracket_notation_allowed_under_proxy_wide_opt_in(self):
        assert (
            is_request_body_safe(
                request_body={"litellm_metadata[langfuse_host]": "https://byok.example"},
                general_settings={"allow_client_side_credentials": True},
                llm_router=None,
                model="gpt-4",
            )
            is True
        )

    def test_benign_bracket_notation_metadata_is_allowed(self):
        assert (
            is_request_body_safe(
                request_body={
                    "purpose": "assistants",
                    "litellm_metadata[spend_logs_metadata][owner]": "john",
                    "litellm_metadata[tags]": "production",
                },
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )
            is True
        )

    def test_bracket_notation_matches_json_encoding_for_deeper_nesting(self):
        """A value nested below the first level is treated the same either way:
        the check descends one level into metadata, for both encodings."""
        deep_bracket = {
            "litellm_metadata[spend_logs_metadata][langfuse_host]": "https://example.invalid"
        }
        deep_json = {
            "litellm_metadata": {"spend_logs_metadata": {"langfuse_host": "https://example.invalid"}}
        }
        kwargs = dict(general_settings={}, llm_router=None, model="gpt-4")
        assert is_request_body_safe(request_body=deep_bracket, **kwargs) is True
        assert is_request_body_safe(request_body=deep_json, **kwargs) is True

    def test_body_without_bracket_keys_is_unaffected(self):
        assert (
            is_request_body_safe(
                request_body={"model": "gpt-4", "messages": [{"role": "user", "content": "hi"}]},
                general_settings={},
                llm_router=None,
                model="gpt-4",
            )
            is True
        )


class TestHasUserSetupSso:
    """has_user_setup_sso must treat SAML IdP metadata as SSO configured.

    Regression: UI discovery used this helper for sso_configured, but it only
    checked OAuth client IDs, so SAML-only setups left the login button gray.
    """

    @pytest.fixture(autouse=True)
    def _clear_sso_env(self, monkeypatch):
        for key in (
            "MICROSOFT_CLIENT_ID",
            "GOOGLE_CLIENT_ID",
            "GENERIC_CLIENT_ID",
            "SAML_IDP_METADATA_URL",
            "SAML_IDP_METADATA_XML",
        ):
            monkeypatch.delenv(key, raising=False)

    def test_false_when_no_sso_env(self):
        from litellm.proxy.auth.auth_utils import has_user_setup_sso

        assert has_user_setup_sso() is False

    def test_true_for_oauth_client_ids(self, monkeypatch):
        from litellm.proxy.auth.auth_utils import has_user_setup_sso

        monkeypatch.setenv("GOOGLE_CLIENT_ID", "google-client")
        assert has_user_setup_sso() is True

    def test_true_for_saml_metadata_url(self, monkeypatch):
        from litellm.proxy.auth.auth_utils import has_user_setup_sso

        monkeypatch.setenv(
            "SAML_IDP_METADATA_URL", "https://idp.example.com/metadata.xml"
        )
        assert has_user_setup_sso() is True

    def test_true_for_saml_metadata_xml(self, monkeypatch):
        from litellm.proxy.auth.auth_utils import has_user_setup_sso

        monkeypatch.setenv("SAML_IDP_METADATA_XML", "<EntityDescriptor/>")
        assert has_user_setup_sso() is True


class TestIsSsoProviderFullyConfigured:
    """A lone client id must not read as ready: `has_user_setup_sso()` only
    checks the client id (correct for a UI-discovery "show the login button"
    decision), but a gate that BLOCKS the password fallback needs every
    companion setting the provider requires, or an incomplete setup locks
    every admin out with no working login path at all."""

    @pytest.fixture(autouse=True)
    def _clear_sso_env(self, monkeypatch):
        for key in (
            "GOOGLE_CLIENT_ID",
            "GOOGLE_CLIENT_SECRET",
            "MICROSOFT_CLIENT_ID",
            "MICROSOFT_CLIENT_SECRET",
            "MICROSOFT_TENANT",
            "GENERIC_CLIENT_ID",
            "GENERIC_CLIENT_SECRET",
            "GENERIC_AUTHORIZATION_ENDPOINT",
            "GENERIC_TOKEN_ENDPOINT",
            "GENERIC_USERINFO_ENDPOINT",
            "SAML_IDP_METADATA_URL",
            "SAML_IDP_METADATA_XML",
        ):
            monkeypatch.delenv(key, raising=False)

    def test_false_when_nothing_configured(self):
        from litellm.proxy.auth.auth_utils import is_sso_provider_fully_configured

        assert is_sso_provider_fully_configured() is False

    def test_google_client_id_alone_is_not_ready(self, monkeypatch):
        from litellm.proxy.auth.auth_utils import is_sso_provider_fully_configured

        monkeypatch.setenv("GOOGLE_CLIENT_ID", "google-client")
        assert is_sso_provider_fully_configured() is False

    def test_google_with_secret_is_ready(self, monkeypatch):
        from litellm.proxy.auth.auth_utils import is_sso_provider_fully_configured

        monkeypatch.setenv("GOOGLE_CLIENT_ID", "google-client")
        monkeypatch.setenv("GOOGLE_CLIENT_SECRET", "google-secret")
        assert is_sso_provider_fully_configured() is True

    def test_microsoft_client_id_alone_is_not_ready(self, monkeypatch):
        from litellm.proxy.auth.auth_utils import is_sso_provider_fully_configured

        monkeypatch.setenv("MICROSOFT_CLIENT_ID", "ms-client")
        assert is_sso_provider_fully_configured() is False

    def test_microsoft_missing_tenant_is_not_ready(self, monkeypatch):
        from litellm.proxy.auth.auth_utils import is_sso_provider_fully_configured

        monkeypatch.setenv("MICROSOFT_CLIENT_ID", "ms-client")
        monkeypatch.setenv("MICROSOFT_CLIENT_SECRET", "ms-secret")
        assert is_sso_provider_fully_configured() is False

    def test_microsoft_with_secret_and_tenant_is_ready(self, monkeypatch):
        from litellm.proxy.auth.auth_utils import is_sso_provider_fully_configured

        monkeypatch.setenv("MICROSOFT_CLIENT_ID", "ms-client")
        monkeypatch.setenv("MICROSOFT_CLIENT_SECRET", "ms-secret")
        monkeypatch.setenv("MICROSOFT_TENANT", "ms-tenant")
        assert is_sso_provider_fully_configured() is True

    def test_generic_client_id_alone_is_not_ready(self, monkeypatch):
        from litellm.proxy.auth.auth_utils import is_sso_provider_fully_configured

        monkeypatch.setenv("GENERIC_CLIENT_ID", "generic-client")
        assert is_sso_provider_fully_configured() is False

    def test_generic_missing_one_endpoint_is_not_ready(self, monkeypatch):
        from litellm.proxy.auth.auth_utils import is_sso_provider_fully_configured

        monkeypatch.setenv("GENERIC_CLIENT_ID", "generic-client")
        monkeypatch.setenv("GENERIC_CLIENT_SECRET", "generic-secret")
        monkeypatch.setenv("GENERIC_AUTHORIZATION_ENDPOINT", "https://idp.example.com/authorize")
        monkeypatch.setenv("GENERIC_TOKEN_ENDPOINT", "https://idp.example.com/token")
        # GENERIC_USERINFO_ENDPOINT deliberately left unset.
        assert is_sso_provider_fully_configured() is False

    def test_generic_with_every_endpoint_is_ready(self, monkeypatch):
        from litellm.proxy.auth.auth_utils import is_sso_provider_fully_configured

        monkeypatch.setenv("GENERIC_CLIENT_ID", "generic-client")
        monkeypatch.setenv("GENERIC_CLIENT_SECRET", "generic-secret")
        monkeypatch.setenv("GENERIC_AUTHORIZATION_ENDPOINT", "https://idp.example.com/authorize")
        monkeypatch.setenv("GENERIC_TOKEN_ENDPOINT", "https://idp.example.com/token")
        monkeypatch.setenv("GENERIC_USERINFO_ENDPOINT", "https://idp.example.com/userinfo")
        assert is_sso_provider_fully_configured() is True

    def test_saml_metadata_url_is_ready_when_runtime_installed(self, monkeypatch):
        from litellm.proxy.auth import auth_utils

        monkeypatch.setenv("SAML_IDP_METADATA_URL", "https://idp.example.com/metadata.xml")
        monkeypatch.setattr(auth_utils.importlib.util, "find_spec", lambda name: object())
        assert auth_utils.is_sso_provider_fully_configured() is True

    def test_saml_metadata_url_is_not_ready_without_runtime(self, monkeypatch):
        """Regression: python3-saml (``onelogin.saml2``) is an optional
        dependency; SAMLAuthHandler fails closed on every request when it is
        not installed, so IdP metadata alone must not read as ready."""
        from litellm.proxy.auth import auth_utils

        monkeypatch.setenv("SAML_IDP_METADATA_URL", "https://idp.example.com/metadata.xml")
        monkeypatch.setattr(auth_utils.importlib.util, "find_spec", lambda name: None)
        assert auth_utils.is_sso_provider_fully_configured() is False

    def test_saml_check_does_not_raise_when_package_entirely_absent(self, monkeypatch):
        """Regression: `importlib.util.find_spec("onelogin.saml2.auth")`
        raises ModuleNotFoundError (not merely returns None) when the
        TOP-LEVEL `onelogin` package is not installed at all, which is
        exactly the real-world "optional extra not installed" case. If the
        gate does not catch this, every password login 500s instead of
        falling back, on a deployment that configured SAML metadata but
        skipped the extra."""
        from litellm.proxy.auth import auth_utils

        def _raise(name: str):
            raise ModuleNotFoundError("No module named 'onelogin'")

        monkeypatch.setenv("SAML_IDP_METADATA_URL", "https://idp.example.com/metadata.xml")
        monkeypatch.setattr(auth_utils.importlib.util, "find_spec", _raise)
        assert auth_utils.is_sso_provider_fully_configured() is False

    def test_incomplete_earlier_provider_does_not_mask_a_ready_later_one(self, monkeypatch):
        """Regression: a stray GOOGLE_CLIENT_ID with no secret (e.g. a
        leftover from a migration) must not stop the check from reaching a
        fully configured Microsoft provider set alongside it — every
        provider is evaluated independently, not in a first-match order."""
        from litellm.proxy.auth.auth_utils import is_sso_provider_fully_configured

        monkeypatch.setenv("GOOGLE_CLIENT_ID", "google-client")
        monkeypatch.setenv("MICROSOFT_CLIENT_ID", "ms-client")
        monkeypatch.setenv("MICROSOFT_CLIENT_SECRET", "ms-secret")
        monkeypatch.setenv("MICROSOFT_TENANT", "ms-tenant")
        assert is_sso_provider_fully_configured() is True


class TestIsRequestBodySafeBlocksAwsIdentitySelectors:
    """A caller must not be able to redirect Bedrock signing to another identity
    reachable from the proxy host. ``get_credentials`` prefers a named profile
    and the AssumeRole knobs over the deployment's static keys, and the file /
    batch endpoints fold the request body and the deployment credentials into a
    single params dict, so these have to be rejected at the boundary (#36155).
    """

    @pytest.mark.parametrize(
        "selector",
        ["aws_profile_name", "aws_session_name", "aws_external_id"],
    )
    def test_aws_identity_selector_in_batch_body_is_rejected(self, selector):
        with pytest.raises(ValueError, match=selector):
            is_request_body_safe(
                request_body={
                    "input_file_id": "file-abc123",
                    "endpoint": "/v1/chat/completions",
                    "completion_window": "24h",
                    "model": "bedrock-batch-model",
                    selector: "attacker-chosen",
                },
                general_settings={},
                llm_router=None,
                model="bedrock-batch-model",
            )

    @pytest.mark.parametrize(
        "selector",
        ["aws_profile_name", "aws_session_name", "aws_external_id"],
    )
    def test_aws_identity_selector_under_extra_body_is_rejected(self, selector):
        with pytest.raises(ValueError, match=selector):
            is_request_body_safe(
                request_body={
                    "model": "bedrock-batch-model",
                    "extra_body": {selector: "attacker-chosen"},
                },
                general_settings={},
                llm_router=None,
                model="bedrock-batch-model",
            )

    def test_aws_identity_selector_allowed_under_proxy_wide_opt_in(self):
        assert (
            is_request_body_safe(
                request_body={
                    "model": "bedrock-batch-model",
                    "aws_profile_name": "admin-approved-profile",
                },
                general_settings={"allow_client_side_credentials": True},
                llm_router=None,
                model="bedrock-batch-model",
            )
            is True
        )

    def test_upload_body_without_identity_selectors_is_accepted(self):
        assert (
            is_request_body_safe(
                request_body={"purpose": "batch", "model": "bedrock-batch-model"},
                general_settings={},
                llm_router=None,
                model="bedrock-batch-model",
            )
            is True
        )
