langchain_security_extension.security_pdp_opa

 35class RemotePDPAdapter(PolicyDecisionEngine):
 36    """Remote PDP adapter connecting to OPA REST API with circuit breaker and fail-closed security.
 37
 38    Attributes:
 39        is_circuit_open (bool): True if circuit breaker is tripped due to consecutive failures.
 40    """
 41
 42    def __init__(
 43        self,
 44        endpoint_url: str,
 45        *args: Any,
 46        **kwargs: Any,
 47    ):
 48        """Initialize RemotePDPAdapter.
 49
 50        Args:
 51            endpoint_url: OPA server REST endpoint URL.
 52            *args: Optional positional arguments (timeout_seconds, failure_threshold, decision_cache, recovery_timeout_seconds, fallback_engine, audit_manager).
 53            **kwargs: Optional keyword arguments.
 54        """
 55        timeout_seconds = kwargs.get("timeout_seconds", args[0] if len(args) > 0 else DEFAULT_REMOTE_PDP_TIMEOUT_SECONDS)
 56        failure_threshold = kwargs.get("failure_threshold", args[1] if len(args) > 1 else DEFAULT_CIRCUIT_BREAKER_THRESHOLD)
 57        decision_cache = kwargs.get("decision_cache", args[2] if len(args) > 2 else None)
 58        recovery_timeout_seconds = kwargs.get(
 59            "recovery_timeout_seconds",
 60            args[3] if len(args) > 3 else DEFAULT_CIRCUIT_BREAKER_RECOVERY_SECONDS,
 61        )
 62        audit_manager = kwargs.get("audit_manager", args[4] if len(args) > 4 else None)
 63
 64        self._endpoint_url = endpoint_url
 65        self._timeout_seconds = float(timeout_seconds)
 66        self._failure_threshold = int(failure_threshold)
 67        self._recovery_timeout_seconds = float(recovery_timeout_seconds)
 68        self._consecutive_failures = 0
 69        self._state = CircuitBreakerState.CLOSED
 70        self._opened_at: Optional[float] = None
 71        self._decision_cache: Optional[TwoTierDecisionCache] = decision_cache
 72        self._audit_manager: Optional[AuditManager] = audit_manager
 73
 74    @classmethod
 75    def create(
 76        cls,
 77        endpoint_url: str,
 78        *args: Any,
 79        **kwargs: Any,
 80    ):
 81        """Factory method to construct a RemotePDPAdapter.
 82
 83        Args:
 84            endpoint_url: OPA endpoint URL.
 85            *args: Optional positional arguments.
 86            **kwargs: Optional keyword arguments.
 87
 88        Returns:
 89            RemotePDPAdapter: Constructed adapter instance.
 90        """
 91        return cls(endpoint_url, *args, **kwargs)
 92
 93    @property
 94    def state(self) -> CircuitBreakerState:
 95        """Get current circuit breaker state, transitioning OPEN -> HALF_OPEN if recovery window elapsed."""
 96        if self._state == CircuitBreakerState.OPEN and self._opened_at is not None:
 97            if time.time() - self._opened_at >= self._recovery_timeout_seconds:
 98                self._state = CircuitBreakerState.HALF_OPEN
 99        return self._state
