mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-14 09:03:16 +00:00
feat(control_plane): splunk hec and microsoft sentinel audit sinks
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yassin
parent
4bba786891
commit
e736add73e
12 changed files with 1328 additions and 14 deletions
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
319
litellm/proxy/control_plane/audit/sentinel.py
Normal file
319
litellm/proxy/control_plane/audit/sentinel.py
Normal file
|
|
@ -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
|
||||
102
litellm/proxy/control_plane/audit/splunk.py
Normal file
102
litellm/proxy/control_plane/audit/splunk.py
Normal file
|
|
@ -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:<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))
|
||||
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
|
||||
26
litellm/proxy/control_plane/audit/transport.py
Normal file
26
litellm/proxy/control_plane/audit/transport.py
Normal file
|
|
@ -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
|
||||
)
|
||||
|
|
@ -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()
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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).
|
||||
|
|
|
|||
301
tests/unit/proxy/control_plane/audit/test_sentinel.py
Normal file
301
tests/unit/proxy/control_plane/audit/test_sentinel.py
Normal file
|
|
@ -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
|
||||
215
tests/unit/proxy/control_plane/audit/test_splunk.py
Normal file
215
tests/unit/proxy/control_plane/audit/test_splunk.py
Normal file
|
|
@ -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)
|
||||
40
tests/unit/proxy/control_plane/audit/test_transport.py
Normal file
40
tests/unit/proxy/control_plane/audit/test_transport.py
Normal file
|
|
@ -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"]
|
||||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue