Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/compliance_checks.py: 72%
98 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2Compliance checker for EU AI Act and GDPR regulations.
4Provides guardrail-agnostic compliance validation based on guardrail modes
5and execution results rather than specific guardrail names.
6"""
8from typing import Final
10from litellm.types.proxy.compliance_endpoints import (
11 ComplianceCheckRequest,
12 ComplianceCheckResult,
13)
16class ComplianceChecker:
17 """
18 Validates compliance with EU AI Act and GDPR regulations.
20 Uses guardrail-agnostic checks based on:
21 - Whether any guardrails ran
22 - Guardrail execution mode (pre-call, post-call, etc.)
23 - Whether guardrails intervened/blocked content
24 - Completeness of audit records
25 """
27 def __init__(self, data: ComplianceCheckRequest):
28 self.data = data
29 self.guardrails = tuple(g for g in data.guardrail_information or () if g.get("guardrail_status") != "not_run")
31 def _get_guardrails_by_mode(self, mode: str) -> list[dict]:
32 """
33 Get all guardrails that ran in a specific mode.
35 If a guardrail doesn't have a mode specified, it's treated as pre-call
36 (the most common case).
37 """
38 return [g for g in self.guardrails if self._mode_matches(g.get("guardrail_mode"), mode)]
40 @staticmethod
41 def _mode_matches(g_mode: object, mode: str) -> bool:
42 """
43 Return True only when a guardrail with logged ``guardrail_mode`` of
44 ``g_mode`` is guaranteed to have run in ``mode`` for the audited request.
46 ``guardrail_mode`` in a spend log can take several shapes because
47 ``LitellmParams.mode`` is typed ``Union[str, List[str], Mode]``, and
48 when the event type cannot be inferred at write time the raw config is
49 logged verbatim. The spend log records the configured mode(s), not the
50 concrete hook that fired for a given request; a match reports a mode
51 satisfied only when every configured branch runs in that mode, so True
52 never claims a hook the guardrail may not have actually executed.
54 Fails safe: if the guarantee cannot be established (missing default,
55 divergent per-tag override, or a list that runs in more than one mode),
56 the guardrail counts for no mode. The precise fix is to log the
57 resolved event mode and match on it; this is the safe interim.
58 """
59 if g_mode is None: 59 ↛ 61line 59 didn't jump to line 61 because the condition on line 59 was always true
60 return mode == "pre_call"
61 if isinstance(g_mode, str):
62 return g_mode == mode
63 if isinstance(g_mode, (list, tuple)):
64 return bool(g_mode) and all(m == mode for m in g_mode)
65 if isinstance(g_mode, dict):
66 default: Final = g_mode.get("default")
67 if default is None:
68 return False
69 tags: Final = g_mode.get("tags")
70 tag_branches: Final = list(tags.values()) if isinstance(tags, dict) else []
72 def _branch_runs_in_mode(branch: object) -> bool:
73 if isinstance(branch, str):
74 return branch == mode
75 if isinstance(branch, (list, tuple)):
76 return bool(branch) and all(m == mode for m in branch)
77 return False
79 return all(_branch_runs_in_mode(branch) for branch in [default, *tag_branches])
80 return False
82 def _has_guardrail_intervention(self, guardrails: list[dict]) -> bool:
83 """Check if any guardrail intervened (blocked/masked content)."""
84 for g in guardrails:
85 status = g.get("guardrail_status", "")
86 if status in ["guardrail_intervened", "failed", "blocked"]: 86 ↛ 87line 86 didn't jump to line 87 because the condition on line 86 was never true
87 return True
88 return False
90 def _all_guardrails_passed(self, guardrails: list[dict]) -> bool:
91 """Check if all guardrails passed (no issues detected)."""
92 if not guardrails:
93 return False
94 return all(g.get("guardrail_status") == "success" for g in guardrails)
96 # ── EU AI Act Helper Methods ────────────────────────────────────────────
98 def _check_art_9_guardrails_applied(self) -> ComplianceCheckResult:
99 """Art. 9: Check if any guardrails were applied."""
100 has_guardrails: Final = len(self.guardrails) > 0
101 return ComplianceCheckResult(
102 check_name="Guardrails applied",
103 article="Art. 9",
104 passed=has_guardrails,
105 detail=(f"{len(self.guardrails)} guardrail(s) applied" if has_guardrails else "No guardrails applied"),
106 )
108 def _check_art_5_content_screened(self) -> ComplianceCheckResult:
109 """Art. 5: Check if content was screened before LLM (pre-call)."""
110 pre_call_guardrails: Final = self._get_guardrails_by_mode("pre_call")
111 has_pre_call: Final = len(pre_call_guardrails) > 0
112 return ComplianceCheckResult(
113 check_name="Content screened before LLM",
114 article="Art. 5",
115 passed=has_pre_call,
116 detail=(
117 f"{len(pre_call_guardrails)} pre-call guardrail(s) screened content"
118 if has_pre_call
119 else "No pre-call screening applied"
120 ),
121 )
123 def _check_art_12_audit_complete(self) -> ComplianceCheckResult:
124 """Art. 12: Check if audit record is complete."""
125 has_user: Final = bool(self.data.user_id)
126 has_model: Final = bool(self.data.model)
127 has_timestamp: Final = bool(self.data.timestamp)
128 has_guardrails: Final = len(self.guardrails) > 0
129 audit_complete: Final = has_user and has_model and has_timestamp and has_guardrails
131 missing: Final = []
132 if not has_user: 132 ↛ 134line 132 didn't jump to line 134 because the condition on line 132 was always true
133 missing.append("user_id")
134 if not has_model:
135 missing.append("model")
136 if not has_timestamp:
137 missing.append("timestamp")
138 if not has_guardrails:
139 missing.append("guardrail_results")
141 return ComplianceCheckResult(
142 check_name="Audit record complete",
143 article="Art. 12",
144 passed=audit_complete,
145 detail=("All required audit fields present" if audit_complete else f"Missing: {', '.join(missing)}"),
146 )
148 # ── GDPR Helper Methods ──────────────────────────────────────────────────
150 def _check_art_32_data_protection(self) -> ComplianceCheckResult:
151 """Art. 32: Check if data protection was applied (pre-call)."""
152 pre_call_guardrails: Final = self._get_guardrails_by_mode("pre_call")
153 has_pre_call: Final = len(pre_call_guardrails) > 0
154 return ComplianceCheckResult(
155 check_name="Data protection applied",
156 article="Art. 32",
157 passed=has_pre_call,
158 detail=(
159 f"{len(pre_call_guardrails)} pre-call guardrail(s) protect data"
160 if has_pre_call
161 else "No pre-call data protection applied"
162 ),
163 )
165 def _check_art_5_1c_sensitive_data_protected(self) -> ComplianceCheckResult:
166 """Art. 5(1)(c): Check if sensitive data was protected."""
167 pre_call_guardrails: Final = self._get_guardrails_by_mode("pre_call")
168 has_intervention: Final = self._has_guardrail_intervention(pre_call_guardrails)
169 all_passed: Final = self._all_guardrails_passed(pre_call_guardrails)
170 data_protected: Final = has_intervention or all_passed
172 if has_intervention: 172 ↛ 173line 172 didn't jump to line 173 because the condition on line 172 was never true
173 detail = "Guardrail intervened to protect sensitive data"
174 elif all_passed: 174 ↛ 175line 174 didn't jump to line 175 because the condition on line 174 was never true
175 detail = "No sensitive data detected"
176 else:
177 detail = "No pre-call guardrails to protect sensitive data"
179 return ComplianceCheckResult(
180 check_name="Sensitive data protected",
181 article="Art. 5(1)(c)",
182 passed=data_protected,
183 detail=detail,
184 )
186 def _check_art_30_audit_complete(self) -> ComplianceCheckResult:
187 """Art. 30: Check if audit record is complete."""
188 has_user: Final = bool(self.data.user_id)
189 has_model: Final = bool(self.data.model)
190 has_timestamp: Final = bool(self.data.timestamp)
191 has_guardrails: Final = len(self.guardrails) > 0
192 audit_complete: Final = has_user and has_model and has_timestamp and has_guardrails
194 missing: Final = []
195 if not has_user:
196 missing.append("user_id")
197 if not has_model:
198 missing.append("model")
199 if not has_timestamp:
200 missing.append("timestamp")
201 if not has_guardrails:
202 missing.append("guardrail_results")
204 return ComplianceCheckResult(
205 check_name="Audit record complete",
206 article="Art. 30",
207 passed=audit_complete,
208 detail=("All required audit fields present" if audit_complete else f"Missing: {', '.join(missing)}"),
209 )
211 # ── Main Compliance Check Methods ────────────────────────────────────────
213 def check_eu_ai_act(self) -> list[ComplianceCheckResult]:
214 """
215 Check EU AI Act compliance.
217 Returns:
218 List of compliance check results for:
219 - Art. 9: Guardrails applied
220 - Art. 5: Content screened before LLM (pre-call screening)
221 - Art. 12: Audit record complete
222 """
223 return [
224 self._check_art_9_guardrails_applied(),
225 self._check_art_5_content_screened(),
226 self._check_art_12_audit_complete(),
227 ]
229 def check_gdpr(self) -> list[ComplianceCheckResult]:
230 """
231 Check GDPR compliance.
233 Returns:
234 List of compliance check results for:
235 - Art. 32: Data protection applied (pre-call screening)
236 - Art. 5(1)(c): Sensitive data protected
237 - Art. 30: Audit record complete
238 """
239 return [
240 self._check_art_32_data_protection(),
241 self._check_art_5_1c_sensitive_data_protected(),
242 self._check_art_30_audit_complete(),
243 ]