100
101    @property
102    def is_circuit_open(self) -> bool:
103        """Check if circuit breaker is currently OPEN (tripped)."""
104        return self.state == CircuitBreakerState.OPEN
105
106    @property
107    def role_graph(self):
108        """No embedded role graph in remote adapter."""
109        return None
110
111    @property
112    def audit_manager(self) -> Optional[AuditManager]:
113        """Get configured AuditManager."""
114        return self._audit_manager
115
116    @property
117    def circuit_state(self) -> CircuitBreakerState:
118        """Get current circuit breaker state."""
119        return self.state
120
121    @property
122    def decision_cache(self) -> Optional[TwoTierDecisionCache]:
123        """Get configured TwoTierDecisionCache."""
124        return self._decision_cache
125
126    def register_invalidation_callback(self, callback: Any) -> None:
127        """Register a callback for cache invalidation events.
128
129        Args:
130            callback: Callable invoked when cache invalidation events occur.
131        """
132        callbacks = getattr(self, "_invalidation_callbacks", None)
133        if callbacks is None:
134            self._invalidation_callbacks = []
135        self._invalidation_callbacks.append(callback)
136
137    def clear_cache(self) -> None:
138        """Flush and clear all cached policy decisions in the decision cache."""
139        self.invalidate_cache()
140
141    def evict_subject_cache(self, subject_id: str, *args: Any, **kwargs: Any) -> int:
142        """Evict cached decisions for a specific subject identifier.
143
144        Args:
145            subject_id: Subject identifier whose cached decisions should be purged.
146            *args: Optional positional arguments.
147            **kwargs: Optional keyword arguments.
148
149        Returns:
150            int: Number of evicted cache entries.
151        """
152        if self._decision_cache:
153            return self._decision_cache.evict_subject(subject_id, *args, **kwargs)
154        return 0
155
156    def invalidate_cache(self) -> None:
157        """Flush cross-request static PDP decision cache upon OPA policy bundle reload."""
158        if self._decision_cache:
159            self._decision_cache.flush()
160
161    def evaluate(self, request: PolicyEvaluationRequest) -> PolicyDecision:
162        """Synchronously evaluate policy request against remote OPA endpoint.
163
164        Args:
165            request: Evaluation request envelope.
166
167        Returns:
168            PolicyDecision: Remote PDP decision verdict.
169        """
170        current_state = self.state
171        if current_state == CircuitBreakerState.OPEN:
172            return self._handle_fail_closed(
173                request=request,
174                reason="Remote PDP circuit breaker OPEN (fail-closed) [CIRCUIT_BREAKER_OPEN | LOCAL_PDP_FALLBACK]",
175            )
176
177        payload = self._build_payload(request)
178
179        try:
180            import urllib.request
181            import json
182
183            data = json.dumps(payload).encode("utf-8")
184            req = urllib.request.Request(
185                self._endpoint_url,
186                data=data,
187                headers={"Content-Type": "application/json"},
188                method="POST",
189            )
190            with urllib.request.urlopen(req, timeout=self._timeout_seconds) as resp:
191                if resp.status == 200:
192                    resp_data = json.loads(resp.read().decode("utf-8"))
193                    result = resp_data.get("result", {})
194                    allow = result.get("allow", False) if isinstance(result, dict) else bool(result)
195                    self._record_success(request)
196                    if allow:
197                        return PolicyDecision.permit("Authorized by Remote OPA PDP", {})
198                    return PolicyDecision.deny("Denied by Remote OPA PDP", {})
199
200            self._record_failure(request, "HTTP non-200 response")
201            return self._handle_fail_closed(
202                request=request,
203                reason="Remote OPA PDP returned invalid HTTP response",
204            )
205
206        except Exception as err:
207            self._record_failure(request, str(err))
208            return self._handle_fail_closed(
209                request=request,
210                reason=f"Remote OPA PDP request error: {str(err)}",
211            )
212
213    async def evaluate_async(self, request: PolicyEvaluationRequest) -> PolicyDecision:
214        """Asynchronous policy evaluation with httpx or asyncio fallback and fail-closed circuit breaker.
215
216        Args:
217            request: Evaluation request envelope.
218
219        Returns:
220            PolicyDecision: Async decision verdict.
221        """
222        current_state = self.state
223        if current_state == CircuitBreakerState.OPEN:
224            return self._handle_fail_closed(
225                request=request,
226                reason="Remote PDP circuit breaker OPEN (fail-closed) [CIRCUIT_BREAKER_OPEN | LOCAL_PDP_FALLBACK]",
227            )
228
229        payload = self._build_payload(request)
230
231        try:
232            try:
233                import httpx
234                async with httpx.AsyncClient(timeout=self._timeout_seconds) as client:
235                    resp = await client.post(self._endpoint_url, json=payload)
236                    if resp.status_code == 200:
237                        resp_data = resp.json()
238                        result = resp_data.get("result", {})
239                        allow = result.get("allow", False) if isinstance(result, dict) else bool(result)
240                        self._record_success(request)
241                        if allow:
242                            return PolicyDecision.permit("Authorized by Remote OPA PDP", {})
243                        return PolicyDecision.deny("Denied by Remote OPA PDP", {})
244            except ImportError:
245                import asyncio
246                return await asyncio.to_thread(self.evaluate, request)
247
248            self._record_failure(request, "HTTP non-200 response")
249            return self._handle_fail_closed(
250                request=request,
251                reason="Remote OPA PDP returned invalid HTTP response",
252            )
253
254        except Exception as err:
255            self._record_failure(request, str(err))
256            return self._handle_fail_closed(
257                request=request,
258                reason=f"Remote OPA PDP request error: {str(err)}",
259            )
260
261    def _build_payload(self, request: PolicyEvaluationRequest) -> Dict[str, Any]:
262        safe_context = {k: v for k, v in request.context.items() if k != "config"}
263        return {
264            "input": {
265                "subject": request.subject.to_dict(),
266                "action": request.action,
267                "resource": request.resource,
268                "context": safe_context,
269            }
270        }
271
272    def _record_success(self, request: PolicyEvaluationRequest) -> None:
273        prev_state = self._state
274        self._consecutive_failures = 0
275        self._state = CircuitBreakerState.CLOSED
276        self._opened_at = None
277        if prev_state != CircuitBreakerState.CLOSED:
278            self._emit_circuit_event(
279                request=request,
280                reason_code=EVENT_CIRCUIT_BREAKER_TRANSITION,
281                diagnostic=f"{EVENT_CIRCUIT_BREAKER_TRANSITION}: {prev_state.value} -> {CircuitBreakerState.CLOSED.value}",
282            )
283
284    def _record_failure(self, request: PolicyEvaluationRequest, error_detail: str) -> None:
285        prev_state = self._state
286        self._consecutive_failures += 1
287        if prev_state == CircuitBreakerState.HALF_OPEN or self._consecutive_failures >= self._failure_threshold:
288            self._state = CircuitBreakerState.OPEN
289            self._opened_at = time.time()
290            self._emit_circuit_event(
291                request=request,
292                reason_code=EVENT_CIRCUIT_BREAKER_OPEN,
293                diagnostic=(
294                    f"{EVENT_CIRCUIT_BREAKER_OPEN} | {EVENT_CIRCUIT_BREAKER_TRANSITION} "
295                    f"({prev_state.value} -> {self._state.value}) | LOCAL_PDP_FALLBACK: {error_detail}"
296                ),
297            )
298
299    def _emit_circuit_event(self, request: PolicyEvaluationRequest, reason_code: str, diagnostic: str) -> None:
300        req_config = request.context.get("config") if isinstance(request.context, dict) else None
301        if isinstance(req_config, dict) and isinstance(req_config.get("configurable"), dict):
302            req_config["configurable"]["sec:circuit_breaker_state"] = self._state.value
303            req_config["configurable"]["sec:circuit_breaker_transition"] = diagnostic
304
305        mgr = self.audit_manager
306        if mgr:
307            event = AuditEvent.create(
308                subject_id=request.subject.subject_id,
309                tenant_id=request.subject.tenant_id,
310                action=request.action,
311                resource=str(request.resource.get("name") or request.resource.get("target_id") or "RemotePDPAdapter"),
312                decision=DECISION_DENY,
313                diagnostic_reason=diagnostic,
314                severity=SEVERITY_WARNING,
315                metadata={"circuit_state": self._state.value},
316                reason_code=reason_code,
317            )
318            mgr.record_event(event)
319
320    def _handle_fail_closed(self, request: PolicyEvaluationRequest, reason: str) -> PolicyDecision:
321        req_config = request.context.get("config") if isinstance(request.context, dict) else None
322        if isinstance(req_config, dict) and isinstance(req_config.get("configurable"), dict):
323            req_config["configurable"]["sec:circuit_breaker_state"] = self._state.value
324
325        diag_reason = reason
326        if self._state == CircuitBreakerState.OPEN and EVENT_CIRCUIT_BREAKER_OPEN not in diag_reason:
327            diag_reason = f"{reason} [{EVENT_CIRCUIT_BREAKER_OPEN} | {EVENT_CIRCUIT_BREAKER_TRANSITION} | LOCAL_PDP_FALLBACK]"
328
329        return PolicyDecision.deny(
330            diagnostic_reason=diag_reason,
331            obligations={},
332        )
333
334    def reset_circuit_breaker(self) -> None:
335        """Reset the circuit breaker status to CLOSED and clear failure counts."""
336        self._state = CircuitBreakerState.CLOSED
337        self._opened_at = None
338        self._consecutive_failures = 0

