From e736add73e5fbb515c9ca6c9ebebd1654584dcc7 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/pipeline.py | 9 +- litellm/proxy/control_plane/audit/s3.py | 3 +- litellm/proxy/control_plane/audit/sentinel.py | 319 ++++++++++++++++++ litellm/proxy/control_plane/audit/splunk.py | 102 ++++++ .../proxy/control_plane/audit/transport.py | 26 ++ litellm/proxy/control_plane/container.py | 24 ++ litellm/proxy/control_plane/settings.py | 161 ++++++++- .../code_coverage_tests/recursive_detector.py | 2 + .../control_plane/audit/test_sentinel.py | 301 +++++++++++++++++ .../proxy/control_plane/audit/test_splunk.py | 215 ++++++++++++ .../control_plane/audit/test_transport.py | 40 +++ .../unit/proxy/control_plane/test_settings.py | 140 +++++++- 12 files changed, 1328 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/pipeline.py b/litellm/proxy/control_plane/audit/pipeline.py index 9e2d1d91f83..5a77edf1c36 100644 --- a/litellm/proxy/control_plane/audit/pipeline.py +++ b/litellm/proxy/control_plane/audit/pipeline.py @@ -25,7 +25,12 @@ class Undelivered: settled: int = 0 -DeliveryResult: TypeAlias = Delivered | Undelivered +@dataclass(frozen=True, slots=True) +class Rejected: + reason: str + + +DeliveryResult: TypeAlias = Delivered | Undelivered | Rejected class AuditSink(Protocol): @@ -138,6 +143,8 @@ class SinkWorker: if not current_batch: break await self.sleep(self.backoff.delay(self.failures - 1)) + if isinstance(result, Rejected): + verbose_proxy_logger.error("control_plane.audit sink %s rejected batch (%s)", self.sink.name, result.reason) self.failures = 0 self.inflight = None diff --git a/litellm/proxy/control_plane/audit/s3.py b/litellm/proxy/control_plane/audit/s3.py index 5900ad4efb3..7a7599910d8 100644 --- a/litellm/proxy/control_plane/audit/s3.py +++ b/litellm/proxy/control_plane/audit/s3.py @@ -40,7 +40,8 @@ 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 diff --git a/litellm/proxy/control_plane/audit/sentinel.py b/litellm/proxy/control_plane/audit/sentinel.py new file mode 100644 index 00000000000..e4e74cb311f --- /dev/null +++ b/litellm/proxy/control_plane/audit/sentinel.py @@ -0,0 +1,319 @@ +import json +import time +from collections.abc import Callable, Iterable, Mapping, Sequence +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, JsonValue, TypeAdapter, ValidationError + +from litellm._logging import verbose_proxy_logger +from litellm.proxy.control_plane.audit.ocsf import AuditRecord, section +from litellm.proxy.control_plane.audit.pipeline import Batch, Delivered, DeliveryResult, Rejected, 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 +MAX_FIELD_BYTES: Final = 64 * 1024 +TRUNCATION_MARKER: Final = "...[truncated]" +_COMPACT_PATH_BYTES: Final = MAX_FIELD_BYTES // 2 +_COMPACT_ACTOR_VALUE_BYTES: Final = MAX_FIELD_BYTES // 16 +FIXED_WIDTH_COLUMNS: Final = frozenset( + { + "TimeGenerated", + "ClassUid", + "ActivityId", + "StatusId", + "SeverityId", + "ResponseCode", + "Sequence", + "Hash", + "PrevHash", + "TraceId", + } +) +REFRESH_MARGIN_SECONDS: Final = 60.0 +Row: TypeAlias = Mapping[str, object] +_JSON_MAPPING: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object]) + + +@dataclass(frozen=True, slots=True) +class TokenRefused: + status: int + + +class TokenSource(Protocol): + async def __call__(self) -> str | TokenRefused: ... + + +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 | TokenRefused: + 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": self.settings.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: + return 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, ensure_ascii=False).encode() + + +def _truncate_text(value: str, limit: int = MAX_FIELD_BYTES) -> str: + encoded: Final = value.encode("utf-8") + marker: Final = TRUNCATION_MARKER.encode() + if len(encoded) <= limit: + return value + prefix: Final = encoded[: limit - len(marker)].decode("utf-8", errors="ignore") + return prefix + TRUNCATION_MARKER + + +def _compact_actor_text(value: str) -> str: + return _truncate_text(value, _COMPACT_ACTOR_VALUE_BYTES) + + +_JSON_VALUE: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue) + + +def _compact_actor_value(value: JsonValue) -> JsonValue: + if isinstance(value, str): + return _compact_actor_text(value) + if isinstance(value, dict): + actor_fields: Final[Mapping[str, JsonValue]] = value + return {key: _compact_actor_value(item) for key, item in actor_fields.items()} + if isinstance(value, list): + actor_items: Final[Sequence[JsonValue]] = value + return [_compact_actor_value(item) for item in actor_items] + return value + + +def _compact_actor(value: object) -> object: + try: + actor: Final = _JSON_MAPPING.validate_python(value) + compact: Final[JsonValue] = _JSON_VALUE.validate_python(actor) + except ValidationError: + return {} + if not isinstance(compact, dict): + return {} + return _compact_actor_value(compact) + + +def _parse_json_mapping(value: str) -> Mapping[str, object]: + try: + return _JSON_MAPPING.validate_json(value) + except ValidationError: + return {} + + +def _compact_ocsf(value: str, row: Row) -> str: + body: Final = _parse_json_mapping(value) + litellm: Final = section(body, "unmapped", "litellm") + request: Final = section(body, "http_request", "url") + path_value: Final = request.get("path", row.get("UrlPath", "")) + path: Final = path_value if isinstance(path_value, str) else "" + detail: Final = body.get("status_detail") + status_detail: Final = _truncate_text(detail, _COMPACT_ACTOR_VALUE_BYTES) if isinstance(detail, str) else detail + actor: Final = _compact_actor(body.get("actor")) + compact: Final = { + "activity_id": body.get("activity_id", row.get("ActivityId")), + "activity_name": body.get("activity_name", row.get("ActivityName")), + "actor": actor or {"user": {"uid": row.get("ActorUid")}}, + "http_request": {"url": {"path": _truncate_text(path, _COMPACT_PATH_BYTES)}}, + "status": body.get("status", row.get("Decision")), + "status_code": body.get("status_code", row.get("ResponseCode")), + "status_detail": status_detail, + "status_id": body.get("status_id", row.get("StatusId")), + "unmapped": { + "litellm": { + "hash": litellm.get("hash", row.get("Hash")), + "prev_hash": litellm.get("prev_hash", row.get("PrevHash")), + "sequence": litellm.get("sequence", row.get("Sequence")), + } + }, + } + serialized: Final = _encoded(compact) + if len(serialized) <= MAX_FIELD_BYTES: + return serialized.decode("utf-8") + metadata: Final = { + key: compact[key] for key in ("activity_id", "activity_name", "status", "status_code", "status_id", "unmapped") + } + return _encoded({**metadata, "Truncated": True}).decode("utf-8") + + +def _fitted(row: Row, limit: int) -> bytes: + """Bound every string field and keep one row from blocking a batch larger than the ingestion limit.""" + oversized: Final = any( + isinstance(value, str) and len(value.encode("utf-8")) > MAX_FIELD_BYTES for value in row.values() + ) + bounded: Final = { + key: _compact_ocsf(value, row) + if key == "Ocsf" and isinstance(value, str) and len(value.encode("utf-8")) > MAX_FIELD_BYTES + else _truncate_text(value) + if isinstance(value, str) + else value + for key, value in row.items() + } + fitted: Final = {**bounded, **({"Truncated": True} if oversized else {})} + whole: Final = _encoded(fitted) + if len(whole) + 2 <= limit: + return whole + trimmed: Final = _encoded({**fitted, "Detail": None, "Truncated": True}) + if len(trimmed) + 2 <= limit: + return trimmed + without_ocsf: Final = _encoded({**fitted, "Detail": None, "Ocsf": None, "Truncated": True}) + if len(without_ocsf) + 2 <= limit: + return without_ocsf + return _encoded({**{key: value for key, value in fitted.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.""" + return tuple(payload for payload, _ in _chunks_with_counts(rows, limit)) + + +def _chunks_with_counts(rows: Iterable[Row], limit: int) -> tuple[tuple[bytes, int], ...]: + 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]) + rows_by_chunk: Final = tuple(tuple(item for _, item in members) for _, members in groups) + return tuple((b"[" + b",".join(group) + b"]", len(group)) for group in rows_by_chunk) + + +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" + _chunk_limit: int = MAX_PAYLOAD_BYTES + + @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 (httpx.HTTPError, ValidationError) as error: + return Undelivered(f"{type(error).__name__} getting the entra token") + if isinstance(bearer, TokenRefused): + return Undelivered(f"entra token refused: HTTP {bearer.status} from the token endpoint") + headers: Final = {"Authorization": f"Bearer {bearer}", "Content-Type": "application/json"} + uploads: Final = _chunks_with_counts( + (sentinel_row(record) for record in batch), + self._chunk_limit, + ) + outcomes: Final = tuple([await self._post_chunk(payload, count, headers) for payload, count in uploads]) + retryable: Final = next( + ( + reason + for status_code, reason, _ in outcomes + if status_code is None or status_code not in (204, 400, 413, 422) + ), + None, + ) + if retryable is not None: + return Undelivered(retryable) + for status_code, _, _ in outcomes: + if status_code in (400, 413, 422): + verbose_proxy_logger.error("control_plane.audit sentinel rejected chunk with HTTP %d", status_code) + if outcomes and all(status_code in (400, 413, 422) for status_code, _, _ in outcomes): + return Rejected("all chunks were permanently rejected") + return Delivered(sum(count for status_code, _, count in outcomes if status_code == 204)) + + async def _post_chunk( + self, payload: bytes, count: int, headers: Mapping[str, str] + ) -> tuple[int | None, str | None, int]: + try: + response: Final = await self.post.post(self.url, payload, headers, self.settings.timeout_seconds) + except httpx.HTTPError as error: + return None, f"{type(error).__name__} posting to the logs ingestion api", count + status_code: Final = response.status_code + if status_code == 204 or status_code in (400, 413, 422): + return status_code, None, count + return status_code, f"HTTP {status_code} from the logs ingestion api", count + + 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..45557d63b7b --- /dev/null +++ b/litellm/proxy/control_plane/audit/splunk.py @@ -0,0 +1,102 @@ +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._logging import verbose_proxy_logger +from litellm.proxy.control_plane.audit.ocsf import AuditRecord +from litellm.proxy.control_plane.audit.pipeline import Batch, Delivered, DeliveryResult, Rejected, 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)) + if response.status_code == 413 and len(batch) > 1: + middle: Final = len(batch) // 2 + first: Final = await self.deliver(batch[:middle]) + if isinstance(first, Undelivered): + return first + second: Final = await self.deliver(batch[middle:]) + if isinstance(second, Undelivered): + return Undelivered(second.reason, settled=middle + second.settled) + halves: Final = (first, second) + delivered_count: Final = sum(result.count for result in halves if isinstance(result, Delivered)) + return ( + Delivered(delivered_count) + if delivered_count > 0 + else Rejected("all split batches were rejected by splunk hec") + ) + if response.status_code in (400, 413, 422): + verbose_proxy_logger.error( + "control_plane.audit splunk hec rejected batch with HTTP %d", response.status_code + ) + return Rejected(f"HTTP {response.status_code} from splunk hec") + 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..c4e0c28d093 --- /dev/null +++ b/litellm/proxy/control_plane/audit/transport.py @@ -0,0 +1,26 @@ +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 9cac4e2335e..026a4479a41 100644 --- a/litellm/proxy/control_plane/container.py +++ b/litellm/proxy/control_plane/container.py @@ -24,7 +24,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 ( @@ -148,12 +151,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 0b157d8305f..6f421e06699 100644 --- a/litellm/proxy/control_plane/settings.py +++ b/litellm/proxy/control_plane/settings.py @@ -16,8 +16,30 @@ from litellm.proxy.control_plane.auth.rbac import ADMIN_ROLE 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 @@ -32,6 +54,42 @@ 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 + token_scope: str + timeout_seconds: float + + +def _sentinel_token_scope(authority_host: str, override: str | None) -> str: + if override is not None: + return override + match urlsplit(authority_host).hostname: + case "login.microsoftonline.us": + return "https://monitor.azure.us/.default" + case "login.chinacloudapi.cn": + return "https://monitor.azure.cn/.default" + case _: + return "https://monitor.azure.com/.default" + + @dataclass(frozen=True, slots=True) class OidcSettings: """External IdP used to verify caller JWTs and, later, to exchange them for dataplane tokens.""" @@ -145,6 +203,21 @@ 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_token_scope: str | None = None + audit_sentinel_timeout_seconds: float = 30.0 @field_validator("audit_s3_bucket") @classmethod @@ -153,20 +226,90 @@ class ControlPlaneSettings(BaseSettings): raise ValueError("CONTROL_PLANE_AUDIT_S3_BUCKET must not be empty or whitespace.") return value + @field_validator( + "audit_splunk_url", + "audit_splunk_token", + "audit_splunk_index", + "audit_splunk_source", + "audit_sentinel_endpoint", + "audit_sentinel_dcr_immutable_id", + "audit_sentinel_stream_name", + "audit_sentinel_tenant_id", + "audit_sentinel_client_id", + "audit_sentinel_client_secret", + "audit_sentinel_token_scope", + ) + @classmethod + def _required_sink_strings_are_not_blank(cls, value: str | SecretStr | None) -> str | SecretStr | None: + text: Final = value.get_secret_value() if isinstance(value, SecretStr) else value + if isinstance(text, str) and not text.strip(): + raise ValueError("Required audit sink settings must not be empty or whitespace.") + return value + @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 + authority_host: Final = self.audit_sentinel_authority_host.rstrip("/") + 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=authority_host, + token_scope=_sentinel_token_scope(authority_host, self.audit_sentinel_token_scope), + timeout_seconds=self.audit_sentinel_timeout_seconds, ) @property diff --git a/tests/code_coverage_tests/recursive_detector.py b/tests/code_coverage_tests/recursive_detector.py index a84a40c67c1..800a2ad52b2 100644 --- a/tests/code_coverage_tests/recursive_detector.py +++ b/tests/code_coverage_tests/recursive_detector.py @@ -3,6 +3,7 @@ import os IGNORE_FUNCTIONS = [ "_json_cost", # bounded at depth 32 and consumed under byte/node limits. + "_compact_actor_value", "detached", # router_config: copies only the plain containers of the router's own constructor arguments, once per build or changeset "_format_type", "remove_additional_properties", @@ -81,6 +82,7 @@ IGNORE_FUNCTIONS = [ "with_id_fields_replaced", # harness/opencode: bounded by the nesting depth of the JsonValue that validate_json parsed from `opencode export`, which refuses nesting past ~200 levels. "completion_cost", # max depth 1: recursion only fires for mixed-tier Responses WS logging objects, and each split part carries a single service_tier so _split_responses_ws_logging_object_by_service_tier returns None. "_string_leaves", # bounded by the nesting depth of a safe_json_structure output (a finite JSON tree, no cycles possible). + "_compact_actor_value", # OCSF actors have schema-bounded depth. "_replace_string_leaves", # bounded by the nesting depth of a safe_json_structure output (a finite JSON tree, no cycles possible). "_redacted_client_field", # bounded by the nesting depth of a JSON-decoded error body (a finite JSON tree, no cycles possible). "_sort_processed_sets", # bounded by the nesting depth of the log-record extra it walks (a finite JSON tree, no cycles possible). 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..e0df0153a20 --- /dev/null +++ b/tests/unit/proxy/control_plane/audit/test_sentinel.py @@ -0,0 +1,301 @@ +import json +from collections.abc import Mapping +from dataclasses import dataclass, field, replace +from datetime import UTC, datetime +from itertools import accumulate, chain as chain_iterables +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, Rejected, Undelivered +from litellm.proxy.control_plane.audit.sentinel import ( + ClientCredentials, + SentinelSink, + TokenRefused, + 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", + token_scope="https://monitor.azure.com/.default", + 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: Final = 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: + events: Final = tuple( + AuditEvent( + T0, "alice", "a@x.test", "eu", "POST", f"/key/{index}", "denied", 403, None, "no", "api_activity", "ab" * 16 + ) + for index in range(count) + ) + if not events: + return () + first: Final = chain(events[0], Product("t"), ChainState()) + state_and_records: Final = tuple( + accumulate(events[1:], lambda previous, event: chain(event, Product("t"), previous[0]), initial=first) + ) + return tuple(record for _, record in state_and_records) + + +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: Final = FakePoster([httpx.Response(204)]) + batch: Final = _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: Final = 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: Final = [{"Ocsf": "x" * 300} for _ in range(10)] + parts: Final = 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: Final = [{"Hash": "h1", "Detail": "d", "Ocsf": "x" * 2_000}, {"Hash": "h2", "Detail": "d", "Ocsf": "y"}] + parts: Final = chunks(rows, limit=1_000) + assert all(len(part) <= 1_000 for part in parts) + decoded_parts: Final = tuple(json.loads(part) for part in parts) + sent: Final = tuple(chain_iterables.from_iterable(decoded_parts)) + 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: Final = FakePoster( + [ + httpx.Response(200, json={"access_token": "t1", "expires_in": 600}), + httpx.Response(200, json={"access_token": "t2", "expires_in": 600}), + ] + ) + clock: Final = Clock() + credentials: Final = 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_client_credentials_returns_token_refusal_as_a_value() -> None: + poster: Final = FakePoster([httpx.Response(401, json={"error": "invalid_client"})]) + + assert await ClientCredentials(SETTINGS, poster)() == TokenRefused(401) + + +@pytest.mark.asyncio +async def test_client_credentials_uses_an_explicit_token_scope() -> None: + scope: Final = "https://monitor.custom.test/.default" + settings: Final = replace(SETTINGS, token_scope=scope) + poster: Final = FakePoster([httpx.Response(200, json={"access_token": "token"})]) + credentials: Final = ClientCredentials(settings, poster) + + assert await credentials() == "token" + ((_, form, _, _),) = poster.posts + assert parse_qs(form.decode())["scope"] == [scope] + + +@pytest.mark.asyncio +async def test_a_refused_token_or_ingestion_error_is_retried_and_names_no_secret() -> None: + poster: Final = FakePoster( + [httpx.Response(401, json={"error": "invalid_client"}), httpx.Response(429), httpx.ConnectError("x")] + ) + refused: Final = await SentinelSink(SETTINGS, poster, ClientCredentials(SETTINGS, poster)).deliver(_batch(1)) + assert refused == Undelivered("entra token refused: HTTP 401 from the token endpoint") + sink: Final = 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") + + +@pytest.mark.parametrize("status_code", [401, 403]) +@pytest.mark.asyncio +async def test_access_refusals_from_ingestion_are_retryable(status_code: int) -> None: + poster: Final = FakePoster([httpx.Response(status_code)]) + assert await SentinelSink(SETTINGS, poster, _token).deliver(_batch(1)) == Undelivered( + f"HTTP {status_code} from the logs ingestion api" + ) + + +@pytest.mark.asyncio +async def test_permanent_413_rejection_does_not_block_the_next_batch() -> None: + poster: Final = FakePoster([httpx.Response(413), httpx.Response(204)]) + sink: Final = SentinelSink(SETTINGS, poster, _token, _chunk_limit=1) + assert await sink.deliver(_batch(1)) == Rejected("all chunks were permanently rejected") + assert await sink.deliver(_batch(1)) == Delivered(1) + assert len(poster.posts) == 2 + + +@pytest.mark.asyncio +async def test_a_permanently_rejected_chunk_does_not_stop_later_chunks( + caplog: pytest.LogCaptureFixture, +) -> None: + poster: Final = FakePoster([httpx.Response(204), httpx.Response(400), httpx.Response(204)]) + sink: Final = SentinelSink(SETTINGS, poster, _token, _chunk_limit=1) + assert await sink.deliver(_batch(3)) == Delivered(2) + assert len(poster.posts) == 3 + assert "rejected chunk with HTTP 400" in caplog.text + + +@pytest.mark.asyncio +async def test_a_retryable_failure_in_a_chunk_makes_the_batch_undelivered() -> None: + poster: Final = FakePoster([httpx.Response(204), httpx.Response(403)]) + sink: Final = SentinelSink(SETTINGS, poster, _token, _chunk_limit=1) + assert await sink.deliver(_batch(2)) == Undelivered("HTTP 403 from the logs ingestion api") + assert len(poster.posts) == 2 + + +def test_a_row_whose_flat_column_alone_exceeds_the_limit_keeps_only_its_fixed_width_columns() -> None: + rows: Final = [{"Hash": "h1", "Sequence": 7, "UrlPath": "/" + "k" * 2_000, "Detail": "d", "Ocsf": "x"}] + parts: Final = 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}]] + + +def test_oversized_ocsf_and_other_fields_are_bounded_in_utf8_bytes() -> None: + path: Final = "/" + "é" * 35_000 + detail: Final = "é" * 35_000 + event: Final = AuditEvent(T0, "alice", "a@x.test", "eu", "POST", path, "denied", 403, None, detail) + record: Final = chain(event, Product("t"), ChainState())[1] + source: Final = sentinel_row(record) + assert len(source["Ocsf"].encode("utf-8")) > 64 * 1024 + + payload: Final = chunks((source,))[0] + (row,) = json.loads(payload) + compact: Final = json.loads(row["Ocsf"]) + assert row["Truncated"] is True and len(payload) <= 1_000_000 + assert len(row["Ocsf"].encode("utf-8")) <= 64 * 1024 + assert len(row["UrlPath"].encode("utf-8")) <= 64 * 1024 and row["UrlPath"].endswith("...[truncated]") + assert len(row["Detail"].encode("utf-8")) <= 64 * 1024 and row["Detail"].endswith("...[truncated]") + assert compact["actor"] == record.body["actor"] + assert (compact["activity_name"], compact["status"]) == ("Create", "Failure") + assert compact["unmapped"]["litellm"] == {"hash": record.hash, "prev_hash": None, "sequence": 1} + assert compact["http_request"]["url"]["path"].endswith("...[truncated]") + + +def test_quote_heavy_paths_keep_the_final_ocsf_string_under_the_field_limit() -> None: + path: Final = "/" + '"' * (32 * 1024 + 100) + record: Final = chain( + AuditEvent(T0, "alice", "a@x.test", "eu", "POST", path, "denied", 403, None, "no"), + Product("t"), + ChainState(), + )[1] + source: Final = sentinel_row(record) + assert len(source["Ocsf"].encode("utf-8")) > 64 * 1024 + + (row,) = json.loads(chunks((source,))[0]) + ocsf: Final = row["Ocsf"] + assert isinstance(ocsf, str) + assert len(ocsf.encode("utf-8")) <= 64 * 1024 + compact: Final = json.loads(ocsf) + assert compact["Truncated"] is True + assert compact["activity_id"] == record.body["activity_id"] + assert compact["activity_name"] == record.body["activity_name"] + assert compact["unmapped"]["litellm"] == {"hash": record.hash, "prev_hash": None, "sequence": 1} + + +def test_oversized_actor_values_are_bounded_in_compacted_ocsf() -> None: + actor: Final = "a" * 70_000 + event: Final = AuditEvent(T0, actor, None, "eu", "POST", "/", "denied", 403, None, None) + record: Final = chain(event, Product("t"), ChainState())[1] + source: Final = sentinel_row(record) + payload: Final = chunks((source,))[0] + (row,) = json.loads(payload) + compact: Final = json.loads(row["Ocsf"]) + compact_actor: Final = compact["actor"]["user"]["uid"] + + assert compact_actor != actor + assert compact_actor.endswith("...[truncated]") + assert len(compact_actor.encode("utf-8")) <= 4 * 1024 + + +def test_nested_actor_values_are_compacted_without_losing_their_shape() -> None: + record: Final = chain( + AuditEvent(T0, "alice", None, "eu", "POST", "/", "denied", 403, None, None), + Product("t"), + ChainState(), + )[1] + nested_record: Final = replace( + record, + body={ + **record.body, + "actor": {"user": {"uid": "a" * 70_000, "groups": ["b" * 70_000]}}, + }, + ) + source: Final = sentinel_row(nested_record) + (row,) = json.loads(chunks((source,))[0]) + compact_actor: Final = json.loads(row["Ocsf"])["actor"] + + assert compact_actor["user"]["uid"].endswith("...[truncated]") + assert compact_actor["user"]["groups"][0].endswith("...[truncated]") + assert len(compact_actor["user"]["uid"].encode("utf-8")) <= 4 * 1024 + assert len(compact_actor["user"]["groups"][0].encode("utf-8")) <= 4 * 1024 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..65339f68ee3 --- /dev/null +++ b/tests/unit/proxy/control_plane/audit/test_splunk.py @@ -0,0 +1,215 @@ +import json +from collections.abc import Mapping +from dataclasses import dataclass, field, replace +from datetime import UTC, datetime +from itertools import accumulate, chain as chain_iterables +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 AuditRecord, ChainState, Product, chain +from litellm.proxy.control_plane.audit.pipeline import Batch, Delivered, Rejected, Undelivered, build_pipeline +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: Final = 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) + + +def _four_record_batch() -> Batch: + def append_record( + previous: tuple[ChainState, tuple[AuditRecord, ...]], event: AuditEvent + ) -> tuple[ChainState, tuple[AuditRecord, ...]]: + state, records = previous + next_state, record = chain(event, Product("t"), state) + return next_state, (*records, record) + + events: Final = tuple( + AuditEvent(T0, "alice", None, "eu", "GET", f"/key/{index}", "allowed", 200, None, None) + for index in range(4) + ) + results: Final = tuple(accumulate(events, append_record, initial=(ChainState(), ()))) + return results[-1][1] + + +@pytest.mark.asyncio +async def test_a_batch_is_one_post_of_newline_separated_hec_events_carrying_the_ocsf_objects() -> None: + poster: Final = FakePoster([httpx.Response(200, json={"text": "Success", "code": 0})]) + batch: Final = _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: Final = [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: Final = 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_permanent_refusals_are_rejected_and_transient_errors_are_retryable() -> None: + poster: Final = FakePoster( + [ + httpx.Response(401, json={"text": "Unauthorized", "code": 4}), + httpx.Response(403, json={"text": "Invalid token", "code": 4}), + httpx.Response(400), + httpx.Response(422), + httpx.Response(200, json={"text": "No data", "code": 5}), + httpx.Response(200), + httpx.Response(503), + httpx.ConnectError("boom"), + ] + ) + sink: Final = SplunkHecSink(SETTINGS, poster) + batch: Final = _batch() + assert await sink.deliver(batch) == Undelivered("HTTP 401 from splunk hec: Unauthorized (code 4)") + assert await sink.deliver(batch) == Undelivered("HTTP 403 from splunk hec: Invalid token (code 4)") + assert await sink.deliver(batch) == Rejected("HTTP 400 from splunk hec") + assert await sink.deliver(batch) == Rejected("HTTP 422 from splunk hec") + 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") + + +@pytest.mark.parametrize("status_code", [401, 403]) +@pytest.mark.asyncio +async def test_access_refusals_are_retryable(status_code: int) -> None: + poster: Final = FakePoster([httpx.Response(status_code)]) + result: Final = await SplunkHecSink(SETTINGS, poster).deliver(_batch()) + assert isinstance(result, Undelivered) + + +@pytest.mark.asyncio +async def test_permanent_413_rejection_does_not_block_the_next_batch() -> None: + poster: Final = FakePoster([httpx.Response(413), httpx.Response(200, json={"code": 0})]) + sink: Final = SplunkHecSink(SETTINGS, poster) + assert await sink.deliver(_batch()[:1]) == Rejected("HTTP 413 from splunk hec") + assert await sink.deliver(_batch()[:1]) == Delivered(1) + assert len(poster.posts) == 2 + + +@pytest.mark.asyncio +async def test_413_splits_a_batch_until_both_halves_are_accepted() -> None: + poster: Final = FakePoster( + [ + httpx.Response(413), + httpx.Response(200, json={"code": 0}), + httpx.Response(200, json={"code": 0}), + ] + ) + batch: Final = _batch() * 2 + result: Final = await SplunkHecSink(SETTINGS, poster).deliver(batch) + assert result == Delivered(4) + assert len(poster.posts) == 3 + + +@pytest.mark.asyncio +async def test_413_for_a_single_record_is_rejected() -> None: + poster: Final = FakePoster([httpx.Response(413)]) + assert await SplunkHecSink(SETTINGS, poster).deliver(_batch()[:1]) == Rejected("HTTP 413 from splunk hec") + assert len(poster.posts) == 1 + + +@pytest.mark.asyncio +async def test_undelivered_split_makes_the_splunk_batch_undelivered() -> None: + poster: Final = FakePoster([httpx.Response(413), httpx.Response(200, json={"code": 0}), httpx.Response(403)]) + assert isinstance(await SplunkHecSink(SETTINGS, poster).deliver(_batch()), Undelivered) + assert len(poster.posts) == 3 + + +@pytest.mark.asyncio +async def test_an_undelivered_first_split_does_not_post_the_second_half() -> None: + poster: Final = FakePoster([httpx.Response(413), httpx.Response(503), httpx.Response(200, json={"code": 0})]) + + result: Final = await SplunkHecSink(SETTINGS, poster).deliver(_batch()) + + assert isinstance(result, Undelivered) and result.settled == 0 + assert len(poster.posts) == 2 + + +@pytest.mark.asyncio +async def test_pipeline_retries_only_the_unaccepted_splunk_split() -> None: + poster: Final = FakePoster( + [ + httpx.Response(413), + httpx.Response(200, json={"text": "Success", "code": 0}), + httpx.Response(503), + httpx.Response(200, json={"text": "Success", "code": 0}), + ] + ) + batch: Final = _four_record_batch() + sink: Final = SplunkHecSink(SETTINGS, poster) + + async def no_sleep(_: float) -> None: + return None + + pipeline: Final = build_pipeline((sink,), queue_size=10, batch_size=4, strict=False, sleep=no_sleep) + for record in batch: + pipeline.publish(record) + + assert await pipeline.workers[0].flush_once() == 4 + posted_counts: Final = tuple( + len(content.decode().split("\n")) for _, content, _, _ in poster.posts + ) + assert posted_counts == (4, 2, 2, 2) + accepted_lines: Final = tuple( + chain_iterables.from_iterable(poster.posts[index][1].decode().split("\n") for index in (1, 3)) + ) + accepted_events: Final = tuple(json.loads(line)["event"] for line in accepted_lines) + assert accepted_events == tuple(record.body for record in batch) 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..c464a4e6719 --- /dev/null +++ b/tests/unit/proxy/control_plane/audit/test_transport.py @@ -0,0 +1,40 @@ +from dataclasses import dataclass, field +from typing import Final, 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: Final = _RedirectingHandler() + poster: Final = HttpxPoster(cast(AsyncHTTPHandler, handler)) # cast-ok: only the client property is used + reply: Final = 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: Final = _RedirectingHandler() + store: Final = HttpxObjectStore(cast(AsyncHTTPHandler, handler)) # cast-ok: only the client property is used + reply: Final = 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 3304b66d118..aad059cac69 100644 --- a/tests/unit/proxy/control_plane/test_settings.py +++ b/tests/unit/proxy/control_plane/test_settings.py @@ -4,7 +4,11 @@ import pytest from pydantic import ValidationError from litellm.proxy.control_plane.audit.event import CONTROL_PLANE_ID -from litellm.proxy.control_plane.settings import ControlPlaneSettings, S3SinkSettings, load_settings +from litellm.proxy.control_plane.settings import ( + ControlPlaneSettings, + S3SinkSettings, + load_settings, +) from litellm.types.proxy.control_plane.auth import PASSWORD_MAX_LENGTH, USERNAME_MAX_LENGTH MASTER_KEY: Final = "01234567890123456789012345678901" @@ -74,9 +78,9 @@ def test_a_plain_http_s3_endpoint_is_refused_unless_explicitly_allowed() -> None "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: Final = ControlPlaneSettings.model_validate({**base, "audit_s3_allow_insecure_http": True}).audit_s3 assert allowed is not None and allowed.endpoint_url == "http://minio:9000" @@ -153,3 +157,133 @@ def test_audit_queue_size_allows_two_slots() -> None: def test_strict_audit_requires_at_least_one_sink() -> None: with pytest.raises(ValidationError, match="requires at least one audit sink"): ControlPlaneSettings.model_validate({"LITELLM_MASTER_KEY": MASTER_KEY, "audit_strict": True, "audit_sinks": ()}) + + +@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": MASTER_KEY, field: value}) + if opt_out is not None: + allowed: Final = ControlPlaneSettings.model_validate( + {"LITELLM_MASTER_KEY": MASTER_KEY, 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": MASTER_KEY, "audit_sinks": ["splunk_hec"]}) + with pytest.raises(ValidationError, match="CONTROL_PLANE_AUDIT_SENTINEL_CLIENT_SECRET"): + ControlPlaneSettings.model_validate( + { + "LITELLM_MASTER_KEY": MASTER_KEY, + "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: Final = ControlPlaneSettings.model_validate( + { + "LITELLM_MASTER_KEY": MASTER_KEY, + "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: Final = settings.audit_splunk + sentinel: Final = 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" + and sentinel.token_scope == "https://monitor.azure.us/.default" + ) + china: Final = settings.model_copy( + update={"audit_sentinel_authority_host": "https://login.chinacloudapi.cn/"} + ).audit_sentinel + public: Final = settings.model_copy( + update={"audit_sentinel_authority_host": "https://login.microsoftonline.com/"} + ).audit_sentinel + override: Final = settings.model_copy( + update={"audit_sentinel_token_scope": "https://monitor.override.test/.default"} + ).audit_sentinel + assert china is not None and china.token_scope == "https://monitor.azure.cn/.default" + assert public is not None and public.token_scope == "https://monitor.azure.com/.default" + assert override is not None and override.token_scope == "https://monitor.override.test/.default" + + +@pytest.mark.parametrize( + ("field", "blank"), + [ + ("audit_splunk_url", ""), + ("audit_splunk_url", " \t "), + ("audit_splunk_token", ""), + ("audit_splunk_token", " \t "), + ("audit_splunk_index", ""), + ("audit_splunk_index", " \t "), + ("audit_splunk_source", ""), + ("audit_splunk_source", " \t "), + ("audit_sentinel_endpoint", ""), + ("audit_sentinel_endpoint", " \t "), + ("audit_sentinel_dcr_immutable_id", ""), + ("audit_sentinel_dcr_immutable_id", " \t "), + ("audit_sentinel_stream_name", ""), + ("audit_sentinel_stream_name", " "), + ("audit_sentinel_tenant_id", ""), + ("audit_sentinel_tenant_id", " \t "), + ("audit_sentinel_client_id", ""), + ("audit_sentinel_client_id", " \t "), + ("audit_sentinel_client_secret", ""), + ("audit_sentinel_client_secret", " \t "), + ("audit_sentinel_token_scope", ""), + ("audit_sentinel_token_scope", " \t "), + ], +) +def test_sink_strings_reject_empty_or_whitespace(field: str, blank: str) -> None: + settings: Final = { + "LITELLM_MASTER_KEY": MASTER_KEY, + "audit_sinks": ["splunk_hec", "sentinel"], + "audit_splunk_url": "https://hec.test:8088", + "audit_splunk_token": "token", + "audit_sentinel_endpoint": "https://dce.test", + "audit_sentinel_dcr_immutable_id": "dcr", + "audit_sentinel_tenant_id": "tenant", + "audit_sentinel_client_id": "client", + "audit_sentinel_client_secret": "secret", + } + with pytest.raises(ValidationError, match="must not be empty or whitespace"): + ControlPlaneSettings.model_validate({**settings, field: blank}) + + +@pytest.mark.parametrize("index", [None, "audit"]) +def test_splunk_index_is_optional_but_not_blank_when_set(index: str | None) -> None: + settings: Final = ControlPlaneSettings.model_validate( + { + "LITELLM_MASTER_KEY": MASTER_KEY, + "audit_splunk_url": "https://hec.test:8088", + "audit_splunk_token": "token", + "audit_splunk_index": index, + } + ) + splunk: Final = settings.audit_splunk + assert splunk is not None and splunk.index == index