From d5486f019c135e136487572b61856bac2a1e88e0 Mon Sep 17 00:00:00 2001 From: Ishaan Jaffer Date: Tue, 17 Feb 2026 16:25:57 -0800 Subject: [PATCH] add backend for checkers --- litellm/proxy/compliance_checks.py | 221 ++++++++++++++++++ .../compliance_endpoints.py | 79 +++++++ litellm/types/proxy/compliance_endpoints.py | 33 +++ 3 files changed, 333 insertions(+) create mode 100644 litellm/proxy/compliance_checks.py create mode 100644 litellm/proxy/management_endpoints/compliance_endpoints.py create mode 100644 litellm/types/proxy/compliance_endpoints.py diff --git a/litellm/proxy/compliance_checks.py b/litellm/proxy/compliance_checks.py new file mode 100644 index 00000000000..381b0f815d2 --- /dev/null +++ b/litellm/proxy/compliance_checks.py @@ -0,0 +1,221 @@ +""" +Compliance checker for EU AI Act and GDPR regulations. + +Provides guardrail-agnostic compliance validation based on guardrail modes +and execution results rather than specific guardrail names. +""" + +from typing import Dict, List + +from litellm.types.proxy.compliance_endpoints import ( + ComplianceCheckRequest, + ComplianceCheckResult, +) + + +class ComplianceChecker: + """ + Validates compliance with EU AI Act and GDPR regulations. + + Uses guardrail-agnostic checks based on: + - Whether any guardrails ran + - Guardrail execution mode (pre-call, post-call, etc.) + - Whether guardrails intervened/blocked content + - Completeness of audit records + """ + + def __init__(self, data: ComplianceCheckRequest): + self.data = data + self.guardrails = data.guardrail_information or [] + + def _get_guardrails_by_mode(self, mode: str) -> List[Dict]: + """ + Get all guardrails that ran in a specific mode. + + If a guardrail doesn't have a mode specified, it's treated as pre-call + (the most common case). + """ + result = [] + for g in self.guardrails: + g_mode = g.get("guardrail_mode") + # If no mode specified, default to pre_call + if g_mode is None and mode == "pre_call": + result.append(g) + elif g_mode == mode: + result.append(g) + return result + + def _has_guardrail_intervention(self, guardrails: List[Dict]) -> bool: + """Check if any guardrail intervened (blocked/masked content).""" + for g in guardrails: + status = g.get("guardrail_status", "") + if status in ["guardrail_intervened", "failed", "blocked"]: + return True + return False + + def _all_guardrails_passed(self, guardrails: List[Dict]) -> bool: + """Check if all guardrails passed (no issues detected).""" + if not guardrails: + return False + return all(g.get("guardrail_status") == "success" for g in guardrails) + + # ── EU AI Act Helper Methods ──────────────────────────────────────────── + + def _check_art_9_guardrails_applied(self) -> ComplianceCheckResult: + """Art. 9: Check if any guardrails were applied.""" + has_guardrails = len(self.guardrails) > 0 + return ComplianceCheckResult( + check_name="Guardrails applied", + article="Art. 9", + passed=has_guardrails, + detail=( + f"{len(self.guardrails)} guardrail(s) applied" + if has_guardrails + else "No guardrails applied" + ), + ) + + def _check_art_5_content_screened(self) -> ComplianceCheckResult: + """Art. 5: Check if content was screened before LLM (pre-call).""" + pre_call_guardrails = self._get_guardrails_by_mode("pre_call") + has_pre_call = len(pre_call_guardrails) > 0 + return ComplianceCheckResult( + check_name="Content screened before LLM", + article="Art. 5", + passed=has_pre_call, + detail=( + f"{len(pre_call_guardrails)} pre-call guardrail(s) screened content" + if has_pre_call + else "No pre-call screening applied" + ), + ) + + def _check_art_12_audit_complete(self) -> ComplianceCheckResult: + """Art. 12: Check if audit record is complete.""" + has_user = bool(self.data.user_id) + has_model = bool(self.data.model) + has_timestamp = bool(self.data.timestamp) + has_guardrails = len(self.guardrails) > 0 + audit_complete = has_user and has_model and has_timestamp and has_guardrails + + missing = [] + if not has_user: + missing.append("user_id") + if not has_model: + missing.append("model") + if not has_timestamp: + missing.append("timestamp") + if not has_guardrails: + missing.append("guardrail_results") + + return ComplianceCheckResult( + check_name="Audit record complete", + article="Art. 12", + passed=audit_complete, + detail=( + "All required audit fields present" + if audit_complete + else f"Missing: {', '.join(missing)}" + ), + ) + + # ── GDPR Helper Methods ────────────────────────────────────────────────── + + def _check_art_32_data_protection(self) -> ComplianceCheckResult: + """Art. 32: Check if data protection was applied (pre-call).""" + pre_call_guardrails = self._get_guardrails_by_mode("pre_call") + has_pre_call = len(pre_call_guardrails) > 0 + return ComplianceCheckResult( + check_name="Data protection applied", + article="Art. 32", + passed=has_pre_call, + detail=( + f"{len(pre_call_guardrails)} pre-call guardrail(s) protect data" + if has_pre_call + else "No pre-call data protection applied" + ), + ) + + def _check_art_5_1c_sensitive_data_protected(self) -> ComplianceCheckResult: + """Art. 5(1)(c): Check if sensitive data was protected.""" + pre_call_guardrails = self._get_guardrails_by_mode("pre_call") + has_intervention = self._has_guardrail_intervention(pre_call_guardrails) + all_passed = self._all_guardrails_passed(pre_call_guardrails) + data_protected = has_intervention or all_passed + + if has_intervention: + detail = "Guardrail intervened to protect sensitive data" + elif all_passed: + detail = "No sensitive data detected" + else: + detail = "No pre-call guardrails to protect sensitive data" + + return ComplianceCheckResult( + check_name="Sensitive data protected", + article="Art. 5(1)(c)", + passed=data_protected, + detail=detail, + ) + + def _check_art_30_audit_complete(self) -> ComplianceCheckResult: + """Art. 30: Check if audit record is complete.""" + has_user = bool(self.data.user_id) + has_model = bool(self.data.model) + has_timestamp = bool(self.data.timestamp) + has_guardrails = len(self.guardrails) > 0 + audit_complete = has_user and has_model and has_timestamp and has_guardrails + + missing = [] + if not has_user: + missing.append("user_id") + if not has_model: + missing.append("model") + if not has_timestamp: + missing.append("timestamp") + if not has_guardrails: + missing.append("guardrail_results") + + return ComplianceCheckResult( + check_name="Audit record complete", + article="Art. 30", + passed=audit_complete, + detail=( + "All required audit fields present" + if audit_complete + else f"Missing: {', '.join(missing)}" + ), + ) + + # ── Main Compliance Check Methods ──────────────────────────────────────── + + def check_eu_ai_act(self) -> List[ComplianceCheckResult]: + """ + Check EU AI Act compliance. + + Returns: + List of compliance check results for: + - Art. 9: Guardrails applied + - Art. 5: Content screened before LLM (pre-call screening) + - Art. 12: Audit record complete + """ + return [ + self._check_art_9_guardrails_applied(), + self._check_art_5_content_screened(), + self._check_art_12_audit_complete(), + ] + + def check_gdpr(self) -> List[ComplianceCheckResult]: + """ + Check GDPR compliance. + + Returns: + List of compliance check results for: + - Art. 32: Data protection applied (pre-call screening) + - Art. 5(1)(c): Sensitive data protected + - Art. 30: Audit record complete + """ + return [ + self._check_art_32_data_protection(), + self._check_art_5_1c_sensitive_data_protected(), + self._check_art_30_audit_complete(), + ] diff --git a/litellm/proxy/management_endpoints/compliance_endpoints.py b/litellm/proxy/management_endpoints/compliance_endpoints.py new file mode 100644 index 00000000000..4db99ecce13 --- /dev/null +++ b/litellm/proxy/management_endpoints/compliance_endpoints.py @@ -0,0 +1,79 @@ +""" +COMPLIANCE CHECK ENDPOINTS + +Endpoints for checking regulatory compliance of LLM request logs. + +/compliance/eu-ai-act - Check EU AI Act compliance +/compliance/gdpr - Check GDPR compliance +""" + +from fastapi import APIRouter, Depends, Request + +from litellm.proxy._types import UserAPIKeyAuth +from litellm.proxy.auth.user_api_key_auth import user_api_key_auth +from litellm.proxy.compliance_checks import ComplianceChecker +from litellm.proxy.management_helpers.utils import management_endpoint_wrapper +from litellm.types.proxy.compliance_endpoints import ( + ComplianceCheckRequest, + ComplianceResponse, +) + +router = APIRouter() + + +@router.post( + "/compliance/eu-ai-act", + tags=["compliance"], + dependencies=[Depends(user_api_key_auth)], + response_model=ComplianceResponse, +) +@management_endpoint_wrapper +async def check_eu_ai_act_compliance( + data: ComplianceCheckRequest, + http_request: Request, + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), +) -> ComplianceResponse: + """ + Check EU AI Act compliance for a spend log entry. + + Checks: + - Art. 9: Guardrails applied (any guardrail) + - Art. 5: Content screened before LLM (pre-call guardrails) + - Art. 12: Audit record complete (user_id, model, timestamp, guardrail_results) + """ + checker = ComplianceChecker(data) + checks = checker.check_eu_ai_act() + return ComplianceResponse( + compliant=all(c.passed for c in checks), + regulation="EU AI Act", + checks=checks, + ) + + +@router.post( + "/compliance/gdpr", + tags=["compliance"], + dependencies=[Depends(user_api_key_auth)], + response_model=ComplianceResponse, +) +@management_endpoint_wrapper +async def check_gdpr_compliance( + data: ComplianceCheckRequest, + http_request: Request, + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), +) -> ComplianceResponse: + """ + Check GDPR compliance for a spend log entry. + + Checks: + - Art. 32: Data protection applied (pre-call guardrails) + - Art. 5(1)(c): Sensitive data protected (masked/blocked or no issues) + - Art. 30: Audit record complete (user_id, model, timestamp, guardrail_results) + """ + checker = ComplianceChecker(data) + checks = checker.check_gdpr() + return ComplianceResponse( + compliant=all(c.passed for c in checks), + regulation="GDPR", + checks=checks, + ) diff --git a/litellm/types/proxy/compliance_endpoints.py b/litellm/types/proxy/compliance_endpoints.py new file mode 100644 index 00000000000..154c9f403af --- /dev/null +++ b/litellm/types/proxy/compliance_endpoints.py @@ -0,0 +1,33 @@ +from typing import List, Optional + +from pydantic import BaseModel + + +class ComplianceCheckResult(BaseModel): + """Result of a single compliance check.""" + + check_name: str + article: str + passed: bool + detail: str + + +class ComplianceResponse(BaseModel): + """Response from a compliance check endpoint.""" + + compliant: bool + regulation: str + checks: List[ComplianceCheckResult] + + +class ComplianceCheckRequest(BaseModel): + """Request payload for compliance check endpoints. + + Mirrors the spend log fields needed for compliance evaluation. + """ + + request_id: str + user_id: Optional[str] = None + model: Optional[str] = None + timestamp: Optional[str] = None + guardrail_information: Optional[List[dict]] = None