341class OPACacheInvalidationHandler:
342    """Webhook callback handler for processing OPA bundle reload events and invalidating PDP cache."""
343
344    def __init__(self, *args: Any, **kwargs: Any):
345        """Initialize OPACacheInvalidationHandler.
346
347        Args:
348            *args: Optional positional target adapter or PDP engine.
349            **kwargs: Optional keyword arguments (remote_adapter, pdp_engine).
350        """
351        target = kwargs.get("remote_adapter", kwargs.get("pdp_engine", args[0] if len(args) > 0 else None))
352        self._remote_adapter = target
353
354    def handle_bundle_reload_webhook(self, payload: Dict[str, Any]) -> bool:
355        """Handles incoming OPA webhook notification and flushes decision cache.
356
357        Args:
358            payload: Webhook payload dictionary.
359
360        Returns:
361            bool: True upon successful processing.
362        """
363        if self._remote_adapter and hasattr(self._remote_adapter, "invalidate_cache"):
364            self._remote_adapter.invalidate_cache()
365        elif self._remote_adapter and hasattr(self._remote_adapter, "clear_cache"):
366            self._remote_adapter.clear_cache()
367        return True
368
369    def on_invalidation_event(self, payload: Dict[str, Any]) -> int:
370        """Handles identity or policy invalidation event and evicts subject or full PDP cache.
371
372        Args:
373            payload: Event payload containing subject identifier or invalidation metadata.
374
375        Returns:
376            int: Number of cache entries evicted, or 0 if a full flush was triggered.
377        """
378        subject_id = payload.get("subject_id") or payload.get("email")
379        if subject_id and self._remote_adapter and hasattr(self._remote_adapter, "evict_subject_cache"):
380            return int(self._remote_adapter.evict_subject_cache(str(subject_id)))
381        self.handle_bundle_reload_webhook(payload)
382        return 0

28class CircuitBreakerState(str, Enum):
29    """Circuit breaker state enumeration for remote PDP resilience."""
30    CLOSED = CB_STATE_CLOSED
31    OPEN = CB_STATE_OPEN
32    HALF_OPEN = CB_STATE_HALF_OPEN