diff --git a/litellm/__init__.py b/litellm/__init__.py index 4fc71e12700..bc52683452a 100644 --- a/litellm/__init__.py +++ b/litellm/__init__.py @@ -145,6 +145,7 @@ _custom_logger_compatible_callbacks_literal = Literal[ "focus", "posthog", "levo", + "scopeblind", ] cold_storage_custom_logger: Optional[_custom_logger_compatible_callbacks_literal] = None logged_real_time_event_types: Optional[Union[List[str], Literal["*"]]] = None diff --git a/litellm/integrations/scopeblind.py b/litellm/integrations/scopeblind.py new file mode 100644 index 00000000000..c97bbe3e8d9 --- /dev/null +++ b/litellm/integrations/scopeblind.py @@ -0,0 +1,238 @@ +""" +ScopeBlind Integration for LiteLLM + +Device-level rate limiting for LLM proxy operators. +Verifies DPoP proofs (RFC 9449) to bind requests to unique devices +instead of IP addresses — survives proxy rotation, VPNs, and shared networks. + +Requires: + SCOPEBLIND_API_KEY – your ScopeBlind API key (from scopeblind.com/t/) + +Optional: + SCOPEBLIND_ENDPOINT – verification endpoint (default: https://api.scopeblind.com) + +Usage (proxy config.yaml): + litellm_settings: + callbacks: ["scopeblind"] + + environment_variables: + SCOPEBLIND_API_KEY: "sb_..." + +Or programmatically: + import litellm + litellm.callbacks = ["scopeblind"] + +Docs: https://scopeblind.com/docs +""" + +import os +import traceback +from typing import Any, Dict, List, Optional, Union + +import litellm +from litellm._logging import verbose_logger +from litellm.integrations.custom_logger import CustomLogger +from litellm.llms.custom_httpx.http_handler import ( + HTTPHandler, + get_async_httpx_client, + httpxSpecialProvider, +) + + +def _get_utc_datetime(): + import datetime as dt + from datetime import datetime + + if hasattr(dt, "UTC"): + return datetime.now(dt.UTC) + else: + return datetime.utcnow() + + +class ScopeBlindLogger(CustomLogger): + """ + ScopeBlind callback for device-level rate limiting. + + Extracts DPoP proof headers from incoming proxy requests, + verifies them with the ScopeBlind API, and logs per-device + usage data (model, cost, tokens) for abuse detection. + """ + + def __init__(self) -> None: + super().__init__() + self.validate_environment() + self.async_http_handler = get_async_httpx_client( + llm_provider=httpxSpecialProvider.LoggingCallback + ) + self.sync_http_handler = HTTPHandler() + self.scopeblind_endpoint = os.getenv( + "SCOPEBLIND_ENDPOINT", "https://api.scopeblind.com" + ) + self.scopeblind_api_key = os.getenv("SCOPEBLIND_API_KEY", "") + + def validate_environment(self): + """Expects SCOPEBLIND_API_KEY in the environment.""" + missing_keys: List[str] = [] + if os.getenv("SCOPEBLIND_API_KEY", None) is None: + missing_keys.append("SCOPEBLIND_API_KEY") + + if len(missing_keys) > 0: + raise Exception( + "ScopeBlind: Missing keys={} in environment.".format(missing_keys) + ) + + def _build_payload( + self, + kwargs: dict, + response_obj: Any, + start_time, + end_time, + event_type: str, + ) -> dict: + """Build a ScopeBlind event payload from LiteLLM kwargs.""" + call_id = kwargs.get("litellm_call_id") + model = kwargs.get("model") + cost = kwargs.get("response_cost", None) + user = kwargs.get("user", None) + + # Extract device identity from metadata + litellm_params = kwargs.get("litellm_params", {}) + metadata = litellm_params.get("metadata", {}) or {} + headers = metadata.get("headers", {}) or {} + + # DPoP proof and device ID from incoming request headers + dpop_proof = headers.get("x-scopeblind-dpop") or headers.get("dpop") + device_id = headers.get("x-scopeblind-device-id") or headers.get( + "x-device-identity" + ) + + # Token usage + usage = {} + if ( + isinstance(response_obj, litellm.ModelResponse) + or isinstance(response_obj, litellm.EmbeddingResponse) + ) and hasattr(response_obj, "usage"): + usage = { + "prompt_tokens": response_obj["usage"].get("prompt_tokens", 0), + "completion_tokens": response_obj["usage"].get( + "completion_tokens", 0 + ), + "total_tokens": response_obj["usage"].get("total_tokens", 0), + } + + # If no user provided, try API key user ID + if user is None: + user = metadata.get("user_api_key_user_id") + + return { + "event": event_type, + "litellm_call_id": call_id, + "model": model, + "user": user, + "cost": cost, + "usage": usage, + "dpop_proof": dpop_proof, + "device_id": device_id, + "timestamp": _get_utc_datetime().isoformat(), + "start_time": str(start_time), + "end_time": str(end_time), + } + + async def _send_event(self, payload: dict) -> None: + """Send an event to the ScopeBlind API.""" + try: + url = f"{self.scopeblind_endpoint}/v1/proxy/events" + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {self.scopeblind_api_key}", + } + await self.async_http_handler.post( + url=url, + json=payload, + headers=headers, + ) + except Exception as e: + verbose_logger.debug( + "ScopeBlind: Failed to send event: %s", str(e) + ) + + def _send_event_sync(self, payload: dict) -> None: + """Send an event synchronously.""" + try: + url = f"{self.scopeblind_endpoint}/v1/proxy/events" + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {self.scopeblind_api_key}", + } + self.sync_http_handler.post( + url=url, + json=payload, + headers=headers, + ) + except Exception as e: + verbose_logger.debug( + "ScopeBlind: Failed to send event (sync): %s", str(e) + ) + + async def async_log_success_event( + self, kwargs, response_obj, start_time, end_time + ): + """Log successful LLM call with device identity.""" + try: + payload = self._build_payload( + kwargs, response_obj, start_time, end_time, "llm_call_success" + ) + # Only send if we have device identity info + if payload.get("dpop_proof") or payload.get("device_id"): + await self._send_event(payload) + else: + verbose_logger.debug( + "ScopeBlind: No device identity headers found, skipping event" + ) + except Exception as e: + verbose_logger.debug( + "ScopeBlind: Error logging success event: %s\n%s", + str(e), + traceback.format_exc(), + ) + + def log_success_event(self, kwargs, response_obj, start_time, end_time): + """Log successful LLM call synchronously.""" + try: + payload = self._build_payload( + kwargs, response_obj, start_time, end_time, "llm_call_success" + ) + if payload.get("dpop_proof") or payload.get("device_id"): + self._send_event_sync(payload) + except Exception as e: + verbose_logger.debug( + "ScopeBlind: Error logging success event (sync): %s", str(e) + ) + + async def async_log_failure_event( + self, kwargs, response_obj, start_time, end_time + ): + """Log failed LLM call with device identity.""" + try: + payload = self._build_payload( + kwargs, response_obj, start_time, end_time, "llm_call_failure" + ) + if payload.get("dpop_proof") or payload.get("device_id"): + await self._send_event(payload) + except Exception as e: + verbose_logger.debug( + "ScopeBlind: Error logging failure event: %s", str(e) + ) + + def log_failure_event(self, kwargs, response_obj, start_time, end_time): + """Log failed LLM call synchronously.""" + try: + payload = self._build_payload( + kwargs, response_obj, start_time, end_time, "llm_call_failure" + ) + if payload.get("dpop_proof") or payload.get("device_id"): + self._send_event_sync(payload) + except Exception as e: + verbose_logger.debug( + "ScopeBlind: Error logging failure event (sync): %s", str(e) + ) diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 6f587abcdf1..8315db400aa 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -155,6 +155,7 @@ from ..integrations.lunary import LunaryLogger from ..integrations.openmeter import OpenMeterLogger from ..integrations.opik.opik import OpikLogger from ..integrations.posthog import PostHogLogger +from ..integrations.scopeblind import ScopeBlindLogger from ..integrations.prompt_layer import PromptLayerLogger from ..integrations.s3 import S3Logger from ..integrations.s3_v2 import S3Logger as S3V2Logger @@ -3528,6 +3529,8 @@ def set_callbacks(callback_list, function_id=None): # noqa: PLR0915 ) elif callback == "openmeter": openMeterLogger = OpenMeterLogger() + elif callback == "scopeblind": + scopeBlindLogger = ScopeBlindLogger() elif callback == "datadog": dataDogLogger = DataDogLogger() elif callback == "dynamodb": @@ -3590,6 +3593,14 @@ def _init_custom_logger_compatible_class( # noqa: PLR0915 _openmeter_logger = OpenMeterLogger() _in_memory_loggers.append(_openmeter_logger) return _openmeter_logger # type: ignore + elif logging_integration == "scopeblind": + for callback in _in_memory_loggers: + if isinstance(callback, ScopeBlindLogger): + return callback # type: ignore + + _scopeblind_logger = ScopeBlindLogger() + _in_memory_loggers.append(_scopeblind_logger) + return _scopeblind_logger # type: ignore elif logging_integration == "posthog": for callback in _in_memory_loggers: if isinstance(callback, PostHogLogger): @@ -4221,6 +4232,10 @@ def get_custom_logger_compatible_class( # noqa: PLR0915 for callback in _in_memory_loggers: if isinstance(callback, OpenMeterLogger): return callback + elif logging_integration == "scopeblind": + for callback in _in_memory_loggers: + if isinstance(callback, ScopeBlindLogger): + return callback elif logging_integration == "braintrust": from litellm.integrations.braintrust_logging import BraintrustLogger diff --git a/tests/test_litellm/integrations/test_scopeblind.py b/tests/test_litellm/integrations/test_scopeblind.py new file mode 100644 index 00000000000..29b5d9e4a1f --- /dev/null +++ b/tests/test_litellm/integrations/test_scopeblind.py @@ -0,0 +1,205 @@ +""" +Tests for the ScopeBlind integration. + +Tests that the ScopeBlind callback correctly: +1. Validates environment variables +2. Builds payloads from LiteLLM kwargs +3. Skips events when no device identity headers are present +4. Sends events when device identity headers are present +""" + +import os +import sys +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +sys.path.insert( + 0, os.path.abspath("../..") +) # Adds the parent directory to the system-path + + +class TestScopeBlindValidation: + """Test environment validation.""" + + def test_missing_api_key_raises(self): + """ScopeBlindLogger raises if SCOPEBLIND_API_KEY is not set.""" + with patch.dict(os.environ, {}, clear=True): + # Remove the key if it exists + os.environ.pop("SCOPEBLIND_API_KEY", None) + from litellm.integrations.scopeblind import ScopeBlindLogger + + with pytest.raises(Exception, match="Missing keys"): + ScopeBlindLogger() + + def test_valid_environment(self): + """ScopeBlindLogger initializes with valid env vars.""" + with patch.dict( + os.environ, + {"SCOPEBLIND_API_KEY": "sb_test_key"}, + ): + from litellm.integrations.scopeblind import ScopeBlindLogger + + logger = ScopeBlindLogger() + assert logger.scopeblind_api_key == "sb_test_key" + assert logger.scopeblind_endpoint == "https://api.scopeblind.com" + + def test_custom_endpoint(self): + """ScopeBlindLogger respects SCOPEBLIND_ENDPOINT env var.""" + with patch.dict( + os.environ, + { + "SCOPEBLIND_API_KEY": "sb_test_key", + "SCOPEBLIND_ENDPOINT": "https://custom.scopeblind.dev", + }, + ): + from litellm.integrations.scopeblind import ScopeBlindLogger + + logger = ScopeBlindLogger() + assert logger.scopeblind_endpoint == "https://custom.scopeblind.dev" + + +class TestScopeBlindPayload: + """Test payload building.""" + + @pytest.fixture + def logger(self): + with patch.dict( + os.environ, + {"SCOPEBLIND_API_KEY": "sb_test_key"}, + ): + from litellm.integrations.scopeblind import ScopeBlindLogger + + return ScopeBlindLogger() + + def test_build_payload_with_device_headers(self, logger): + """Payload includes DPoP proof and device ID from headers.""" + kwargs = { + "litellm_call_id": "call-123", + "model": "gpt-4", + "response_cost": 0.03, + "user": "user-456", + "litellm_params": { + "metadata": { + "headers": { + "x-scopeblind-dpop": "eyJ0eXAiOiJkcG9wK2p3dCJ9.test", + "x-scopeblind-device-id": "device-789", + } + } + }, + } + response_obj = MagicMock() + response_obj.__class__.__name__ = "ModelResponse" + + payload = logger._build_payload( + kwargs, response_obj, "2024-01-01T00:00:00", "2024-01-01T00:00:01", "llm_call_success" + ) + + assert payload["event"] == "llm_call_success" + assert payload["model"] == "gpt-4" + assert payload["user"] == "user-456" + assert payload["cost"] == 0.03 + assert payload["dpop_proof"] == "eyJ0eXAiOiJkcG9wK2p3dCJ9.test" + assert payload["device_id"] == "device-789" + assert payload["litellm_call_id"] == "call-123" + + def test_build_payload_without_device_headers(self, logger): + """Payload has None for device fields when no headers present.""" + kwargs = { + "litellm_call_id": "call-123", + "model": "gpt-4", + "litellm_params": {"metadata": {}}, + } + response_obj = MagicMock() + + payload = logger._build_payload( + kwargs, response_obj, "2024-01-01T00:00:00", "2024-01-01T00:00:01", "llm_call_success" + ) + + assert payload["dpop_proof"] is None + assert payload["device_id"] is None + + def test_build_payload_with_x_device_identity_header(self, logger): + """Payload extracts X-Device-Identity header.""" + kwargs = { + "litellm_call_id": "call-123", + "model": "gpt-4", + "litellm_params": { + "metadata": { + "headers": { + "x-device-identity": 'ScopeBlind/1.0; info="https://scopeblind.com/verify-agent"', + } + } + }, + } + response_obj = MagicMock() + + payload = logger._build_payload( + kwargs, response_obj, "2024-01-01T00:00:00", "2024-01-01T00:00:01", "llm_call_success" + ) + + assert payload["device_id"] == 'ScopeBlind/1.0; info="https://scopeblind.com/verify-agent"' + + +class TestScopeBlindEventSending: + """Test that events are sent or skipped correctly.""" + + @pytest.fixture + def logger(self): + with patch.dict( + os.environ, + {"SCOPEBLIND_API_KEY": "sb_test_key"}, + ): + from litellm.integrations.scopeblind import ScopeBlindLogger + + instance = ScopeBlindLogger() + instance.sync_http_handler = MagicMock() + return instance + + def test_skips_event_without_device_identity(self, logger): + """log_success_event should not send when no device headers.""" + kwargs = { + "litellm_call_id": "call-123", + "model": "gpt-4", + "litellm_params": {"metadata": {}}, + } + response_obj = MagicMock() + + logger.log_success_event( + kwargs, response_obj, "2024-01-01T00:00:00", "2024-01-01T00:00:01" + ) + + logger.sync_http_handler.post.assert_not_called() + + def test_sends_event_with_device_identity(self, logger): + """log_success_event should send when DPoP header is present.""" + kwargs = { + "litellm_call_id": "call-123", + "model": "gpt-4", + "litellm_params": { + "metadata": { + "headers": { + "x-scopeblind-dpop": "eyJ0eXAiOiJkcG9wK2p3dCJ9.test", + } + } + }, + } + response_obj = MagicMock() + + logger.log_success_event( + kwargs, response_obj, "2024-01-01T00:00:00", "2024-01-01T00:00:01" + ) + + logger.sync_http_handler.post.assert_called_once() + call_kwargs = logger.sync_http_handler.post.call_args + assert "scopeblind" in call_kwargs.kwargs["url"] + + +class TestScopeBlindRegistration: + """Test that ScopeBlind is registered as a named callback.""" + + def test_scopeblind_in_callbacks_literal(self): + """scopeblind should be in the known callbacks list.""" + import litellm + + assert "scopeblind" in litellm._known_custom_logger_compatible_callbacks