Merge bda035b989 into c22d84d999

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)

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

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:<class>`` 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

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
+        )

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()
 

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

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}]]

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")

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"]

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"
+    )