From bda035b98987f05cbdd00be8cb2600bad0055c1f Mon Sep 17 00:00:00 2001 From: yassin Date: Wed, 14 Oct 2026 04:36:41 +0000 Subject: [PATCH] feat(control_plane): splunk hec and microsoft sentinel audit sinks Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/proxy/control_plane/audit/s3.py | 7 +- litellm/proxy/control_plane/audit/sentinel.py | 193 ++++++++++++++++++ litellm/proxy/control_plane/audit/splunk.py | 84 ++++++++ .../proxy/control_plane/audit/transport.py | 28 +++ litellm/proxy/control_plane/container.py | 24 +++ litellm/proxy/control_plane/settings.py | 125 +++++++++++- .../control_plane/audit/test_sentinel.py | 152 ++++++++++++++ .../proxy/control_plane/audit/test_splunk.py | 104 ++++++++++ .../control_plane/audit/test_transport.py | 40 ++++ .../unit/proxy/control_plane/test_settings.py | 67 +++++- 10 files changed, 810 insertions(+), 14 deletions(-) create mode 100644 litellm/proxy/control_plane/audit/sentinel.py create mode 100644 litellm/proxy/control_plane/audit/splunk.py create mode 100644 litellm/proxy/control_plane/audit/transport.py create mode 100644 tests/unit/proxy/control_plane/audit/test_sentinel.py create mode 100644 tests/unit/proxy/control_plane/audit/test_splunk.py create mode 100644 tests/unit/proxy/control_plane/audit/test_transport.py diff --git a/litellm/proxy/control_plane/audit/s3.py b/litellm/proxy/control_plane/audit/s3.py index 3d4fcd97ed1..ef7b590eaff 100644 --- a/litellm/proxy/control_plane/audit/s3.py +++ b/litellm/proxy/control_plane/audit/s3.py @@ -39,12 +39,15 @@ class Signer(Protocol): @dataclass(frozen=True, slots=True) class HttpxObjectStore: - """Holds the handler, not its client: the shared cache may close and replace the client between batches.""" + """Holds the handler, not its client: the shared cache may close and replace the client between batches. Never + follows a redirect, a sink URL that bounces to plain http would otherwise carry the credential and the records there.""" handler: AsyncHTTPHandler async def put(self, url: str, content: bytes, headers: Mapping[str, str], timeout: float) -> httpx.Response: - return await self.handler.client.put(url, content=content, headers=dict(headers), timeout=timeout) + return await self.handler.client.put( + url, content=content, headers=dict(headers), timeout=timeout, follow_redirects=False + ) @dataclass(frozen=True, slots=True) diff --git a/litellm/proxy/control_plane/audit/sentinel.py b/litellm/proxy/control_plane/audit/sentinel.py new file mode 100644 index 00000000000..7d7ae91d123 --- /dev/null +++ b/litellm/proxy/control_plane/audit/sentinel.py @@ -0,0 +1,193 @@ +"""Microsoft Sentinel sink through the Azure Monitor Logs Ingestion API: an Entra client-credentials token, then one +JSON array of flattened rows per POST to the data collection rule's stream, split at the API's 1 MB limit +(https://learn.microsoft.com/en-us/azure/azure-monitor/logs/logs-ingestion-api-overview).""" + +import json +import time +from collections.abc import Callable, Iterable, Mapping +from dataclasses import dataclass +from itertools import accumulate, groupby +from typing import Final, Protocol, TypeAlias +from urllib.parse import urlencode + +import httpx +from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError + +from litellm.proxy.control_plane.audit.ocsf import AuditRecord, section +from litellm.proxy.control_plane.audit.pipeline import Batch, Delivered, DeliveryResult, Undelivered +from litellm.proxy.control_plane.audit.sinks import ocsf_line +from litellm.proxy.control_plane.audit.transport import Poster +from litellm.proxy.control_plane.settings import SentinelSettings + +API_VERSION: Final = "2023-01-01" +MAX_PAYLOAD_BYTES: Final = 1_000_000 +FIXED_WIDTH_COLUMNS: Final = frozenset( + { + "TimeGenerated", + "ClassUid", + "ActivityId", + "StatusId", + "SeverityId", + "ResponseCode", + "Sequence", + "Hash", + "PrevHash", + "TraceId", + } +) +TOKEN_SCOPE: Final = "https://monitor.azure.com/.default" +REFRESH_MARGIN_SECONDS: Final = 60.0 +Row: TypeAlias = Mapping[str, object] + + +class TokenSource(Protocol): + async def __call__(self) -> str: ... + + +class TokenRefused(Exception): + def __init__(self, status: int) -> None: + super().__init__(f"HTTP {status} from the token endpoint") + self.status = status + + +class TokenReply(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + access_token: str + expires_in: float = 3600.0 + + +_TOKEN_REPLY: Final = TypeAdapter(TokenReply) + + +@dataclass(slots=True) +class ClientCredentials: + """Entra ``client_credentials`` grant for the Azure Monitor scope. The secret only ever travels in the form body.""" + + settings: SentinelSettings + post: Poster + now: Callable[[], float] = time.time + token: str = "" + expires_at: float = 0.0 + + async def __call__(self) -> str: + if self.token and self.now() < self.expires_at - REFRESH_MARGIN_SECONDS: + return self.token + form: Final = urlencode( + { + "client_id": self.settings.client_id, + "client_secret": self.settings.client_secret.get_secret_value(), + "scope": TOKEN_SCOPE, + "grant_type": "client_credentials", + } + ).encode() + url: Final = f"{self.settings.authority_host}/{self.settings.tenant_id}/oauth2/v2.0/token" + response: Final = await self.post.post( + url, form, {"Content-Type": "application/x-www-form-urlencoded"}, self.settings.timeout_seconds + ) + if response.status_code != 200: + raise TokenRefused(response.status_code) + reply: Final = _TOKEN_REPLY.validate_json(response.content) + self.token = reply.access_token + self.expires_at = self.now() + reply.expires_in + return self.token + + +def sentinel_row(record: AuditRecord) -> Row: + """One flat row per record for a custom table, plus the whole OCSF object in ``Ocsf`` for KQL ``parse_json``.""" + event: Final = record.event + body: Final = record.body + litellm_block: Final = section(body, "unmapped", "litellm") + return { + "TimeGenerated": event.at.isoformat(), + "ClassUid": body.get("class_uid"), + "ClassName": body.get("class_name"), + "ActivityId": body.get("activity_id"), + "ActivityName": body.get("activity_name"), + "StatusId": body.get("status_id"), + "SeverityId": body.get("severity_id"), + "ActorUid": event.actor, + "ActorEmail": event.actor_email, + "DataplaneId": event.dataplane_id, + "HttpMethod": event.method, + "UrlPath": event.path, + "ResponseCode": event.status, + "Decision": event.decision, + "Detail": event.detail, + "TraceId": event.trace_id, + "Sequence": record.sequence, + "Hash": record.hash, + "PrevHash": litellm_block.get("prev_hash"), + "Ocsf": ocsf_line(record), + } + + +def _encoded(row: Row) -> bytes: + return json.dumps(row, separators=(",", ":"), default=str).encode() + + +def _fitted(row: Row, limit: int) -> bytes: + """A row that alone would exceed the request limit first loses ``Ocsf`` and ``Detail``, the free text, and if a flat + column is still too long keeps only its fixed-width columns, so the batch can never be stuck behind one record.""" + whole: Final = _encoded(row) + if len(whole) + 2 <= limit: + return whole + trimmed: Final = _encoded({**row, "Detail": None, "Ocsf": None, "Truncated": True}) + if len(trimmed) + 2 <= limit: + return trimmed + return _encoded({**{key: value for key, value in row.items() if key in FIXED_WIDTH_COLUMNS}, "Truncated": True}) + + +def chunks(rows: Iterable[Row], limit: int = MAX_PAYLOAD_BYTES) -> tuple[bytes, ...]: + """JSON arrays of rows, each at most ``limit`` bytes.""" + encoded: Final = tuple(_fitted(row, limit) for row in rows) + sizes: Final = tuple(len(item) + 1 for item in encoded) + fills: Final = tuple(accumulate(sizes, _step(limit), initial=(0, 1)))[1:] + groups: Final = groupby(zip((chunk for chunk, _ in fills), encoded, strict=True), key=lambda pair: pair[0]) + return tuple(b"[" + b",".join(item for _, item in members) + b"]" for _, members in groups) + + +def _step(limit: int) -> Callable[[tuple[int, int], int], tuple[int, int]]: + def step(fill: tuple[int, int], size: int) -> tuple[int, int]: + chunk, used = fill + return (chunk, used + size) if used + size <= limit else (chunk + 1, size + 1) + + return step + + +@dataclass(frozen=True, slots=True) +class SentinelSink: + """Posts are at least once: a batch whose later chunk fails is retried whole, so Sentinel may hold a duplicate of + the earlier chunks, deduplicable on ``Hash``.""" + + settings: SentinelSettings + post: Poster + token: TokenSource + name: str = "sentinel" + + @property + def url(self) -> str: + return ( + f"{self.settings.endpoint.rstrip('/')}/dataCollectionRules/{self.settings.dcr_immutable_id}" + f"/streams/{self.settings.stream_name}?api-version={API_VERSION}" + ) + + async def deliver(self, batch: Batch) -> DeliveryResult: + try: + bearer: Final = await self.token() + except TokenRefused as refused: + return Undelivered(f"entra token refused: {refused}") + except (httpx.HTTPError, ValidationError) as error: + return Undelivered(f"{type(error).__name__} getting the entra token") + headers: Final = {"Authorization": f"Bearer {bearer}", "Content-Type": "application/json"} + for payload in chunks(sentinel_row(record) for record in batch): + try: + response = await self.post.post(self.url, payload, headers, self.settings.timeout_seconds) + except httpx.HTTPError as error: + return Undelivered(f"{type(error).__name__} posting to the logs ingestion api") + if response.status_code != 204: + return Undelivered(f"HTTP {response.status_code} from the logs ingestion api") + return Delivered(len(batch)) + + async def close(self) -> None: + return None diff --git a/litellm/proxy/control_plane/audit/splunk.py b/litellm/proxy/control_plane/audit/splunk.py new file mode 100644 index 00000000000..341e1d82e0b --- /dev/null +++ b/litellm/proxy/control_plane/audit/splunk.py @@ -0,0 +1,84 @@ +"""Splunk HTTP Event Collector sink: one HEC event per audit record, the OCSF object as the event body, newline +separated in one POST per batch (https://docs.splunk.com/Documentation/Splunk/latest/Data/FormateventsforHTTPEventCollector).""" + +import json +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Final + +import httpx +from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError + +from litellm.proxy.control_plane.audit.ocsf import AuditRecord +from litellm.proxy.control_plane.audit.pipeline import Batch, Delivered, DeliveryResult, Undelivered +from litellm.proxy.control_plane.audit.transport import Poster +from litellm.proxy.control_plane.settings import SplunkHecSettings + +EVENT_PATH: Final = "/services/collector/event" + + +class HecReply(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + text: str = "" + code: int = -1 + + +_REPLY: Final = TypeAdapter(HecReply) + + +def hec_event(record: AuditRecord, settings: SplunkHecSettings) -> Mapping[str, object]: + """``sourcetype`` is ``ocsf:`` so a CIM or OCSF add-on can pick the class without parsing the body.""" + index: Final = {"index": settings.index} if settings.index is not None else {} + return { + "time": record.event.at.timestamp(), + "source": settings.source, + "sourcetype": f"ocsf:{record.event.kind}", + **index, + "event": record.body, + } + + +def hec_payload(batch: Batch, settings: SplunkHecSettings) -> bytes: + return "\n".join( + json.dumps(hec_event(record, settings), separators=(",", ":"), default=str) for record in batch + ).encode() + + +def _reply(response: httpx.Response) -> HecReply: + """HEC reports its own result in the body, ``code`` 0 is the only success, so a 200 without it is not one.""" + try: + return _REPLY.validate_json(response.content) + except ValidationError: + return HecReply(text="no HEC reply body") + + +def _reply_text(response: httpx.Response) -> str: + reply: Final = _reply(response) + return f"{reply.text} (code {reply.code})" + + +@dataclass(frozen=True, slots=True) +class SplunkHecSink: + settings: SplunkHecSettings + post: Poster + name: str = "splunk_hec" + + async def deliver(self, batch: Batch) -> DeliveryResult: + url: Final = f"{self.settings.url.rstrip('/')}{EVENT_PATH}" + headers: Final = { + "Authorization": f"Splunk {self.settings.token.get_secret_value()}", + "Content-Type": "application/json", + } + try: + response: Final = await self.post.post( + url, hec_payload(batch, self.settings), headers, self.settings.timeout_seconds + ) + except httpx.HTTPError as error: + return Undelivered(f"{type(error).__name__} posting to splunk hec") + if response.status_code == 200 and _reply(response).code == 0: + return Delivered(len(batch)) + return Undelivered(f"HTTP {response.status_code} from splunk hec: {_reply_text(response)}") + + async def close(self) -> None: + return None diff --git a/litellm/proxy/control_plane/audit/transport.py b/litellm/proxy/control_plane/audit/transport.py new file mode 100644 index 00000000000..4f4624962fe --- /dev/null +++ b/litellm/proxy/control_plane/audit/transport.py @@ -0,0 +1,28 @@ +"""The one HTTP call the HTTP sinks make, as a protocol so tests script the answer.""" + +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Protocol + +import httpx + +from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler + + +class Poster(Protocol): + """Raises only httpx errors and never retries on its own, the worker owns retries.""" + + async def post(self, url: str, content: bytes, headers: Mapping[str, str], timeout: float) -> httpx.Response: ... + + +@dataclass(frozen=True, slots=True) +class HttpxPoster: + """Holds the handler, not its client: the shared cache may close and replace the client between batches. Never + follows a redirect, a sink URL that bounces to plain http would otherwise carry the credential and the records there.""" + + handler: AsyncHTTPHandler + + async def post(self, url: str, content: bytes, headers: Mapping[str, str], timeout: float) -> httpx.Response: + return await self.handler.client.post( + url, content=content, headers=dict(headers), timeout=timeout, follow_redirects=False + ) diff --git a/litellm/proxy/control_plane/container.py b/litellm/proxy/control_plane/container.py index 9807fb9ae0b..f03916c6f4a 100644 --- a/litellm/proxy/control_plane/container.py +++ b/litellm/proxy/control_plane/container.py @@ -22,7 +22,10 @@ from litellm.proxy.control_plane.audit.log import AuditLog, build_audit_log from litellm.proxy.control_plane.audit.ocsf import Product from litellm.proxy.control_plane.audit.pipeline import AuditPipeline, AuditSink, build_pipeline from litellm.proxy.control_plane.audit.s3 import HttpxObjectStore, S3ObjectLockSink, SigV4Signer +from litellm.proxy.control_plane.audit.sentinel import ClientCredentials, SentinelSink from litellm.proxy.control_plane.audit.sinks import LogSink, OtlpLogSink +from litellm.proxy.control_plane.audit.splunk import SplunkHecSink +from litellm.proxy.control_plane.audit.transport import HttpxPoster from litellm.proxy.control_plane.auth.jwt import JwksKeyLookup, SessionTokenCodec, TokenVerifier, build_token_verifier from litellm.proxy.control_plane.auth.login import LoginService, LoginThrottle from litellm.proxy.control_plane.auth.rbac import ( @@ -142,12 +145,33 @@ def _environment_region(aws: BaseAWSLLM) -> str: return configured or _REGION.validate_python(aws.get_aws_region_name_for_non_llm_api_calls()) +def _splunk_sink(settings: ControlPlaneSettings) -> AuditSink: + splunk: Final = settings.audit_splunk + if splunk is None: + raise ValueError("CONTROL_PLANE_AUDIT_SINKS names splunk_hec but its url and token are not both set.") + return SplunkHecSink(splunk, HttpxPoster(_shared_http_handler(settings))) + + +def _sentinel_sink(settings: ControlPlaneSettings) -> AuditSink: + sentinel: Final = settings.audit_sentinel + if sentinel is None: + raise ValueError( + "CONTROL_PLANE_AUDIT_SINKS names sentinel but its endpoint, rule, tenant and app are not all set." + ) + poster: Final = HttpxPoster(_shared_http_handler(settings)) + return SentinelSink(sentinel, poster, ClientCredentials(sentinel, poster)) + + def _sink(kind: str, settings: ControlPlaneSettings) -> AuditSink: match kind: case "otlp": return OtlpLogSink(build_log_exporter(settings), build_resource(settings)) case "s3": return _s3_sink(settings) + case "splunk_hec": + return _splunk_sink(settings) + case "sentinel": + return _sentinel_sink(settings) case _: return LogSink() diff --git a/litellm/proxy/control_plane/settings.py b/litellm/proxy/control_plane/settings.py index ecc7f0931ab..ea9eb79f7ef 100644 --- a/litellm/proxy/control_plane/settings.py +++ b/litellm/proxy/control_plane/settings.py @@ -13,8 +13,30 @@ from litellm.litellm_core_utils.oauth_token_exchange import TokenExchangeProfile from litellm.types.proxy.control_plane.auth import PASSWORD_MAX_LENGTH, USERNAME_MAX_LENGTH DEFAULT_SESSION_TTL_SECONDS: Final = 12 * 60 * 60 -AuditSinkKind: TypeAlias = Literal["log", "otlp", "s3"] +AuditSinkKind: TypeAlias = Literal["log", "otlp", "s3", "splunk_hec", "sentinel"] ObjectLockMode: TypeAlias = Literal["COMPLIANCE", "GOVERNANCE"] +_SINK_URL_FIELDS: Final = ( + ("audit_s3_endpoint_url", "audit_s3_allow_insecure_http"), + ("audit_splunk_url", "audit_splunk_allow_insecure_http"), + ("audit_sentinel_endpoint", None), + ("audit_sentinel_authority_host", None), +) + + +def _plain_http(url: object) -> bool: + return isinstance(url, str) and urlsplit(url).scheme.lower() != "https" + + +_REQUIRED_SINK_FIELDS: Final = ( + ("s3", "audit_s3_bucket"), + ("splunk_hec", "audit_splunk_url"), + ("splunk_hec", "audit_splunk_token"), + ("sentinel", "audit_sentinel_endpoint"), + ("sentinel", "audit_sentinel_dcr_immutable_id"), + ("sentinel", "audit_sentinel_tenant_id"), + ("sentinel", "audit_sentinel_client_id"), + ("sentinel", "audit_sentinel_client_secret"), +) DEFAULT_DATAPLANE_TIMEOUT_SECONDS: Final = 10.0 @@ -31,6 +53,29 @@ class S3SinkSettings: timeout_seconds: float +@dataclass(frozen=True, slots=True) +class SplunkHecSettings: + url: str + token: SecretStr + index: str | None + source: str + timeout_seconds: float + + +@dataclass(frozen=True, slots=True) +class SentinelSettings: + """Azure Monitor Logs Ingestion API target plus the Entra app that may write to it.""" + + endpoint: str + dcr_immutable_id: str + stream_name: str + tenant_id: str + client_id: str + client_secret: SecretStr + authority_host: str + timeout_seconds: float + + @dataclass(frozen=True, slots=True) class OidcSettings: """External IdP used to verify caller JWTs and, later, to exchange them for dataplane tokens.""" @@ -140,21 +185,83 @@ class ControlPlaneSettings(BaseSettings): audit_s3_object_lock_mode: ObjectLockMode | None = None audit_s3_retention_days: int = Field(default=365, ge=1) audit_s3_timeout_seconds: float = 30.0 + audit_splunk_url: str | None = None + audit_splunk_allow_insecure_http: bool = False + audit_splunk_token: SecretStr | None = None + audit_splunk_index: str | None = None + audit_splunk_source: str = "litellm-control-plane" + audit_splunk_timeout_seconds: float = 30.0 + audit_sentinel_endpoint: str | None = None + audit_sentinel_dcr_immutable_id: str | None = None + audit_sentinel_stream_name: str = "Custom-LiteLLMControlPlaneAudit_CL" + audit_sentinel_tenant_id: str | None = None + audit_sentinel_client_id: str | None = None + audit_sentinel_client_secret: SecretStr | None = None + audit_sentinel_authority_host: str = "https://login.microsoftonline.com" + audit_sentinel_timeout_seconds: float = 30.0 @model_validator(mode="after") - def _s3_sink_needs_a_bucket(self) -> "ControlPlaneSettings": - if "s3" in self.audit_sinks and self.audit_s3_bucket is None: - raise ValueError("CONTROL_PLANE_AUDIT_SINKS names s3 but CONTROL_PLANE_AUDIT_S3_BUCKET is not set.") + def _listed_sinks_have_their_settings(self) -> "ControlPlaneSettings": + missing: Final = tuple( + f"CONTROL_PLANE_{name.upper()}" + for kind, name in _REQUIRED_SINK_FIELDS + if kind in self.audit_sinks and getattr(self, name) is None + ) + if missing: + raise ValueError( + f"CONTROL_PLANE_AUDIT_SINKS lists a sink whose settings are not set: {', '.join(missing)}." + ) return self @model_validator(mode="after") - def _s3_endpoint_is_https_unless_opted_out(self) -> "ControlPlaneSettings": - endpoint: Final = self.audit_s3_endpoint_url - if endpoint is None or self.audit_s3_allow_insecure_http or urlsplit(endpoint).scheme.lower() == "https": + def _sink_urls_are_https_unless_opted_out(self) -> "ControlPlaneSettings": + """Every sink URL carries a credential (SigV4, HEC token, bearer, client secret), so plain http is refused. + Only the stores that can live in-cluster (S3 compatible, Splunk) have an opt-out.""" + insecure: Final = tuple( + (name, flag) + for name, flag in _SINK_URL_FIELDS + if _plain_http(getattr(self, name)) and not (flag is not None and getattr(self, flag)) + ) + if not insecure: return self + name, flag = insecure[0] + remedy: Final = f" or set CONTROL_PLANE_{flag.upper()}=true for an in-cluster endpoint" if flag else "" raise ValueError( - f"audit s3 endpoint {endpoint!r} is not https, audit records and the SigV4 authorization would travel " - "unencrypted. Use https or set CONTROL_PLANE_AUDIT_S3_ALLOW_INSECURE_HTTP=true for an in-cluster store." + f"CONTROL_PLANE_{name.upper()} is not https, audit records and the sink credential would travel " + f"unencrypted. Use https{remedy}." + ) + + @property + def audit_splunk(self) -> SplunkHecSettings | None: + if self.audit_splunk_url is None or self.audit_splunk_token is None: + return None + return SplunkHecSettings( + url=self.audit_splunk_url, + token=self.audit_splunk_token, + index=self.audit_splunk_index, + source=self.audit_splunk_source, + timeout_seconds=self.audit_splunk_timeout_seconds, + ) + + @property + def audit_sentinel(self) -> SentinelSettings | None: + if ( + self.audit_sentinel_endpoint is None + or self.audit_sentinel_dcr_immutable_id is None + or self.audit_sentinel_tenant_id is None + or self.audit_sentinel_client_id is None + or self.audit_sentinel_client_secret is None + ): + return None + return SentinelSettings( + endpoint=self.audit_sentinel_endpoint, + dcr_immutable_id=self.audit_sentinel_dcr_immutable_id, + stream_name=self.audit_sentinel_stream_name, + tenant_id=self.audit_sentinel_tenant_id, + client_id=self.audit_sentinel_client_id, + client_secret=self.audit_sentinel_client_secret, + authority_host=self.audit_sentinel_authority_host.rstrip("/"), + timeout_seconds=self.audit_sentinel_timeout_seconds, ) @property diff --git a/tests/unit/proxy/control_plane/audit/test_sentinel.py b/tests/unit/proxy/control_plane/audit/test_sentinel.py new file mode 100644 index 00000000000..e4489eac7b4 --- /dev/null +++ b/tests/unit/proxy/control_plane/audit/test_sentinel.py @@ -0,0 +1,152 @@ +import json +from collections.abc import Mapping +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import Final +from urllib.parse import parse_qs + +import httpx +import pytest +from pydantic import SecretStr + +from litellm.proxy.control_plane.audit.event import AuditEvent +from litellm.proxy.control_plane.audit.ocsf import AuditRecord, ChainState, Product, chain +from litellm.proxy.control_plane.audit.pipeline import Batch, Delivered, Undelivered +from litellm.proxy.control_plane.audit.sentinel import ClientCredentials, SentinelSink, chunks, sentinel_row +from litellm.proxy.control_plane.settings import SentinelSettings + +T0 = datetime(2026, 10, 13, 12, 0, tzinfo=UTC) +SETTINGS: Final = SentinelSettings( + endpoint="https://dce.ingest.monitor.azure.com/", + dcr_immutable_id="dcr-abc", + stream_name="Custom-LiteLLMControlPlaneAudit_CL", + tenant_id="tenant", + client_id="app", + client_secret=SecretStr("app-secret"), + authority_host="https://login.microsoftonline.com", + timeout_seconds=4.0, +) +INGEST_URL: Final = ( + "https://dce.ingest.monitor.azure.com/dataCollectionRules/dcr-abc/streams/Custom-LiteLLMControlPlaneAudit_CL" + "?api-version=2023-01-01" +) +TOKEN_URL: Final = "https://login.microsoftonline.com/tenant/oauth2/v2.0/token" + + +@dataclass +class FakePoster: # mutable-ok: scripts responses and records posts for assertions + outcomes: list[httpx.Response | Exception] + posts: list[tuple[str, bytes, Mapping[str, str], float]] = field(default_factory=list) + + async def post(self, url: str, content: bytes, headers: Mapping[str, str], timeout: float) -> httpx.Response: + self.posts.append((url, content, headers, timeout)) + outcome = self.outcomes.pop(0) + if isinstance(outcome, Exception): + raise outcome + return outcome + + +@dataclass +class Clock: # mutable-ok: the test moves time forward + now: float = 1_000.0 + + def __call__(self) -> float: + return self.now + + +def _batch(count: int) -> Batch: + state: Final[list[ChainState]] = [ChainState()] + out: Final[list[AuditRecord]] = [] + for index in range(count): + event = AuditEvent( + T0, "alice", "a@x.test", "eu", "POST", f"/key/{index}", "denied", 403, None, "no", "api_activity", "ab" * 16 + ) + state[0], record = chain(event, Product("t"), state[0]) + out.append(record) + return tuple(out) + + +async def _token() -> str: + return "bearer-1" + + +@pytest.mark.asyncio +async def test_a_batch_is_one_json_array_of_flat_rows_posted_to_the_rule_stream_with_the_bearer() -> None: + poster = FakePoster([httpx.Response(204)]) + batch = _batch(2) + assert await SentinelSink(SETTINGS, poster, _token).deliver(batch) == Delivered(2) + ((url, content, headers, timeout),) = poster.posts + assert url == INGEST_URL and timeout == 4.0 + assert headers == {"Authorization": "Bearer bearer-1", "Content-Type": "application/json"} + rows = json.loads(content) + assert rows == [sentinel_row(record) for record in batch] + assert rows[1]["TimeGenerated"] == "2026-10-13T12:00:00+00:00" + assert (rows[1]["Sequence"], rows[1]["PrevHash"], rows[1]["Hash"]) == (2, batch[0].hash, batch[1].hash) + assert (rows[1]["ClassUid"], rows[1]["Decision"], rows[1]["ResponseCode"], rows[1]["ActorEmail"]) == ( + 6003, + "denied", + 403, + "a@x.test", + ) + assert json.loads(rows[1]["Ocsf"]) == batch[1].body and rows[1]["TraceId"] == "ab" * 16 + + +def test_rows_are_split_into_arrays_under_the_api_payload_limit() -> None: + rows = [{"Ocsf": "x" * 300} for _ in range(10)] + parts = chunks(rows, limit=1_000) + assert [len(json.loads(part)) for part in parts] == [3, 3, 3, 1] + assert all(len(part) <= 1_000 for part in parts) + assert [json.loads(part) for part in chunks([], limit=1_000)] == [] + + +def test_a_row_that_alone_exceeds_the_limit_is_sent_without_its_free_text_instead_of_blocking_the_batch() -> None: + rows = [{"Hash": "h1", "Detail": "d", "Ocsf": "x" * 2_000}, {"Hash": "h2", "Detail": "d", "Ocsf": "y"}] + parts = chunks(rows, limit=1_000) + assert all(len(part) <= 1_000 for part in parts) + sent = [row for part in parts for row in json.loads(part)] + assert sent == [ + {"Hash": "h1", "Detail": None, "Ocsf": None, "Truncated": True}, + {"Hash": "h2", "Detail": "d", "Ocsf": "y"}, + ] + + +@pytest.mark.asyncio +async def test_the_entra_token_is_fetched_once_and_renewed_only_near_expiry() -> None: + poster = FakePoster( + [ + httpx.Response(200, json={"access_token": "t1", "expires_in": 600}), + httpx.Response(200, json={"access_token": "t2", "expires_in": 600}), + ] + ) + clock = Clock() + credentials = ClientCredentials(SETTINGS, poster, now=clock) + assert [await credentials() for _ in range(3)] == ["t1", "t1", "t1"] + ((url, form, headers, _),) = poster.posts + assert url == TOKEN_URL and headers == {"Content-Type": "application/x-www-form-urlencoded"} + assert parse_qs(form.decode()) == { + "client_id": ["app"], + "client_secret": ["app-secret"], + "scope": ["https://monitor.azure.com/.default"], + "grant_type": ["client_credentials"], + } + clock.now += 541 + assert await credentials() == "t2" and len(poster.posts) == 2 + + +@pytest.mark.asyncio +async def test_a_refused_token_or_ingestion_error_is_retried_and_names_no_secret() -> None: + poster = FakePoster( + [httpx.Response(401, json={"error": "invalid_client"}), httpx.Response(429), httpx.ConnectError("x")] + ) + refused = await SentinelSink(SETTINGS, poster, ClientCredentials(SETTINGS, poster)).deliver(_batch(1)) + assert refused == Undelivered("entra token refused: HTTP 401 from the token endpoint") + sink = SentinelSink(SETTINGS, poster, _token) + assert await sink.deliver(_batch(1)) == Undelivered("HTTP 429 from the logs ingestion api") + assert await sink.deliver(_batch(1)) == Undelivered("ConnectError posting to the logs ingestion api") + + +def test_a_row_whose_flat_column_alone_exceeds_the_limit_keeps_only_its_fixed_width_columns() -> None: + rows = [{"Hash": "h1", "Sequence": 7, "UrlPath": "/" + "k" * 2_000, "Detail": "d", "Ocsf": "x"}] + parts = chunks(rows, limit=1_000) + assert all(len(part) <= 1_000 for part in parts) + assert [json.loads(part) for part in parts] == [[{"Hash": "h1", "Sequence": 7, "Truncated": True}]] diff --git a/tests/unit/proxy/control_plane/audit/test_splunk.py b/tests/unit/proxy/control_plane/audit/test_splunk.py new file mode 100644 index 00000000000..25c300f5f9a --- /dev/null +++ b/tests/unit/proxy/control_plane/audit/test_splunk.py @@ -0,0 +1,104 @@ +import json +from collections.abc import Mapping +from dataclasses import dataclass, field, replace +from datetime import UTC, datetime +from typing import Final + +import httpx +import pytest +from pydantic import SecretStr + +from litellm.proxy.control_plane.audit.event import AuditEvent +from litellm.proxy.control_plane.audit.ocsf import ChainState, Product, chain +from litellm.proxy.control_plane.audit.pipeline import Batch, Delivered, Undelivered +from litellm.proxy.control_plane.audit.splunk import SplunkHecSink +from litellm.proxy.control_plane.settings import SplunkHecSettings + +T0 = datetime(2026, 10, 13, 12, 0, tzinfo=UTC) +SETTINGS: Final = SplunkHecSettings( + url="https://hec.example.test:8088/", token=SecretStr("hec-token"), index="audit", source="cp", timeout_seconds=4.0 +) + + +@dataclass +class FakePoster: # mutable-ok: scripts responses and records posts for assertions + outcomes: list[httpx.Response | Exception] + posts: list[tuple[str, bytes, Mapping[str, str], float]] = field(default_factory=list) + + async def post(self, url: str, content: bytes, headers: Mapping[str, str], timeout: float) -> httpx.Response: + self.posts.append((url, content, headers, timeout)) + outcome = self.outcomes.pop(0) + if isinstance(outcome, Exception): + raise outcome + return outcome + + +def _batch() -> Batch: + state1, call = chain( + AuditEvent(T0, "alice", None, "eu", "POST", "/key/generate", "allowed", 200, None, None), + Product("t"), + ChainState(), + ) + _, login = chain( + AuditEvent( + T0, "admin", None, "control-plane", "POST", "/auth/login", "denied", None, None, "no", "authentication" + ), + Product("t"), + state1, + ) + return (call, login) + + +@pytest.mark.asyncio +async def test_a_batch_is_one_post_of_newline_separated_hec_events_carrying_the_ocsf_objects() -> None: + poster = FakePoster([httpx.Response(200, json={"text": "Success", "code": 0})]) + batch = _batch() + assert await SplunkHecSink(SETTINGS, poster, "splunk_hec").deliver(batch) == Delivered(2) + ((url, content, headers, timeout),) = poster.posts + assert url == "https://hec.example.test:8088/services/collector/event" and timeout == 4.0 + assert headers == {"Authorization": "Splunk hec-token", "Content-Type": "application/json"} + events = [json.loads(line) for line in content.decode().split("\n")] + assert events == [ + { + "time": T0.timestamp(), + "source": "cp", + "sourcetype": "ocsf:api_activity", + "index": "audit", + "event": batch[0].body, + }, + { + "time": T0.timestamp(), + "source": "cp", + "sourcetype": "ocsf:authentication", + "index": "audit", + "event": batch[1].body, + }, + ] + + +@pytest.mark.asyncio +async def test_without_an_index_the_event_leaves_it_to_the_token_default() -> None: + poster = FakePoster([httpx.Response(200)]) + await SplunkHecSink(replace(SETTINGS, index=None), poster).deliver(_batch()[:1]) + ((_, content, _, _),) = poster.posts + assert "index" not in json.loads(content) + + +@pytest.mark.asyncio +async def test_hec_refusals_and_network_errors_are_retried_without_leaking_the_token() -> None: + poster = FakePoster( + [ + httpx.Response(403, json={"text": "Invalid token", "code": 4}), + httpx.Response(200, json={"text": "No data", "code": 5}), + httpx.Response(200), + httpx.Response(503), + httpx.ConnectError("boom"), + ] + ) + sink = SplunkHecSink(SETTINGS, poster) + batch = _batch() + assert await sink.deliver(batch) == Undelivered("HTTP 403 from splunk hec: Invalid token (code 4)") + assert await sink.deliver(batch) == Undelivered("HTTP 200 from splunk hec: No data (code 5)") + assert await sink.deliver(batch) == Undelivered("HTTP 200 from splunk hec: no HEC reply body (code -1)") + assert await sink.deliver(batch) == Undelivered("HTTP 503 from splunk hec: no HEC reply body (code -1)") + assert await sink.deliver(batch) == Undelivered("ConnectError posting to splunk hec") diff --git a/tests/unit/proxy/control_plane/audit/test_transport.py b/tests/unit/proxy/control_plane/audit/test_transport.py new file mode 100644 index 00000000000..848823e1349 --- /dev/null +++ b/tests/unit/proxy/control_plane/audit/test_transport.py @@ -0,0 +1,40 @@ +from dataclasses import dataclass, field +from typing import cast + +import httpx +import pytest + +from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler +from litellm.proxy.control_plane.audit.s3 import HttpxObjectStore +from litellm.proxy.control_plane.audit.transport import HttpxPoster + + +@dataclass +class _RedirectingHandler: # mutable-ok: counts the hops a following pool would take + hops: list[str] = field(default_factory=list) + + @property + def client(self) -> httpx.AsyncClient: + def route(request: httpx.Request) -> httpx.Response: + self.hops.append(str(request.url)) + if request.url.host == "sink.test": + return httpx.Response(302, headers={"Location": "http://127.0.0.1:9/collect"}) + return httpx.Response(200, json={"leaked": True}) + + return httpx.AsyncClient(transport=httpx.MockTransport(route), follow_redirects=True) + + +@pytest.mark.asyncio +async def test_the_poster_never_follows_a_redirect_even_on_a_following_pool() -> None: + handler = _RedirectingHandler() + poster = HttpxPoster(cast(AsyncHTTPHandler, handler)) # cast-ok: only the client property is used + reply = await poster.post("https://sink.test/collect", b"[]", {"Authorization": "Splunk t"}, 1.0) + assert reply.status_code == 302 and handler.hops == ["https://sink.test/collect"] + + +@pytest.mark.asyncio +async def test_the_object_store_never_follows_a_redirect_even_on_a_following_pool() -> None: + handler = _RedirectingHandler() + store = HttpxObjectStore(cast(AsyncHTTPHandler, handler)) # cast-ok: only the client property is used + reply = await store.put("https://sink.test/k", b"{}", {"Authorization": "AWS4 x"}, 1.0) + assert reply.status_code == 302 and handler.hops == ["https://sink.test/k"] diff --git a/tests/unit/proxy/control_plane/test_settings.py b/tests/unit/proxy/control_plane/test_settings.py index 393041a6dd8..ccffb9446c2 100644 --- a/tests/unit/proxy/control_plane/test_settings.py +++ b/tests/unit/proxy/control_plane/test_settings.py @@ -1,7 +1,10 @@ import pytest from pydantic import ValidationError -from litellm.proxy.control_plane.settings import ControlPlaneSettings, S3SinkSettings +from litellm.proxy.control_plane.settings import ( + ControlPlaneSettings, + S3SinkSettings, +) from litellm.types.proxy.control_plane.auth import PASSWORD_MAX_LENGTH, USERNAME_MAX_LENGTH @@ -35,9 +38,9 @@ def test_the_s3_sink_refuses_to_start_without_a_bucket_and_reads_its_block_with_ def test_a_plain_http_s3_endpoint_is_refused_unless_explicitly_allowed() -> None: base = {"LITELLM_MASTER_KEY": "sk-m", "audit_s3_bucket": "audit", "audit_s3_endpoint_url": "http://minio:9000"} - with pytest.raises(ValidationError, match="not https"): + with pytest.raises(ValidationError, match="CONTROL_PLANE_AUDIT_S3_ENDPOINT_URL is not https"): ControlPlaneSettings.model_validate(base) - with pytest.raises(ValidationError, match="not https"): + with pytest.raises(ValidationError, match="CONTROL_PLANE_AUDIT_S3_ALLOW_INSECURE_HTTP"): ControlPlaneSettings.model_validate({**base, "audit_s3_endpoint_url": "HTTP://minio:9000"}) allowed = ControlPlaneSettings.model_validate({**base, "audit_s3_allow_insecure_http": True}).audit_s3 assert allowed is not None and allowed.endpoint_url == "http://minio:9000" @@ -52,3 +55,61 @@ def test_credentials_at_the_bound_are_accepted() -> None: } ) assert settings.ui_username is not None and len(settings.ui_username) == USERNAME_MAX_LENGTH + + +@pytest.mark.parametrize( + ("field", "value", "opt_out"), + [ + ("audit_splunk_url", "http://hec.internal:8088", "audit_splunk_allow_insecure_http"), + ("audit_sentinel_endpoint", "HTTP://dce.test", None), + ("audit_sentinel_authority_host", "http://login.test", None), + ], +) +def test_sink_urls_that_carry_a_credential_must_be_https(field: str, value: str, opt_out: str | None) -> None: + with pytest.raises(ValidationError, match=f"CONTROL_PLANE_{field.upper()} is not https"): + ControlPlaneSettings.model_validate({"LITELLM_MASTER_KEY": "sk-m", field: value}) + if opt_out is not None: + allowed = ControlPlaneSettings.model_validate({"LITELLM_MASTER_KEY": "sk-m", field: value, opt_out: True}) + assert getattr(allowed, field) == value + + +def test_the_splunk_and_sentinel_sinks_refuse_to_start_without_their_settings_and_read_them_when_set() -> None: + with pytest.raises(ValidationError, match="CONTROL_PLANE_AUDIT_SPLUNK_URL, CONTROL_PLANE_AUDIT_SPLUNK_TOKEN"): + ControlPlaneSettings.model_validate({"LITELLM_MASTER_KEY": "sk-m", "audit_sinks": ["splunk_hec"]}) + with pytest.raises(ValidationError, match="CONTROL_PLANE_AUDIT_SENTINEL_CLIENT_SECRET"): + ControlPlaneSettings.model_validate( + { + "LITELLM_MASTER_KEY": "sk-m", + "audit_sinks": ["sentinel"], + "audit_sentinel_endpoint": "https://dce.test", + "audit_sentinel_dcr_immutable_id": "dcr", + "audit_sentinel_tenant_id": "t", + "audit_sentinel_client_id": "c", + } + ) + settings = ControlPlaneSettings.model_validate( + { + "LITELLM_MASTER_KEY": "sk-m", + "audit_sinks": ["splunk_hec", "sentinel"], + "audit_splunk_url": "https://hec.test:8088", + "audit_splunk_token": "tok", + "audit_sentinel_endpoint": "https://dce.test", + "audit_sentinel_dcr_immutable_id": "dcr", + "audit_sentinel_tenant_id": "t", + "audit_sentinel_client_id": "c", + "audit_sentinel_client_secret": "s", + "audit_sentinel_authority_host": "https://login.microsoftonline.us/", + } + ) + splunk = settings.audit_splunk + sentinel = settings.audit_sentinel + assert splunk is not None and (splunk.url, splunk.token.get_secret_value(), splunk.index) == ( + "https://hec.test:8088", + "tok", + None, + ) + assert sentinel is not None and sentinel.authority_host == "https://login.microsoftonline.us" + assert ( + sentinel.stream_name == "Custom-LiteLLMControlPlaneAudit_CL" + and sentinel.client_secret.get_secret_value() == "s" + )