langchain_security_extension.security_core
Security Core package exports.
Re-exports core primitives including SecurityIdentity, PolicyDecisionEngine, AuditManager, CELEvaluator, and exception classes for zero-trust LLM security architecture.
1""" 2Security Core package exports. 3 4Re-exports core primitives including SecurityIdentity, PolicyDecisionEngine, 5AuditManager, CELEvaluator, and exception classes for zero-trust LLM security architecture. 6""" 7 8from langchain_security_extension.security_core.identity import ( 9 SecurityIdentity, 10 InvocationParameters, 11 IdentityBridge, 12 SecurityContextHolder, 13) 14from langchain_security_extension.security_core.pdp import ( 15 PolicyEvaluationRequest, 16 PolicyDecision, 17 PolicyDecisionEngine, 18 EmbeddedPolicyEngine, 19 RoleGraph, 20 TwoTierDecisionCache, 21 PolicyRequest, 22) 23from langchain_security_extension.security_core.cel_evaluator import ( 24 CELEvaluator, 25 EntityAttributeResolverSPI, 26) 27from langchain_security_extension.security_core.registry import SecurityRegistry 28from langchain_security_extension.security_core.audit import ( 29 AuditEvent, 30 AuditManager, 31 AuditPublisherSPI, 32 AuditSanitizer, 33 ConsoleAuditLogger, 34 JsonFileAuditLogger, 35 PythonLoggingAuditPublisher, 36) 37from langchain_security_extension.security_core.prompt_injection import ( 38 PromptInjectionScanner, 39 PromptInjectionScanResult, 40) 41from langchain_security_extension.security_core.exceptions import ( 42 DiagnosticReasonEnum, 43 SecurityAuthorizationError, 44 AgentAccessDeniedException, 45 TenantIsolationError, 46 ToolExecutionDeniedError, 47 ToolExecutionFeedbackException, 48 PolicyViolationError, 49 SecurityValidationException, 50) 51from langchain_security_extension.security_core.constants import ( 52 DECISION_PERMIT, 53 DECISION_DENY, 54 ACTION_AGENT_INVOKE, 55 ACTION_TOOL_EXECUTE, 56 ACTION_CHUNK_RETRIEVE, 57 STRATEGY_HARD_ABORT, 58 STRATEGY_RETURN_FEEDBACK, 59) 60from langchain_security_extension.security_core.pii import ( 61 PiiSanitizerSPI, 62 BuiltInPiiSanitizer, 63 PiiContext, 64 get_pii_sanitizer, 65 set_pii_sanitizer, 66 reset_pii_sanitizer, 67 get_pii_scope, 68 set_pii_scope, 69 reset_pii_scope, 70 is_valid_luhn, 71 is_valid_verhoeff, 72 is_valid_aadhaar, 73 is_valid_pan, 74) 75 76__all__ = [ 77 "SecurityIdentity", 78 "InvocationParameters", 79 "IdentityBridge", 80 "SecurityContextHolder", 81 "EntityAttributeResolverSPI", 82 "PolicyEvaluationRequest", 83 "PolicyDecision", 84 "PolicyDecisionEngine", 85 "EmbeddedPolicyEngine", 86 "RoleGraph", 87 "TwoTierDecisionCache", 88 "PolicyRequest", 89 "CELEvaluator", 90 "SecurityRegistry", 91 "AuditEvent", 92 "AuditManager", 93 "AuditPublisherSPI", 94 "AuditSanitizer", 95 "ConsoleAuditLogger", 96 "JsonFileAuditLogger", 97 "PythonLoggingAuditPublisher", 98 "PromptInjectionScanner", 99 "PromptInjectionScanResult", 100 "DiagnosticReasonEnum", 101 "SecurityAuthorizationError", 102 "AgentAccessDeniedException", 103 "TenantIsolationError", 104 "ToolExecutionDeniedError", 105 "ToolExecutionFeedbackException", 106 "PolicyViolationError", 107 "SecurityValidationException", 108 "DECISION_PERMIT", 109 "DECISION_DENY", 110 "ACTION_AGENT_INVOKE", 111 "ACTION_TOOL_EXECUTE", 112 "ACTION_CHUNK_RETRIEVE", 113 "STRATEGY_HARD_ABORT", 114 "STRATEGY_RETURN_FEEDBACK", 115 "PiiSanitizerSPI", 116 "BuiltInPiiSanitizer", 117 "PiiContext", 118 "get_pii_sanitizer", 119 "set_pii_sanitizer", 120 "reset_pii_sanitizer", 121 "get_pii_scope", 122 "set_pii_scope", 123 "reset_pii_scope", 124 "is_valid_luhn", 125 "is_valid_verhoeff", 126 "is_valid_aadhaar", 127 "is_valid_pan", 128]
class
SecurityIdentity:
21class SecurityIdentity: 22 """Immutable caller identity envelope containing authenticated attributes. 23 24 Attributes: 25 subject_id (str): Unique subject identifier for caller identity. 26 tenant_id (str): Multi-tenant isolation boundary identifier. 27 roles (Set[str]): Set of security roles assigned to the subject. 28 clearance_level (int): Numeric clearance rank assigned to identity. 29 department_id (str): Department identifier for organizational access controls. 30 attributes (Dict[str, Any]): Dynamic key-value pairs associated with identity. 31 """ 32 33 def __init__( 34 self, 35 subject_id: str, 36 tenant_id: str, 37 roles: Set[str], 38 clearance_level: int, 39 department_id: str, 40 attributes: Dict[str, Any], 41 ): 42 """Initialize SecurityIdentity. 43 44 Args: 45 subject_id: Subject identifier. 46 tenant_id: Tenant isolation identifier. 47 roles: Set of granted role strings. 48 clearance_level: Numeric clearance level rank. 49 department_id: Organizational department ID. 50 attributes: Additional attribute key-value map. 51 """ 52 self._subject_id = subject_id 53 self._tenant_id = tenant_id 54 self._roles = frozenset(roles) 55 self._clearance_level = clearance_level 56 self._department_id = department_id 57 self._attributes = dict(attributes) 58 59 @classmethod 60 def create( 61 self_cls, 62 subject_id: str, 63 tenant_id: str, 64 roles: Set[str], 65 clearance_level: int, 66 department_id: str, 67 *args: Any, 68 **kwargs: Any, 69 ): 70 """Factory method to construct a new SecurityIdentity instance. 71 72 Args: 73 subject_id: Caller subject identifier. 74 tenant_id: Tenant identifier. 75 roles: Set of role strings. 76 clearance_level: Clearance level rank. 77 department_id: Department ID. 78 *args: Optional positional attributes map. 79 **kwargs: Optional keyword attributes map. 80 81 Returns: 82 SecurityIdentity: Constructed immutable identity object. 83 """ 84 attributes: Optional[Dict[str, Any]] = args[0] if len(args) > 0 else kwargs.get("attributes") 85 return self_cls( 86 subject_id=subject_id, 87 tenant_id=tenant_id, 88 roles=roles, 89 clearance_level=clearance_level, 90 department_id=department_id, 91 attributes=attributes if isinstance(attributes, dict) else {}, 92 ) 93 94 @classmethod 95 def anonymous(self_cls): 96 """Factory method constructing an unprivileged anonymous identity. 97 98 Returns: 99 SecurityIdentity: Default anonymous identity envelope. 100 """ 101 return self_cls( 102 subject_id=ANONYMOUS_SUBJECT_ID, 103 tenant_id=DEFAULT_TENANT_ID, 104 roles=set(), 105 clearance_level=DEFAULT_CLEARANCE_LEVEL, 106 department_id=DEFAULT_DEPARTMENT_ID, 107 attributes={}, 108 ) 109 110 @property 111 def subject_id(self) -> str: 112 """Get the unique subject identifier.""" 113 return self._subject_id 114 115 @property 116 def tenant_id(self) -> str: 117 """Get the tenant isolation boundary identifier.""" 118 return self._tenant_id 119 120 @property 121 def roles(self) -> Set[str]: 122 """Get the set of role entitlements held by this identity.""" 123 return set(self._roles) 124 125 @property 126 def clearance_level(self) -> int: 127 """Get the numeric clearance rank of this identity.""" 128 return self._clearance_level 129 130 @property 131 def department_id(self) -> str: 132 """Get the department identifier of this identity.""" 133 return self._department_id 134 135 @property 136 def is_anonymous(self) -> bool: 137 """Return True if identity represents an anonymous or unauthenticated subject.""" 138 return bool(self._subject_id is None or str(self._subject_id).upper() == ANONYMOUS_SUBJECT_ID) 139 140 @property 141 def attributes(self) -> Dict[str, Any]: 142 """Get a copy of dynamic attributes attached to identity.""" 143 return dict(self._attributes) 144 145 def to_dict(self) -> Dict[str, Any]: 146 """Serialize SecurityIdentity into a JSON-compatible dictionary. 147 148 Returns: 149 Dict[str, Any]: Serialized identity dictionary. 150 """ 151 return { 152 "subject_id": self._subject_id, 153 "tenant_id": self._tenant_id, 154 "roles": list(self._roles), 155 "clearance_level": self._clearance_level, 156 "department_id": self._department_id, 157 "attributes": self._attributes, 158 } 159 160 @classmethod 161 def from_dict(self_cls, data: Dict[str, Any]): 162 """Deserialize dictionary into a SecurityIdentity instance. 163 164 Args: 165 data: Dictionary containing identity fields. 166 167 Returns: 168 SecurityIdentity: Constructed identity instance. 169 """ 170 return self_cls( 171 subject_id=data.get("subject_id", ANONYMOUS_SUBJECT_ID), 172 tenant_id=data.get("tenant_id", DEFAULT_TENANT_ID), 173 roles=set(data.get("roles", [])), 174 clearance_level=int(data.get("clearance_level", DEFAULT_CLEARANCE_LEVEL)), 175 department_id=data.get("department_id", DEFAULT_DEPARTMENT_ID), 176 attributes=dict(data.get("attributes", {})), 177 ) 178 179 def with_trace_context( 180 self, 181 trace_id: Optional[str], 182 span_id: Optional[str], 183 ) -> "SecurityIdentity": 184 """Return a new SecurityIdentity with W3C OpenTelemetry trace and span identifiers attached. 185 186 Args: 187 trace_id: OpenTelemetry 32-character hexadecimal trace identifier. 188 span_id: OpenTelemetry 16-character hexadecimal span identifier. 189 190 Returns: 191 SecurityIdentity: New immutable identity instance containing the trace context. 192 """ 193 attrs = dict(self._attributes) 194 if trace_id is not None: 195 attrs["trace_id"] = trace_id 196 if span_id is not None: 197 attrs["span_id"] = span_id 198 return self.__class__.create( 199 subject_id=self._subject_id, 200 tenant_id=self._tenant_id, 201 roles=self.roles, 202 clearance_level=self._clearance_level, 203 department_id=self._department_id, 204 attributes=attrs, 205 )
Immutable caller identity envelope containing authenticated attributes.
Attributes: subject_id (str): Unique subject identifier for caller identity. tenant_id (str): Multi-tenant isolation boundary identifier. roles (Set[str]): Set of security roles assigned to the subject. clearance_level (int): Numeric clearance rank assigned to identity. department_id (str): Department identifier for organizational access controls. attributes (Dict[str, Any]): Dynamic key-value pairs associated with identity.
SecurityIdentity( subject_id: str, tenant_id: str, roles: Set[str], clearance_level: int, department_id: str, attributes: Dict[str, Any])
33 def __init__( 34 self, 35 subject_id: str, 36 tenant_id: str, 37 roles: Set[str], 38 clearance_level: int, 39 department_id: str, 40 attributes: Dict[str, Any], 41 ): 42 """Initialize SecurityIdentity. 43 44 Args: 45 subject_id: Subject identifier. 46 tenant_id: Tenant isolation identifier. 47 roles: Set of granted role strings. 48 clearance_level: Numeric clearance level rank. 49 department_id: Organizational department ID. 50 attributes: Additional attribute key-value map. 51 """ 52 self._subject_id = subject_id 53 self._tenant_id = tenant_id 54 self._roles = frozenset(roles) 55 self._clearance_level = clearance_level 56 self._department_id = department_id 57 self._attributes = dict(attributes)
Initialize SecurityIdentity.
Args: subject_id: Subject identifier. tenant_id: Tenant isolation identifier. roles: Set of granted role strings. clearance_level: Numeric clearance level rank. department_id: Organizational department ID. attributes: Additional attribute key-value map.
@classmethod
def
create( self_cls, subject_id: str, tenant_id: str, roles: Set[str], clearance_level: int, department_id: str, *args: Any, **kwargs: Any):
59 @classmethod 60 def create( 61 self_cls, 62 subject_id: str, 63 tenant_id: str, 64 roles: Set[str], 65 clearance_level: int, 66 department_id: str, 67 *args: Any, 68 **kwargs: Any, 69 ): 70 """Factory method to construct a new SecurityIdentity instance. 71 72 Args: 73 subject_id: Caller subject identifier. 74 tenant_id: Tenant identifier. 75 roles: Set of role strings. 76 clearance_level: Clearance level rank. 77 department_id: Department ID. 78 *args: Optional positional attributes map. 79 **kwargs: Optional keyword attributes map. 80 81 Returns: 82 SecurityIdentity: Constructed immutable identity object. 83 """ 84 attributes: Optional[Dict[str, Any]] = args[0] if len(args) > 0 else kwargs.get("attributes") 85 return self_cls( 86 subject_id=subject_id, 87 tenant_id=tenant_id, 88 roles=roles, 89 clearance_level=clearance_level, 90 department_id=department_id, 91 attributes=attributes if isinstance(attributes, dict) else {}, 92 )
Factory method to construct a new SecurityIdentity instance.
Args: subject_id: Caller subject identifier. tenant_id: Tenant identifier. roles: Set of role strings. clearance_level: Clearance level rank. department_id: Department ID. *args: Optional positional attributes map. **kwargs: Optional keyword attributes map.
Returns: SecurityIdentity: Constructed immutable identity object.
@classmethod
def
anonymous(self_cls):
94 @classmethod 95 def anonymous(self_cls): 96 """Factory method constructing an unprivileged anonymous identity. 97 98 Returns: 99 SecurityIdentity: Default anonymous identity envelope. 100 """ 101 return self_cls( 102 subject_id=ANONYMOUS_SUBJECT_ID, 103 tenant_id=DEFAULT_TENANT_ID, 104 roles=set(), 105 clearance_level=DEFAULT_CLEARANCE_LEVEL, 106 department_id=DEFAULT_DEPARTMENT_ID, 107 attributes={}, 108 )
Factory method constructing an unprivileged anonymous identity.
Returns: SecurityIdentity: Default anonymous identity envelope.
subject_id: str
110 @property 111 def subject_id(self) -> str: 112 """Get the unique subject identifier.""" 113 return self._subject_id
Get the unique subject identifier.
tenant_id: str
115 @property 116 def tenant_id(self) -> str: 117 """Get the tenant isolation boundary identifier.""" 118 return self._tenant_id
Get the tenant isolation boundary identifier.
roles: Set[str]
120 @property 121 def roles(self) -> Set[str]: 122 """Get the set of role entitlements held by this identity.""" 123 return set(self._roles)
Get the set of role entitlements held by this identity.
clearance_level: int
125 @property 126 def clearance_level(self) -> int: 127 """Get the numeric clearance rank of this identity.""" 128 return self._clearance_level
Get the numeric clearance rank of this identity.
department_id: str
130 @property 131 def department_id(self) -> str: 132 """Get the department identifier of this identity.""" 133 return self._department_id
Get the department identifier of this identity.
is_anonymous: bool
135 @property 136 def is_anonymous(self) -> bool: 137 """Return True if identity represents an anonymous or unauthenticated subject.""" 138 return bool(self._subject_id is None or str(self._subject_id).upper() == ANONYMOUS_SUBJECT_ID)
Return True if identity represents an anonymous or unauthenticated subject.
attributes: Dict[str, Any]
140 @property 141 def attributes(self) -> Dict[str, Any]: 142 """Get a copy of dynamic attributes attached to identity.""" 143 return dict(self._attributes)
Get a copy of dynamic attributes attached to identity.
def
to_dict(self) -> Dict[str, Any]:
145 def to_dict(self) -> Dict[str, Any]: 146 """Serialize SecurityIdentity into a JSON-compatible dictionary. 147 148 Returns: 149 Dict[str, Any]: Serialized identity dictionary. 150 """ 151 return { 152 "subject_id": self._subject_id, 153 "tenant_id": self._tenant_id, 154 "roles": list(self._roles), 155 "clearance_level": self._clearance_level, 156 "department_id": self._department_id, 157 "attributes": self._attributes, 158 }
Serialize SecurityIdentity into a JSON-compatible dictionary.
Returns: Dict[str, Any]: Serialized identity dictionary.
@classmethod
def
from_dict(self_cls, data: Dict[str, Any]):
160 @classmethod 161 def from_dict(self_cls, data: Dict[str, Any]): 162 """Deserialize dictionary into a SecurityIdentity instance. 163 164 Args: 165 data: Dictionary containing identity fields. 166 167 Returns: 168 SecurityIdentity: Constructed identity instance. 169 """ 170 return self_cls( 171 subject_id=data.get("subject_id", ANONYMOUS_SUBJECT_ID), 172 tenant_id=data.get("tenant_id", DEFAULT_TENANT_ID), 173 roles=set(data.get("roles", [])), 174 clearance_level=int(data.get("clearance_level", DEFAULT_CLEARANCE_LEVEL)), 175 department_id=data.get("department_id", DEFAULT_DEPARTMENT_ID), 176 attributes=dict(data.get("attributes", {})), 177 )
Deserialize dictionary into a SecurityIdentity instance.
Args: data: Dictionary containing identity fields.
Returns: SecurityIdentity: Constructed identity instance.
179 def with_trace_context( 180 self, 181 trace_id: Optional[str], 182 span_id: Optional[str], 183 ) -> "SecurityIdentity": 184 """Return a new SecurityIdentity with W3C OpenTelemetry trace and span identifiers attached. 185 186 Args: 187 trace_id: OpenTelemetry 32-character hexadecimal trace identifier. 188 span_id: OpenTelemetry 16-character hexadecimal span identifier. 189 190 Returns: 191 SecurityIdentity: New immutable identity instance containing the trace context. 192 """ 193 attrs = dict(self._attributes) 194 if trace_id is not None: 195 attrs["trace_id"] = trace_id 196 if span_id is not None: 197 attrs["span_id"] = span_id 198 return self.__class__.create( 199 subject_id=self._subject_id, 200 tenant_id=self._tenant_id, 201 roles=self.roles, 202 clearance_level=self._clearance_level, 203 department_id=self._department_id, 204 attributes=attrs, 205 )
Return a new SecurityIdentity with W3C OpenTelemetry trace and span identifiers attached.
Args: trace_id: OpenTelemetry 32-character hexadecimal trace identifier. span_id: OpenTelemetry 16-character hexadecimal span identifier.
Returns: SecurityIdentity: New immutable identity instance containing the trace context.
class
InvocationParameters:
208class InvocationParameters: 209 """In-band execution parameters envelope carried inside framework metadata/state. 210 211 Attributes: 212 security_identity (SecurityIdentity): Caller identity envelope. 213 metadata (Dict[str, Any]): Framework invocation metadata dictionary. 214 """ 215 216 def __init__(self, security_identity: SecurityIdentity, metadata: Dict[str, Any]): 217 """Initialize InvocationParameters. 218 219 Args: 220 security_identity: Embedded caller identity. 221 metadata: Associated invocation metadata dictionary. 222 """ 223 self._security_identity = security_identity 224 self._metadata = dict(metadata) 225 226 @classmethod 227 def create(self_cls, security_identity: SecurityIdentity, metadata: Dict[str, Any]): 228 """Factory method constructing an InvocationParameters instance. 229 230 Args: 231 security_identity: Security identity object. 232 metadata: Metadata mapping. 233 234 Returns: 235 InvocationParameters: Constructed parameters instance. 236 """ 237 return self_cls(security_identity=security_identity, metadata=metadata) 238 239 @property 240 def security_identity(self) -> SecurityIdentity: 241 """Get the embedded SecurityIdentity.""" 242 return self._security_identity 243 244 @property 245 def metadata(self) -> Dict[str, Any]: 246 """Get copy of invocation metadata dictionary.""" 247 return dict(self._metadata) 248 249 def serialize_to_state(self) -> Dict[str, Any]: 250 """Serialize parameters into a state dictionary using standard security key. 251 252 Returns: 253 Dict[str, Any]: Dictionary suitable for graph state or framework config. 254 """ 255 return { 256 SECURITY_INVOCATION_PARAMS_KEY: { 257 "identity": self._security_identity.to_dict(), 258 "metadata": self._metadata, 259 } 260 } 261 262 @classmethod 263 def deserialize_from_state(self_cls, state_dict: Dict[str, Any]): 264 """Extract and deserialize InvocationParameters from state dictionary. 265 266 Args: 267 state_dict: State or config dictionary. 268 269 Returns: 270 InvocationParameters: Deserialized parameters or default anonymous instance. 271 """ 272 if SECURITY_INVOCATION_PARAMS_KEY not in state_dict: 273 return self_cls( 274 security_identity=SecurityIdentity.anonymous(), 275 metadata={}, 276 ) 277 sec_payload = state_dict[SECURITY_INVOCATION_PARAMS_KEY] 278 identity_data = sec_payload.get("identity", {}) 279 metadata_data = sec_payload.get("metadata", {}) 280 identity = SecurityIdentity.from_dict(identity_data) 281 return self_cls(security_identity=identity, metadata=metadata_data)
In-band execution parameters envelope carried inside framework metadata/state.
Attributes: security_identity (SecurityIdentity): Caller identity envelope. metadata (Dict[str, Any]): Framework invocation metadata dictionary.
InvocationParameters( security_identity: SecurityIdentity, metadata: Dict[str, Any])
216 def __init__(self, security_identity: SecurityIdentity, metadata: Dict[str, Any]): 217 """Initialize InvocationParameters. 218 219 Args: 220 security_identity: Embedded caller identity. 221 metadata: Associated invocation metadata dictionary. 222 """ 223 self._security_identity = security_identity 224 self._metadata = dict(metadata)
Initialize InvocationParameters.
Args: security_identity: Embedded caller identity. metadata: Associated invocation metadata dictionary.
226 @classmethod 227 def create(self_cls, security_identity: SecurityIdentity, metadata: Dict[str, Any]): 228 """Factory method constructing an InvocationParameters instance. 229 230 Args: 231 security_identity: Security identity object. 232 metadata: Metadata mapping. 233 234 Returns: 235 InvocationParameters: Constructed parameters instance. 236 """ 237 return self_cls(security_identity=security_identity, metadata=metadata)
Factory method constructing an InvocationParameters instance.
Args: security_identity: Security identity object. metadata: Metadata mapping.
Returns: InvocationParameters: Constructed parameters instance.
security_identity: SecurityIdentity
239 @property 240 def security_identity(self) -> SecurityIdentity: 241 """Get the embedded SecurityIdentity.""" 242 return self._security_identity
Get the embedded SecurityIdentity.
metadata: Dict[str, Any]
244 @property 245 def metadata(self) -> Dict[str, Any]: 246 """Get copy of invocation metadata dictionary.""" 247 return dict(self._metadata)
Get copy of invocation metadata dictionary.
def
serialize_to_state(self) -> Dict[str, Any]:
249 def serialize_to_state(self) -> Dict[str, Any]: 250 """Serialize parameters into a state dictionary using standard security key. 251 252 Returns: 253 Dict[str, Any]: Dictionary suitable for graph state or framework config. 254 """ 255 return { 256 SECURITY_INVOCATION_PARAMS_KEY: { 257 "identity": self._security_identity.to_dict(), 258 "metadata": self._metadata, 259 } 260 }
Serialize parameters into a state dictionary using standard security key.
Returns: Dict[str, Any]: Dictionary suitable for graph state or framework config.
@classmethod
def
deserialize_from_state(self_cls, state_dict: Dict[str, Any]):
262 @classmethod 263 def deserialize_from_state(self_cls, state_dict: Dict[str, Any]): 264 """Extract and deserialize InvocationParameters from state dictionary. 265 266 Args: 267 state_dict: State or config dictionary. 268 269 Returns: 270 InvocationParameters: Deserialized parameters or default anonymous instance. 271 """ 272 if SECURITY_INVOCATION_PARAMS_KEY not in state_dict: 273 return self_cls( 274 security_identity=SecurityIdentity.anonymous(), 275 metadata={}, 276 ) 277 sec_payload = state_dict[SECURITY_INVOCATION_PARAMS_KEY] 278 identity_data = sec_payload.get("identity", {}) 279 metadata_data = sec_payload.get("metadata", {}) 280 identity = SecurityIdentity.from_dict(identity_data) 281 return self_cls(security_identity=identity, metadata=metadata_data)
Extract and deserialize InvocationParameters from state dictionary.
Args: state_dict: State or config dictionary.
Returns: InvocationParameters: Deserialized parameters or default anonymous instance.
class
IdentityBridge:
312class IdentityBridge: 313 """Interface and helper utilities for resolving SecurityIdentity from contexts, configs, or headers.""" 314 315 @staticmethod 316 def resolve_identity(candidate_identity: Optional[SecurityIdentity]) -> SecurityIdentity: 317 """Resolve identity from candidate, current context holder, or fall back to anonymous identity. 318 319 Args: 320 candidate_identity: Direct identity instance if available. 321 322 Returns: 323 SecurityIdentity: Resolved identity object. 324 """ 325 if candidate_identity is not None: 326 return candidate_identity 327 ctx_identity = SecurityContextHolder.get_identity() 328 if ctx_identity is not None: 329 return ctx_identity 330 return SecurityIdentity.anonymous() 331 332 @staticmethod 333 def extract_from_config(config: Optional[Dict[str, Any]]) -> Optional[SecurityIdentity]: 334 """Extract SecurityIdentity from Runnable standard config dictionary. 335 336 Args: 337 config: Runnable config dictionary containing 'configurable' keys. 338 339 Returns: 340 Optional[SecurityIdentity]: Extracted identity object or None if missing. 341 """ 342 if not config or not isinstance(config, dict): 343 return None 344 configurable = config.get("configurable") 345 if isinstance(configurable, dict): 346 sec_id = configurable.get("security_identity") 347 if isinstance(sec_id, SecurityIdentity): 348 return sec_id 349 elif isinstance(sec_id, dict): 350 return SecurityIdentity.from_dict(sec_id) 351 return None 352 353 @staticmethod 354 def create_runnable_config(identity: SecurityIdentity) -> Dict[str, Any]: 355 """Wrap SecurityIdentity into native LangChain RunnableConfig dictionary. 356 357 Args: 358 identity: SecurityIdentity envelope. 359 360 Returns: 361 Dict[str, Any]: RunnableConfig dictionary with 'configurable' mapping. 362 """ 363 return { 364 "configurable": { 365 "security_identity": identity 366 } 367 } 368 369 @classmethod 370 def build_identity_from_claims( 371 cls, 372 claims: Dict[str, Any], 373 clearance_rank_map: Optional[Dict[str, int]], 374 default_clearance: int, 375 ) -> SecurityIdentity: 376 """Construct SecurityIdentity from verified JWT claims or caller metadata. 377 378 Args: 379 claims: Token claims or user identity mapping. 380 clearance_rank_map: Optional mapping of clearance names to numeric ranks. 381 default_clearance: Numeric clearance rank floor. 382 383 Returns: 384 SecurityIdentity: Immutable security envelope. 385 """ 386 sub = str(claims.get("sub") or claims.get("email") or claims.get("user_id") or ANONYMOUS_SUBJECT_ID) 387 tenant = str(claims.get("tenantId") or claims.get("tenant_id") or DEFAULT_TENANT_ID) 388 dept = str(claims.get("department") or claims.get("dept") or DEFAULT_DEPARTMENT_ID) 389 390 raw_roles = claims.get("roles") or claims.get("assignedRoles") or [] 391 roles_set: Set[str] = set() 392 if isinstance(raw_roles, list): 393 for r in raw_roles: 394 if isinstance(r, str): 395 roles_set.update([role.strip() for role in r.split(",") if role.strip()]) 396 elif isinstance(raw_roles, str): 397 roles_set.update([role.strip() for role in raw_roles.split(",") if role.strip()]) 398 399 raw_clr = claims.get("clearanceLevel") or claims.get("clearance") or claims.get("clearance_level") 400 if isinstance(raw_clr, int): 401 clr = raw_clr 402 elif isinstance(raw_clr, str) and raw_clr.isdigit(): 403 clr = int(raw_clr) 404 elif clearance_rank_map and isinstance(raw_clr, str): 405 clr = clearance_rank_map.get(raw_clr.upper(), default_clearance) 406 elif clearance_rank_map and roles_set: 407 role_ranks = [clearance_rank_map.get(r, default_clearance) for r in roles_set if r != "ROLE_GUEST"] 408 clr = max(role_ranks) if role_ranks else default_clearance 409 else: 410 clr = default_clearance 411 412 attrs = dict(claims.get("attributes", {})) 413 if claims.get("trace_id"): 414 attrs["trace_id"] = claims["trace_id"] 415 if claims.get("span_id"): 416 attrs["span_id"] = claims["span_id"] 417 418 return SecurityIdentity.create( 419 subject_id=sub, 420 tenant_id=tenant, 421 roles=roles_set, 422 clearance_level=clr, 423 department_id=dept, 424 attributes=attrs, 425 ) 426 427 def resolve_identity_from_context(self, headers_or_context: Dict[str, Any]) -> SecurityIdentity: 428 """Abstract method for resolving identity from raw request context or HTTP headers. 429 430 Args: 431 headers_or_context: Context or headers mapping. 432 433 Returns: 434 SecurityIdentity: Resolved identity. 435 436 Raises: 437 NotImplementedError: If subclass does not implement method. 438 """ 439 raise NotImplementedError("Subclasses must implement resolve_identity_from_context") 440 441 @staticmethod 442 def record_tenant_violation( 443 audit_manager: Any, 444 subject_id: str, 445 tenant_id: str, 446 action: str, 447 resource: str, 448 expected_tenant: str, 449 trace_id: Optional[str], 450 span_id: Optional[str], 451 ) -> Any: 452 """Record cross-tenant access violation event to audit manager. 453 454 Args: 455 audit_manager: Target audit manager instance. 456 subject_id: Subject attempting access. 457 tenant_id: Claimed tenant ID. 458 action: Requested action. 459 resource: Resource path or name. 460 expected_tenant: Valid tenant boundary ID. 461 trace_id: OpenTelemetry trace ID. 462 span_id: OpenTelemetry span ID. 463 464 Returns: 465 AuditEvent: Constructed and published audit event object. 466 """ 467 from langchain_security_extension.security_core.audit import AuditEvent 468 diagnostic_reason = f"Cross-tenant isolation violation: tenantId={tenant_id}" 469 metadata = { 470 "threat_category": THREAT_CATEGORY_IDENTITY_SPOOFING, 471 "reason_code": REASON_TENANT_BOUND_VIOLATION, 472 "mitigation_action": MITIGATION_BLOCKED, 473 "claimedTenant": tenant_id, 474 "expectedTenant": expected_tenant, 475 "trace_id": trace_id, 476 "span_id": span_id, 477 } 478 event = AuditEvent.create( 479 subject_id=subject_id, 480 tenant_id=tenant_id, 481 action=action, 482 resource=resource, 483 decision=DECISION_DENY, 484 diagnostic_reason=diagnostic_reason, 485 severity=SEVERITY_CRITICAL, 486 metadata=metadata, 487 enforcement_point=ENFORCEMENT_POINT_IDENTITY_BRIDGE, 488 reason_code=REASON_TENANT_BOUND_VIOLATION, 489 ) 490 event.threat_category = THREAT_CATEGORY_IDENTITY_SPOOFING 491 event.mitigation_action = MITIGATION_BLOCKED 492 if trace_id: 493 event.trace_id = trace_id 494 if span_id: 495 event.span_id = span_id 496 if audit_manager: 497 audit_manager.record_event(event) 498 return event 499 500 @staticmethod 501 def record_auth_success( 502 audit_manager: Any, 503 subject_id: str, 504 tenant_id: str, 505 action: str, 506 resource: str, 507 department: str, 508 roles: List[str], 509 clearance_level: Any, 510 trace_id: Optional[str], 511 span_id: Optional[str], 512 ) -> Any: 513 """Record successful identity authentication event to audit manager. 514 515 Args: 516 audit_manager: Target audit manager. 517 subject_id: Subject identifier. 518 tenant_id: Tenant identifier. 519 action: Action identifier. 520 resource: Resource identifier. 521 department: Department ID. 522 roles: List of granted roles. 523 clearance_level: Clearance rank. 524 trace_id: OpenTelemetry trace ID. 525 span_id: OpenTelemetry span ID. 526 527 Returns: 528 AuditEvent: Constructed and logged audit event. 529 """ 530 from langchain_security_extension.security_core.audit import AuditEvent 531 metadata = { 532 "department": department, 533 "roles": roles, 534 "clearanceLevel": clearance_level, 535 "trace_id": trace_id, 536 "span_id": span_id, 537 } 538 event = AuditEvent.create( 539 subject_id=subject_id, 540 tenant_id=tenant_id, 541 action=action, 542 resource=resource, 543 decision=DECISION_PERMIT, 544 diagnostic_reason="AUTHENTICATED", 545 severity=SEVERITY_INFO, 546 metadata=metadata, 547 enforcement_point=ENFORCEMENT_POINT_IDENTITY_BRIDGE, 548 reason_code=REASON_AUTHENTICATION_SUCCESS, 549 ) 550 if trace_id: 551 event.trace_id = trace_id 552 if span_id: 553 event.span_id = span_id 554 if audit_manager: 555 audit_manager.record_event(event) 556 return event
Interface and helper utilities for resolving SecurityIdentity from contexts, configs, or headers.
@staticmethod
def
resolve_identity( candidate_identity: Optional[SecurityIdentity]) -> SecurityIdentity:
315 @staticmethod 316 def resolve_identity(candidate_identity: Optional[SecurityIdentity]) -> SecurityIdentity: 317 """Resolve identity from candidate, current context holder, or fall back to anonymous identity. 318 319 Args: 320 candidate_identity: Direct identity instance if available. 321 322 Returns: 323 SecurityIdentity: Resolved identity object. 324 """ 325 if candidate_identity is not None: 326 return candidate_identity 327 ctx_identity = SecurityContextHolder.get_identity() 328 if ctx_identity is not None: 329 return ctx_identity 330 return SecurityIdentity.anonymous()
Resolve identity from candidate, current context holder, or fall back to anonymous identity.
Args: candidate_identity: Direct identity instance if available.
Returns: SecurityIdentity: Resolved identity object.
@staticmethod
def
extract_from_config( config: Optional[Dict[str, Any]]) -> Optional[SecurityIdentity]:
332 @staticmethod 333 def extract_from_config(config: Optional[Dict[str, Any]]) -> Optional[SecurityIdentity]: 334 """Extract SecurityIdentity from Runnable standard config dictionary. 335 336 Args: 337 config: Runnable config dictionary containing 'configurable' keys. 338 339 Returns: 340 Optional[SecurityIdentity]: Extracted identity object or None if missing. 341 """ 342 if not config or not isinstance(config, dict): 343 return None 344 configurable = config.get("configurable") 345 if isinstance(configurable, dict): 346 sec_id = configurable.get("security_identity") 347 if isinstance(sec_id, SecurityIdentity): 348 return sec_id 349 elif isinstance(sec_id, dict): 350 return SecurityIdentity.from_dict(sec_id) 351 return None
Extract SecurityIdentity from Runnable standard config dictionary.
Args: config: Runnable config dictionary containing 'configurable' keys.
Returns: Optional[SecurityIdentity]: Extracted identity object or None if missing.
353 @staticmethod 354 def create_runnable_config(identity: SecurityIdentity) -> Dict[str, Any]: 355 """Wrap SecurityIdentity into native LangChain RunnableConfig dictionary. 356 357 Args: 358 identity: SecurityIdentity envelope. 359 360 Returns: 361 Dict[str, Any]: RunnableConfig dictionary with 'configurable' mapping. 362 """ 363 return { 364 "configurable": { 365 "security_identity": identity 366 } 367 }
Wrap SecurityIdentity into native LangChain RunnableConfig dictionary.
Args: identity: SecurityIdentity envelope.
Returns: Dict[str, Any]: RunnableConfig dictionary with 'configurable' mapping.
@classmethod
def
build_identity_from_claims( cls, claims: Dict[str, Any], clearance_rank_map: Optional[Dict[str, int]], default_clearance: int) -> SecurityIdentity:
369 @classmethod 370 def build_identity_from_claims( 371 cls, 372 claims: Dict[str, Any], 373 clearance_rank_map: Optional[Dict[str, int]], 374 default_clearance: int, 375 ) -> SecurityIdentity: 376 """Construct SecurityIdentity from verified JWT claims or caller metadata. 377 378 Args: 379 claims: Token claims or user identity mapping. 380 clearance_rank_map: Optional mapping of clearance names to numeric ranks. 381 default_clearance: Numeric clearance rank floor. 382 383 Returns: 384 SecurityIdentity: Immutable security envelope. 385 """ 386 sub = str(claims.get("sub") or claims.get("email") or claims.get("user_id") or ANONYMOUS_SUBJECT_ID) 387 tenant = str(claims.get("tenantId") or claims.get("tenant_id") or DEFAULT_TENANT_ID) 388 dept = str(claims.get("department") or claims.get("dept") or DEFAULT_DEPARTMENT_ID) 389 390 raw_roles = claims.get("roles") or claims.get("assignedRoles") or [] 391 roles_set: Set[str] = set() 392 if isinstance(raw_roles, list): 393 for r in raw_roles: 394 if isinstance(r, str): 395 roles_set.update([role.strip() for role in r.split(",") if role.strip()]) 396 elif isinstance(raw_roles, str): 397 roles_set.update([role.strip() for role in raw_roles.split(",") if role.strip()]) 398 399 raw_clr = claims.get("clearanceLevel") or claims.get("clearance") or claims.get("clearance_level") 400 if isinstance(raw_clr, int): 401 clr = raw_clr 402 elif isinstance(raw_clr, str) and raw_clr.isdigit(): 403 clr = int(raw_clr) 404 elif clearance_rank_map and isinstance(raw_clr, str): 405 clr = clearance_rank_map.get(raw_clr.upper(), default_clearance) 406 elif clearance_rank_map and roles_set: 407 role_ranks = [clearance_rank_map.get(r, default_clearance) for r in roles_set if r != "ROLE_GUEST"] 408 clr = max(role_ranks) if role_ranks else default_clearance 409 else: 410 clr = default_clearance 411 412 attrs = dict(claims.get("attributes", {})) 413 if claims.get("trace_id"): 414 attrs["trace_id"] = claims["trace_id"] 415 if claims.get("span_id"): 416 attrs["span_id"] = claims["span_id"] 417 418 return SecurityIdentity.create( 419 subject_id=sub, 420 tenant_id=tenant, 421 roles=roles_set, 422 clearance_level=clr, 423 department_id=dept, 424 attributes=attrs, 425 )
Construct SecurityIdentity from verified JWT claims or caller metadata.
Args: claims: Token claims or user identity mapping. clearance_rank_map: Optional mapping of clearance names to numeric ranks. default_clearance: Numeric clearance rank floor.
Returns: SecurityIdentity: Immutable security envelope.
427 def resolve_identity_from_context(self, headers_or_context: Dict[str, Any]) -> SecurityIdentity: 428 """Abstract method for resolving identity from raw request context or HTTP headers. 429 430 Args: 431 headers_or_context: Context or headers mapping. 432 433 Returns: 434 SecurityIdentity: Resolved identity. 435 436 Raises: 437 NotImplementedError: If subclass does not implement method. 438 """ 439 raise NotImplementedError("Subclasses must implement resolve_identity_from_context")
Abstract method for resolving identity from raw request context or HTTP headers.
Args: headers_or_context: Context or headers mapping.
Returns: SecurityIdentity: Resolved identity.
Raises: NotImplementedError: If subclass does not implement method.
@staticmethod
def
record_tenant_violation( audit_manager: Any, subject_id: str, tenant_id: str, action: str, resource: str, expected_tenant: str, trace_id: Optional[str], span_id: Optional[str]) -> Any:
441 @staticmethod 442 def record_tenant_violation( 443 audit_manager: Any, 444 subject_id: str, 445 tenant_id: str, 446 action: str, 447 resource: str, 448 expected_tenant: str, 449 trace_id: Optional[str], 450 span_id: Optional[str], 451 ) -> Any: 452 """Record cross-tenant access violation event to audit manager. 453 454 Args: 455 audit_manager: Target audit manager instance. 456 subject_id: Subject attempting access. 457 tenant_id: Claimed tenant ID. 458 action: Requested action. 459 resource: Resource path or name. 460 expected_tenant: Valid tenant boundary ID. 461 trace_id: OpenTelemetry trace ID. 462 span_id: OpenTelemetry span ID. 463 464 Returns: 465 AuditEvent: Constructed and published audit event object. 466 """ 467 from langchain_security_extension.security_core.audit import AuditEvent 468 diagnostic_reason = f"Cross-tenant isolation violation: tenantId={tenant_id}" 469 metadata = { 470 "threat_category": THREAT_CATEGORY_IDENTITY_SPOOFING, 471 "reason_code": REASON_TENANT_BOUND_VIOLATION, 472 "mitigation_action": MITIGATION_BLOCKED, 473 "claimedTenant": tenant_id, 474 "expectedTenant": expected_tenant, 475 "trace_id": trace_id, 476 "span_id": span_id, 477 } 478 event = AuditEvent.create( 479 subject_id=subject_id, 480 tenant_id=tenant_id, 481 action=action, 482 resource=resource, 483 decision=DECISION_DENY, 484 diagnostic_reason=diagnostic_reason, 485 severity=SEVERITY_CRITICAL, 486 metadata=metadata, 487 enforcement_point=ENFORCEMENT_POINT_IDENTITY_BRIDGE, 488 reason_code=REASON_TENANT_BOUND_VIOLATION, 489 ) 490 event.threat_category = THREAT_CATEGORY_IDENTITY_SPOOFING 491 event.mitigation_action = MITIGATION_BLOCKED 492 if trace_id: 493 event.trace_id = trace_id 494 if span_id: 495 event.span_id = span_id 496 if audit_manager: 497 audit_manager.record_event(event) 498 return event
Record cross-tenant access violation event to audit manager.
Args: audit_manager: Target audit manager instance. subject_id: Subject attempting access. tenant_id: Claimed tenant ID. action: Requested action. resource: Resource path or name. expected_tenant: Valid tenant boundary ID. trace_id: OpenTelemetry trace ID. span_id: OpenTelemetry span ID.
Returns: AuditEvent: Constructed and published audit event object.
@staticmethod
def
record_auth_success( audit_manager: Any, subject_id: str, tenant_id: str, action: str, resource: str, department: str, roles: List[str], clearance_level: Any, trace_id: Optional[str], span_id: Optional[str]) -> Any:
500 @staticmethod 501 def record_auth_success( 502 audit_manager: Any, 503 subject_id: str, 504 tenant_id: str, 505 action: str, 506 resource: str, 507 department: str, 508 roles: List[str], 509 clearance_level: Any, 510 trace_id: Optional[str], 511 span_id: Optional[str], 512 ) -> Any: 513 """Record successful identity authentication event to audit manager. 514 515 Args: 516 audit_manager: Target audit manager. 517 subject_id: Subject identifier. 518 tenant_id: Tenant identifier. 519 action: Action identifier. 520 resource: Resource identifier. 521 department: Department ID. 522 roles: List of granted roles. 523 clearance_level: Clearance rank. 524 trace_id: OpenTelemetry trace ID. 525 span_id: OpenTelemetry span ID. 526 527 Returns: 528 AuditEvent: Constructed and logged audit event. 529 """ 530 from langchain_security_extension.security_core.audit import AuditEvent 531 metadata = { 532 "department": department, 533 "roles": roles, 534 "clearanceLevel": clearance_level, 535 "trace_id": trace_id, 536 "span_id": span_id, 537 } 538 event = AuditEvent.create( 539 subject_id=subject_id, 540 tenant_id=tenant_id, 541 action=action, 542 resource=resource, 543 decision=DECISION_PERMIT, 544 diagnostic_reason="AUTHENTICATED", 545 severity=SEVERITY_INFO, 546 metadata=metadata, 547 enforcement_point=ENFORCEMENT_POINT_IDENTITY_BRIDGE, 548 reason_code=REASON_AUTHENTICATION_SUCCESS, 549 ) 550 if trace_id: 551 event.trace_id = trace_id 552 if span_id: 553 event.span_id = span_id 554 if audit_manager: 555 audit_manager.record_event(event) 556 return event
Record successful identity authentication event to audit manager.
Args: audit_manager: Target audit manager. subject_id: Subject identifier. tenant_id: Tenant identifier. action: Action identifier. resource: Resource identifier. department: Department ID. roles: List of granted roles. clearance_level: Clearance rank. trace_id: OpenTelemetry trace ID. span_id: OpenTelemetry span ID.
Returns: AuditEvent: Constructed and logged audit event.
class
SecurityContextHolder:
564class SecurityContextHolder: 565 """Thread-local / async ContextVar holder for propagating active SecurityIdentity context.""" 566 567 @staticmethod 568 def set_identity(identity: Optional[SecurityIdentity]) -> contextvars.Token: 569 """Bind SecurityIdentity to the current async task context. 570 571 Args: 572 identity: Active security identity to set. 573 574 Returns: 575 contextvars.Token: Context reset token. 576 """ 577 return _SECURITY_IDENTITY_CONTEXT_VAR.set(identity) 578 579 @staticmethod 580 def get_identity() -> Optional[SecurityIdentity]: 581 """Retrieve SecurityIdentity bound to current async task context. 582 583 Returns: 584 Optional[SecurityIdentity]: Active security identity or None if unbound. 585 """ 586 return _SECURITY_IDENTITY_CONTEXT_VAR.get() 587 588 @staticmethod 589 def reset_identity(token: contextvars.Token) -> None: 590 """Reset SecurityIdentity in current context using a previously returned Token. 591 592 Args: 593 token: ContextVar token returned from set_identity call. 594 """ 595 _SECURITY_IDENTITY_CONTEXT_VAR.reset(token)
Thread-local / async ContextVar holder for propagating active SecurityIdentity context.
567 @staticmethod 568 def set_identity(identity: Optional[SecurityIdentity]) -> contextvars.Token: 569 """Bind SecurityIdentity to the current async task context. 570 571 Args: 572 identity: Active security identity to set. 573 574 Returns: 575 contextvars.Token: Context reset token. 576 """ 577 return _SECURITY_IDENTITY_CONTEXT_VAR.set(identity)
Bind SecurityIdentity to the current async task context.
Args: identity: Active security identity to set.
Returns: contextvars.Token: Context reset token.
579 @staticmethod 580 def get_identity() -> Optional[SecurityIdentity]: 581 """Retrieve SecurityIdentity bound to current async task context. 582 583 Returns: 584 Optional[SecurityIdentity]: Active security identity or None if unbound. 585 """ 586 return _SECURITY_IDENTITY_CONTEXT_VAR.get()
Retrieve SecurityIdentity bound to current async task context.
Returns: Optional[SecurityIdentity]: Active security identity or None if unbound.
@staticmethod
def
reset_identity(token: _contextvars.Token) -> None:
588 @staticmethod 589 def reset_identity(token: contextvars.Token) -> None: 590 """Reset SecurityIdentity in current context using a previously returned Token. 591 592 Args: 593 token: ContextVar token returned from set_identity call. 594 """ 595 _SECURITY_IDENTITY_CONTEXT_VAR.reset(token)
Reset SecurityIdentity in current context using a previously returned Token.
Args: token: ContextVar token returned from set_identity call.
class
EntityAttributeResolverSPI(abc.ABC):
15class EntityAttributeResolverSPI(ABC): 16 """Hydrates dynamic entity attributes for CEL parameter constraint validation.""" 17 18 @abstractmethod 19 def resolve(self, entity_type: str, entity_id: str) -> Dict[str, Any]: 20 """Returns entity attribute dictionary for given entity type and ID. 21 22 Args: 23 entity_type: Entity type category identifier. 24 entity_id: Unique entity instance ID. 25 26 Returns: 27 Dict[str, Any]: Entity attribute dictionary mapping. 28 """ 29 pass
Hydrates dynamic entity attributes for CEL parameter constraint validation.
@abstractmethod
def
resolve(self, entity_type: str, entity_id: str) -> Dict[str, Any]:
18 @abstractmethod 19 def resolve(self, entity_type: str, entity_id: str) -> Dict[str, Any]: 20 """Returns entity attribute dictionary for given entity type and ID. 21 22 Args: 23 entity_type: Entity type category identifier. 24 entity_id: Unique entity instance ID. 25 26 Returns: 27 Dict[str, Any]: Entity attribute dictionary mapping. 28 """ 29 pass
Returns entity attribute dictionary for given entity type and ID.
Args: entity_type: Entity type category identifier. entity_id: Unique entity instance ID.
Returns: Dict[str, Any]: Entity attribute dictionary mapping.
class
PolicyEvaluationRequest:
53class PolicyEvaluationRequest: 54 """Four-part policy evaluation request envelope containing Subject, Action, Resource, and Context. 55 56 Attributes: 57 subject (SecurityIdentity): Caller security identity envelope. 58 action (str): Target action identifier (e.g., 'tool:execute', 'agent:invoke'). 59 resource (Dict[str, Any]): Map of resource attributes and target metadata. 60 context (Dict[str, Any]): Dynamic execution environment context dictionary. 61 """ 62 63 def __init__( 64 self, 65 subject: SecurityIdentity, 66 action: str, 67 resource: Dict[str, Any], 68 context: Dict[str, Any], 69 ): 70 """Initialize PolicyEvaluationRequest. 71 72 Args: 73 subject: Caller security identity. 74 action: Action string identifier. 75 resource: Resource attributes map. 76 context: Execution context dictionary. 77 """ 78 self._subject = subject 79 self._action = action 80 self._resource = dict(resource) 81 self._context = dict(context) 82 83 @classmethod 84 def create( 85 cls, 86 subject: SecurityIdentity, 87 action: str, 88 resource: Dict[str, Any], 89 *args: Any, 90 **kwargs: Any, 91 ): 92 """Factory method to construct a PolicyEvaluationRequest. 93 94 Args: 95 subject: Caller security identity envelope. 96 action: Action identifier string. 97 resource: Resource attributes dictionary. 98 *args: Optional positional context dictionary. 99 **kwargs: Optional keyword context dictionary. 100 101 Returns: 102 PolicyEvaluationRequest: Constructed request instance. 103 """ 104 context: Optional[Dict[str, Any]] = args[0] if len(args) > 0 else kwargs.get("context") 105 return cls( 106 subject=subject, 107 action=action, 108 resource=resource, 109 context=context if isinstance(context, dict) else {}, 110 ) 111 112 @property 113 def subject(self) -> SecurityIdentity: 114 """Get the caller SecurityIdentity.""" 115 return self._subject 116 117 @property 118 def action(self) -> str: 119 """Get the requested action identifier.""" 120 return self._action 121 122 @property 123 def resource(self) -> Dict[str, Any]: 124 """Get copy of target resource attributes dictionary.""" 125 return dict(self._resource) 126 127 @property 128 def context(self) -> Dict[str, Any]: 129 """Get copy of execution environment context dictionary.""" 130 return dict(self._context)
Four-part policy evaluation request envelope containing Subject, Action, Resource, and Context.
Attributes: subject (SecurityIdentity): Caller security identity envelope. action (str): Target action identifier (e.g., 'tool:execute', 'agent:invoke'). resource (Dict[str, Any]): Map of resource attributes and target metadata. context (Dict[str, Any]): Dynamic execution environment context dictionary.
PolicyEvaluationRequest( subject: SecurityIdentity, action: str, resource: Dict[str, Any], context: Dict[str, Any])
63 def __init__( 64 self, 65 subject: SecurityIdentity, 66 action: str, 67 resource: Dict[str, Any], 68 context: Dict[str, Any], 69 ): 70 """Initialize PolicyEvaluationRequest. 71 72 Args: 73 subject: Caller security identity. 74 action: Action string identifier. 75 resource: Resource attributes map. 76 context: Execution context dictionary. 77 """ 78 self._subject = subject 79 self._action = action 80 self._resource = dict(resource) 81 self._context = dict(context)
Initialize PolicyEvaluationRequest.
Args: subject: Caller security identity. action: Action string identifier. resource: Resource attributes map. context: Execution context dictionary.
@classmethod
def
create( cls, subject: SecurityIdentity, action: str, resource: Dict[str, Any], *args: Any, **kwargs: Any):
83 @classmethod 84 def create( 85 cls, 86 subject: SecurityIdentity, 87 action: str, 88 resource: Dict[str, Any], 89 *args: Any, 90 **kwargs: Any, 91 ): 92 """Factory method to construct a PolicyEvaluationRequest. 93 94 Args: 95 subject: Caller security identity envelope. 96 action: Action identifier string. 97 resource: Resource attributes dictionary. 98 *args: Optional positional context dictionary. 99 **kwargs: Optional keyword context dictionary. 100 101 Returns: 102 PolicyEvaluationRequest: Constructed request instance. 103 """ 104 context: Optional[Dict[str, Any]] = args[0] if len(args) > 0 else kwargs.get("context") 105 return cls( 106 subject=subject, 107 action=action, 108 resource=resource, 109 context=context if isinstance(context, dict) else {}, 110 )
Factory method to construct a PolicyEvaluationRequest.
Args: subject: Caller security identity envelope. action: Action identifier string. resource: Resource attributes dictionary. *args: Optional positional context dictionary. **kwargs: Optional keyword context dictionary.
Returns: PolicyEvaluationRequest: Constructed request instance.
subject: SecurityIdentity
112 @property 113 def subject(self) -> SecurityIdentity: 114 """Get the caller SecurityIdentity.""" 115 return self._subject
Get the caller SecurityIdentity.
action: str
117 @property 118 def action(self) -> str: 119 """Get the requested action identifier.""" 120 return self._action
Get the requested action identifier.
class
PolicyDecision:
138class PolicyDecision: 139 """Standardized decision outcome object returned by PDP evaluation. 140 141 Attributes: 142 decision (str): Decision verdict string ('PERMIT', 'DENY', 'NOT_APPLICABLE', 'INDETERMINATE'). 143 diagnostic_reason (str): Reason code or message describing decision rationale. 144 obligations (Dict[str, Any]): Dictionary of enforcement obligations required by PEP. 145 """ 146 147 def __init__( 148 self, 149 decision: str, 150 diagnostic_reason: str, 151 obligations: Dict[str, Any], 152 ): 153 """Initialize PolicyDecision. 154 155 Args: 156 decision: Verdict status string. 157 diagnostic_reason: Rationale description or reason code. 158 obligations: Dictionary of obligation parameters. 159 """ 160 self._decision = decision 161 self._diagnostic_reason = diagnostic_reason 162 self._obligations = dict(obligations) 163 164 @classmethod 165 def create(cls, decision: str, diagnostic_reason: str, obligations: Dict[str, Any]): 166 """Factory method creating a PolicyDecision. 167 168 Args: 169 decision: Decision status string. 170 diagnostic_reason: Reason code. 171 obligations: Obligations dictionary. 172 173 Returns: 174 PolicyDecision: New decision instance. 175 """ 176 return cls(decision=decision, diagnostic_reason=diagnostic_reason, obligations=obligations) 177 178 @classmethod 179 def permit(cls, diagnostic_reason: str, obligations: Dict[str, Any]): 180 """Factory method creating a PERMIT decision. 181 182 Args: 183 diagnostic_reason: Authorization reason description. 184 obligations: Enforcement obligations dictionary. 185 186 Returns: 187 PolicyDecision: PERMIT decision object. 188 """ 189 return cls(decision=DECISION_PERMIT, diagnostic_reason=diagnostic_reason, obligations=obligations) 190 191 @classmethod 192 def deny(cls, diagnostic_reason: str, obligations: Dict[str, Any]): 193 """Factory method creating a DENY decision. 194 195 Args: 196 diagnostic_reason: Denial reason description. 197 obligations: Enforcement obligations dictionary (e.g., handling strategy). 198 199 Returns: 200 PolicyDecision: DENY decision object. 201 """ 202 return cls(decision=DECISION_DENY, diagnostic_reason=diagnostic_reason, obligations=obligations) 203 204 @classmethod 205 def not_applicable(cls, diagnostic_reason: str): 206 """Factory method creating a NOT_APPLICABLE decision. 207 208 Args: 209 diagnostic_reason: Rationale description. 210 211 Returns: 212 PolicyDecision: NOT_APPLICABLE decision object. 213 """ 214 return cls(decision=DECISION_NOT_APPLICABLE, diagnostic_reason=diagnostic_reason, obligations={}) 215 216 @classmethod 217 def indeterminate(cls, diagnostic_reason: str): 218 """Factory method creating an INDETERMINATE decision. 219 220 Args: 221 diagnostic_reason: Error rationale description. 222 223 Returns: 224 PolicyDecision: INDETERMINATE decision object. 225 """ 226 return cls(decision=DECISION_INDETERMINATE, diagnostic_reason=diagnostic_reason, obligations={}) 227 228 @property 229 def decision(self) -> str: 230 """Get raw decision verdict string.""" 231 return self._decision 232 233 @property 234 def decision_enum(self) -> DecisionEnum: 235 """Get decision verdict as a typed DecisionEnum.""" 236 return DecisionEnum(self._decision) 237 238 @property 239 def diagnostic_reason(self) -> str: 240 """Get diagnostic reason string.""" 241 return self._diagnostic_reason 242 243 @property 244 def obligations(self) -> Dict[str, Any]: 245 """Get copy of obligations dictionary.""" 246 return dict(self._obligations) 247 248 @property 249 def obligations_list(self) -> List[str]: 250 """Get obligations formatted as string list for legacy compatibility. 251 252 Returns: 253 List[str]: Formatted obligation string list. 254 """ 255 res: List[str] = [] 256 strategy = self._obligations.get("strategy") 257 if strategy: 258 res.append(str(strategy)) 259 for k, v in self._obligations.items(): 260 if k != "strategy": 261 res.append(f"{k}:{v}") 262 return res 263 264 @property 265 def is_permitted(self) -> bool: 266 """Check if decision is PERMIT. 267 268 Returns: 269 bool: True if decision is PERMIT, False otherwise. 270 """ 271 return self._decision == DECISION_PERMIT
Standardized decision outcome object returned by PDP evaluation.
Attributes: decision (str): Decision verdict string ('PERMIT', 'DENY', 'NOT_APPLICABLE', 'INDETERMINATE'). diagnostic_reason (str): Reason code or message describing decision rationale. obligations (Dict[str, Any]): Dictionary of enforcement obligations required by PEP.
PolicyDecision(decision: str, diagnostic_reason: str, obligations: Dict[str, Any])
147 def __init__( 148 self, 149 decision: str, 150 diagnostic_reason: str, 151 obligations: Dict[str, Any], 152 ): 153 """Initialize PolicyDecision. 154 155 Args: 156 decision: Verdict status string. 157 diagnostic_reason: Rationale description or reason code. 158 obligations: Dictionary of obligation parameters. 159 """ 160 self._decision = decision 161 self._diagnostic_reason = diagnostic_reason 162 self._obligations = dict(obligations)
Initialize PolicyDecision.
Args: decision: Verdict status string. diagnostic_reason: Rationale description or reason code. obligations: Dictionary of obligation parameters.
@classmethod
def
create( cls, decision: str, diagnostic_reason: str, obligations: Dict[str, Any]):
164 @classmethod 165 def create(cls, decision: str, diagnostic_reason: str, obligations: Dict[str, Any]): 166 """Factory method creating a PolicyDecision. 167 168 Args: 169 decision: Decision status string. 170 diagnostic_reason: Reason code. 171 obligations: Obligations dictionary. 172 173 Returns: 174 PolicyDecision: New decision instance. 175 """ 176 return cls(decision=decision, diagnostic_reason=diagnostic_reason, obligations=obligations)
Factory method creating a PolicyDecision.
Args: decision: Decision status string. diagnostic_reason: Reason code. obligations: Obligations dictionary.
Returns: PolicyDecision: New decision instance.
@classmethod
def
permit(cls, diagnostic_reason: str, obligations: Dict[str, Any]):
178 @classmethod 179 def permit(cls, diagnostic_reason: str, obligations: Dict[str, Any]): 180 """Factory method creating a PERMIT decision. 181 182 Args: 183 diagnostic_reason: Authorization reason description. 184 obligations: Enforcement obligations dictionary. 185 186 Returns: 187 PolicyDecision: PERMIT decision object. 188 """ 189 return cls(decision=DECISION_PERMIT, diagnostic_reason=diagnostic_reason, obligations=obligations)
Factory method creating a PERMIT decision.
Args: diagnostic_reason: Authorization reason description. obligations: Enforcement obligations dictionary.
Returns: PolicyDecision: PERMIT decision object.
@classmethod
def
deny(cls, diagnostic_reason: str, obligations: Dict[str, Any]):
191 @classmethod 192 def deny(cls, diagnostic_reason: str, obligations: Dict[str, Any]): 193 """Factory method creating a DENY decision. 194 195 Args: 196 diagnostic_reason: Denial reason description. 197 obligations: Enforcement obligations dictionary (e.g., handling strategy). 198 199 Returns: 200 PolicyDecision: DENY decision object. 201 """ 202 return cls(decision=DECISION_DENY, diagnostic_reason=diagnostic_reason, obligations=obligations)
Factory method creating a DENY decision.
Args: diagnostic_reason: Denial reason description. obligations: Enforcement obligations dictionary (e.g., handling strategy).
Returns: PolicyDecision: DENY decision object.
@classmethod
def
not_applicable(cls, diagnostic_reason: str):
204 @classmethod 205 def not_applicable(cls, diagnostic_reason: str): 206 """Factory method creating a NOT_APPLICABLE decision. 207 208 Args: 209 diagnostic_reason: Rationale description. 210 211 Returns: 212 PolicyDecision: NOT_APPLICABLE decision object. 213 """ 214 return cls(decision=DECISION_NOT_APPLICABLE, diagnostic_reason=diagnostic_reason, obligations={})
Factory method creating a NOT_APPLICABLE decision.
Args: diagnostic_reason: Rationale description.
Returns: PolicyDecision: NOT_APPLICABLE decision object.
@classmethod
def
indeterminate(cls, diagnostic_reason: str):
216 @classmethod 217 def indeterminate(cls, diagnostic_reason: str): 218 """Factory method creating an INDETERMINATE decision. 219 220 Args: 221 diagnostic_reason: Error rationale description. 222 223 Returns: 224 PolicyDecision: INDETERMINATE decision object. 225 """ 226 return cls(decision=DECISION_INDETERMINATE, diagnostic_reason=diagnostic_reason, obligations={})
Factory method creating an INDETERMINATE decision.
Args: diagnostic_reason: Error rationale description.
Returns: PolicyDecision: INDETERMINATE decision object.
decision: str
228 @property 229 def decision(self) -> str: 230 """Get raw decision verdict string.""" 231 return self._decision
Get raw decision verdict string.
decision_enum: langchain_security_extension.security_core.pdp.DecisionEnum
233 @property 234 def decision_enum(self) -> DecisionEnum: 235 """Get decision verdict as a typed DecisionEnum.""" 236 return DecisionEnum(self._decision)
Get decision verdict as a typed DecisionEnum.
diagnostic_reason: str
238 @property 239 def diagnostic_reason(self) -> str: 240 """Get diagnostic reason string.""" 241 return self._diagnostic_reason
Get diagnostic reason string.
obligations: Dict[str, Any]
243 @property 244 def obligations(self) -> Dict[str, Any]: 245 """Get copy of obligations dictionary.""" 246 return dict(self._obligations)
Get copy of obligations dictionary.
obligations_list: List[str]
248 @property 249 def obligations_list(self) -> List[str]: 250 """Get obligations formatted as string list for legacy compatibility. 251 252 Returns: 253 List[str]: Formatted obligation string list. 254 """ 255 res: List[str] = [] 256 strategy = self._obligations.get("strategy") 257 if strategy: 258 res.append(str(strategy)) 259 for k, v in self._obligations.items(): 260 if k != "strategy": 261 res.append(f"{k}:{v}") 262 return res
Get obligations formatted as string list for legacy compatibility.
Returns: List[str]: Formatted obligation string list.
is_permitted: bool
264 @property 265 def is_permitted(self) -> bool: 266 """Check if decision is PERMIT. 267 268 Returns: 269 bool: True if decision is PERMIT, False otherwise. 270 """ 271 return self._decision == DECISION_PERMIT
Check if decision is PERMIT.
Returns: bool: True if decision is PERMIT, False otherwise.
class
PolicyDecisionEngine:
414class PolicyDecisionEngine: 415 """Abstract SPI interface for Policy Decision Point engines.""" 416 417 def evaluate(self, request: PolicyEvaluationRequest) -> PolicyDecision: 418 """Evaluate a policy decision request. 419 420 Args: 421 request: Evaluation request envelope. 422 423 Returns: 424 PolicyDecision: Computed policy decision outcome. 425 426 Raises: 427 NotImplementedError: Must be implemented by concrete PDP engines. 428 """ 429 raise NotImplementedError("Subclasses must implement evaluate")
Abstract SPI interface for Policy Decision Point engines.
417 def evaluate(self, request: PolicyEvaluationRequest) -> PolicyDecision: 418 """Evaluate a policy decision request. 419 420 Args: 421 request: Evaluation request envelope. 422 423 Returns: 424 PolicyDecision: Computed policy decision outcome. 425 426 Raises: 427 NotImplementedError: Must be implemented by concrete PDP engines. 428 """ 429 raise NotImplementedError("Subclasses must implement evaluate")
Evaluate a policy decision request.
Args: request: Evaluation request envelope.
Returns: PolicyDecision: Computed policy decision outcome.
Raises: NotImplementedError: Must be implemented by concrete PDP engines.
593class EmbeddedPolicyEngine(PolicyDecisionEngine): 594 """In-memory PDP evaluating tenant, clearance, roles, and CEL constraints with Deny-Overrides rules.""" 595 596 def __init__( 597 self, 598 role_graph: RoleGraph, 599 cel_evaluator: CELEvaluator, 600 audit_manager: AuditManager, 601 *args: Any, 602 **kwargs: Any, 603 ): 604 """Initialize EmbeddedPolicyEngine. 605 606 Args: 607 role_graph: RoleGraph instance for role inheritance expansion. 608 cel_evaluator: CELEvaluator instance for parameter expression validation. 609 audit_manager: AuditManager instance for security event logging. 610 *args: Optional positional arguments (rules, decision_cache). 611 **kwargs: Optional keyword arguments (rules, decision_cache). 612 """ 613 rules = kwargs.get("rules", args[0] if len(args) > 0 else None) 614 decision_cache = kwargs.get("decision_cache", args[1] if len(args) > 1 else None) 615 self._role_graph = role_graph 616 self._cel_evaluator = cel_evaluator 617 self._audit_manager = audit_manager 618 self._rules = list(rules) if rules is not None else [] 619 self._cache = decision_cache if isinstance(decision_cache, TwoTierDecisionCache) else TwoTierDecisionCache.create() 620 self.enforce_strict_tenant_isolation = True 621 self._invalidation_callbacks: List[Any] = [] 622 623 @classmethod 624 def create(cls): 625 """Factory method constructing default EmbeddedPolicyEngine. 626 627 Returns: 628 EmbeddedPolicyEngine: Engine initialized with default components. 629 """ 630 return cls( 631 role_graph=RoleGraph.create(), 632 cel_evaluator=CELEvaluator.create(), 633 audit_manager=AuditManager.create(publishers=[]), 634 ) 635 636 @classmethod 637 def with_components( 638 cls, 639 role_graph: RoleGraph, 640 cel_evaluator: CELEvaluator, 641 audit_manager: AuditManager, 642 *args: Any, 643 **kwargs: Any, 644 ): 645 """Factory method constructing EmbeddedPolicyEngine with specified components. 646 647 Args: 648 role_graph: Custom RoleGraph. 649 cel_evaluator: Custom CELEvaluator. 650 audit_manager: Custom AuditManager. 651 *args: Optional positional arguments (rules, decision_cache). 652 **kwargs: Optional keyword arguments (rules, decision_cache). 653 654 Returns: 655 EmbeddedPolicyEngine: Engine instance. 656 """ 657 return cls( 658 role_graph, 659 cel_evaluator, 660 audit_manager, 661 *args, 662 **kwargs, 663 ) 664 665 @property 666 def role_graph(self) -> RoleGraph: 667 """Get the configured RoleGraph.""" 668 return self._role_graph 669 670 @property 671 def audit_manager(self) -> AuditManager: 672 """Get the configured AuditManager.""" 673 return self._audit_manager 674 675 @property 676 def decision_cache(self) -> TwoTierDecisionCache: 677 """Get the configured TwoTierDecisionCache.""" 678 return self._cache 679 680 def register_invalidation_callback(self, callback: Any) -> None: 681 """Register a callback to be invoked upon cache eviction or subject invalidation. 682 683 Args: 684 callback: Callable invoked with subject_id on subject eviction, or with no arguments on full cache clear. 685 """ 686 self._invalidation_callbacks.append(callback) 687 688 def clear_cache(self) -> None: 689 """Flush all Tier 2 static cache entries and trigger registered invalidation callbacks.""" 690 self._cache.flush() 691 for cb in list(getattr(self, "_invalidation_callbacks", [])): 692 try: 693 cb() 694 except Exception: 695 pass 696 697 def evict_subject_cache(self, subject_id: str, *args: Any, **kwargs: Any) -> int: 698 """Evict cached decisions for a subject upon role revocation and emit an audit event. 699 700 Args: 701 subject_id: Subject identifier whose cached decisions are invalidated. 702 *args: Optional positional config dict. 703 **kwargs: Optional keyword config dict. 704 705 Returns: 706 int: Count of evicted entries. 707 """ 708 evicted_count = self._cache.evict_subject(subject_id, *args, **kwargs) 709 for cb in list(getattr(self, "_invalidation_callbacks", [])): 710 try: 711 cb(subject_id) 712 except Exception: 713 pass 714 event = AuditEvent.create( 715 subject_id=subject_id, 716 tenant_id=str(kwargs.get("tenant_id", "DEFAULT")), 717 action="pdp:cache_evict", 718 resource="TwoTierDecisionCache", 719 decision=DECISION_PERMIT, 720 diagnostic_reason=f"{EVENT_CACHE_EVICTED}: Evicted {evicted_count} cached decisions for {subject_id}", 721 severity=SEVERITY_INFORMATIONAL, 722 metadata={"evicted_count": evicted_count}, 723 reason_code=EVENT_CACHE_EVICTED, 724 ) 725 self._audit_manager.record_event(event) 726 return evicted_count 727 728 def add_rule(self, rule: PolicyRule) -> None: 729 """Add a PolicyRule to engine rule store. 730 731 Args: 732 rule: PolicyRule instance. 733 """ 734 self._rules.append(rule) 735 736 def evaluate(self, request: PolicyEvaluationRequest) -> PolicyDecision: 737 """Evaluate access request through tenant isolation, quarantine, clearance lattice, roles, and CEL constraints. 738 739 Args: 740 request: Evaluation request envelope. 741 742 Returns: 743 PolicyDecision: Evaluation decision status and obligations. 744 """ 745 subject = request.subject 746 resource = request.resource 747 req_config = request.context.get("config") if isinstance(request.context, dict) else None 748 749 # Check Tier 1 Invocation Cache and Tier 2 Static Cache 750 cache_key = self._cache.compute_request_hash(request) 751 tier1_decision = self._cache.get_invocation_cache(req_config, cache_key) 752 if tier1_decision: 753 if req_config and isinstance(req_config, dict) and isinstance(req_config.get("configurable"), dict): 754 req_config["configurable"]["sec:pdp_cache_hit"] = True 755 req_config["configurable"]["sec:pdp_cache_tier"] = "tier1" 756 return tier1_decision 757 cached_decision = self._cache.get_static(cache_key) 758 if cached_decision: 759 if req_config and isinstance(req_config, dict) and isinstance(req_config.get("configurable"), dict): 760 req_config["configurable"]["sec:pdp_cache_hit"] = True 761 req_config["configurable"]["sec:pdp_cache_tier"] = "tier2" 762 return cached_decision 763 764 # 0. Quarantine Check 765 is_quarantined = bool(resource.get("sec:quarantined") or resource.get("quarantined")) 766 if is_quarantined: 767 expanded_roles = self._role_graph.expand_roles(subject.roles) 768 if "ROLE_ADMIN" not in expanded_roles or subject.clearance_level < 3: 769 decision = self._record_and_return( 770 request=request, 771 decision_status=DECISION_DENY, 772 diagnostic_reason=REASON_QUARANTINED_CHUNK, 773 severity=SEVERITY_WARNING, 774 obligations={}, 775 reason_code=REASON_QUARANTINED_CHUNK, 776 ) 777 self._cache.put_static(cache_key, decision) 778 return decision 779 780 # 0.1 Unauthenticated Subject Check (Fail-closed for anonymous unless explicitly permitted by rule) 781 is_anon = ( 782 subject is None 783 or subject.is_anonymous 784 or subject.subject_id is None 785 or str(subject.subject_id).upper() == ANONYMOUS_SUBJECT_ID 786 ) 787 if is_anon: 788 anon_permitted = False 789 if self._rules: 790 for rule in self._rules: 791 if (rule.target_action == "*" or rule.target_action == request.action) and rule.effect == DECISION_PERMIT: 792 anon_permitted = True 793 break 794 if not anon_permitted: 795 decision = self._record_and_return( 796 request=request, 797 decision_status=DECISION_DENY, 798 diagnostic_reason=REASON_UNAUTHENTICATED, 799 severity=SEVERITY_WARNING, 800 obligations={}, 801 reason_code=REASON_UNAUTHENTICATED, 802 ) 803 self._cache.put_static(cache_key, decision) 804 return decision 805 806 # 1. Multi-Tenancy Check 807 required_tenant = resource.get("tenant_id") 808 if required_tenant: 809 tenant_mismatch = (subject.tenant_id not in required_tenant) if isinstance(required_tenant, (set, list, tuple)) else (required_tenant != subject.tenant_id) 810 if tenant_mismatch: 811 reason_code = EVENT_TENANT_BOUND_VIOLATION if self.enforce_strict_tenant_isolation else None 812 severity = SEVERITY_CRITICAL_ALERT if self.enforce_strict_tenant_isolation else SEVERITY_WARNING 813 decision = self._record_and_return( 814 request=request, 815 decision_status=DECISION_DENY, 816 diagnostic_reason=REASON_TENANT_MISMATCH, 817 severity=severity, 818 obligations={}, 819 reason_code=reason_code, 820 ) 821 self._cache.put_static(cache_key, decision) 822 return decision 823 824 # 2. Clearance Lattice Check (Supports numeric rank or string classification level) 825 req_clr_val = resource.get("min_clearance") or resource.get("clearance_floor") or resource.get("sec:clearance_level") 826 if req_clr_val is not None: 827 req_rank = CLASSIFICATION_TO_RANK.get(str(req_clr_val).upper(), int(req_clr_val) if str(req_clr_val).isdigit() else 0) 828 subj_rank = subject.clearance_level 829 if subj_rank < req_rank: 830 decision = self._record_and_return( 831 request=request, 832 decision_status=DECISION_DENY, 833 diagnostic_reason=REASON_INSUFFICIENT_CLEARANCE, 834 severity=SEVERITY_WARNING, 835 obligations={}, 836 reason_code=None, 837 ) 838 self._cache.put_static(cache_key, decision) 839 return decision 840 841 # 3. Role Inheritance Matching 842 required_roles = resource.get("required_roles") or resource.get("sec:required_roles") or resource.get("sec:allowed_roles") 843 if required_roles: 844 expanded_roles = self._role_graph.expand_roles(subject.roles) 845 if not set(required_roles).intersection(expanded_roles): 846 decision = self._record_and_return( 847 request=request, 848 decision_status=DECISION_DENY, 849 diagnostic_reason=REASON_MISSING_REQUIRED_ROLE, 850 severity=SEVERITY_WARNING, 851 obligations={}, 852 reason_code=None, 853 ) 854 self._cache.put_static(cache_key, decision) 855 return decision 856 857 # 4. CEL Parameter Expression Evaluation 858 cel_expression = resource.get("cel_expression") 859 if cel_expression: 860 runtime_args = resource.get("arguments", {}) 861 tool_meta = resource.get("tool_metadata", {}) 862 passed = self._cel_evaluator.evaluate( 863 expression=cel_expression, 864 subject=subject, 865 args=runtime_args, 866 tool=tool_meta, 867 config=req_config, 868 ) 869 if not passed: 870 strategy = STRATEGY_HARD_ABORT if resource.get("is_mutative") else STRATEGY_RETURN_FEEDBACK 871 diag_reason = REASON_PARAMETER_CONSTRAINT_FAILED 872 if self._cel_evaluator.last_diagnostic_reason: 873 diag_reason = self._cel_evaluator.last_diagnostic_reason 874 elif "entities." in str(cel_expression): 875 diag_reason = "CEL_HYDRATION_FAILED (hydrated owner mismatch)" 876 decision = self._record_and_return( 877 request=request, 878 decision_status=DECISION_DENY, 879 diagnostic_reason=diag_reason, 880 severity=SEVERITY_SECURITY_ALERT, 881 obligations={"strategy": strategy}, 882 reason_code=None, 883 ) 884 self._cache.put_static(cache_key, decision) 885 return decision 886 887 # 5. Declarative Deny-Overrides Evaluation (if custom rules are defined) 888 if self._rules: 889 matched_permit = False 890 for rule in self._rules: 891 if rule.target_action == "*" or rule.target_action == request.action: 892 # Validate rule constraints: roles, clearance, cel_condition 893 rule_matches = True 894 895 if rule.required_roles: 896 expanded_roles = self._role_graph.expand_roles(subject.roles) 897 if not set(rule.required_roles).intersection(expanded_roles): 898 rule_matches = False 899 900 if rule_matches and rule.min_clearance is not None: 901 if subject.clearance_level < int(rule.min_clearance): 902 rule_matches = False 903 904 if rule_matches and rule.cel_condition: 905 runtime_args = resource.get("arguments", {}) 906 tool_meta = resource.get("tool_metadata", {}) 907 if not self._cel_evaluator.evaluate( 908 expression=rule.cel_condition, 909 subject=subject, 910 args=runtime_args, 911 tool=tool_meta, 912 config=req_config, 913 ): 914 rule_matches = False 915 916 if rule_matches: 917 if rule.effect == DECISION_DENY: 918 decision = self._record_and_return( 919 request=request, 920 decision_status=DECISION_DENY, 921 diagnostic_reason=f"Rule {rule.rule_id} DENY", 922 severity=SEVERITY_SECURITY_ALERT, 923 obligations={"strategy": rule.on_denial}, 924 reason_code=None, 925 ) 926 self._cache.put_static(cache_key, decision) 927 return decision 928 elif rule.effect == DECISION_PERMIT: 929 matched_permit = True 930 931 if not matched_permit: 932 decision = self._record_and_return( 933 request=request, 934 decision_status=DECISION_NOT_APPLICABLE, 935 diagnostic_reason="No matching permit policy rule found", 936 severity=SEVERITY_WARNING, 937 obligations={}, 938 reason_code=None, 939 ) 940 self._cache.put_static(cache_key, decision) 941 return decision 942 943 # 6. Permit Default 944 decision = self._record_and_return( 945 request=request, 946 decision_status=DECISION_PERMIT, 947 diagnostic_reason=REASON_AUTHORIZED, 948 severity=SEVERITY_INFORMATIONAL, 949 obligations={}, 950 reason_code=None, 951 ) 952 self._cache.put_static(cache_key, decision) 953 return decision 954 955 def _record_and_return( 956 self, 957 request: PolicyEvaluationRequest, 958 decision_status: str, 959 diagnostic_reason: str, 960 severity: str, 961 obligations: Dict[str, Any], 962 reason_code: Optional[str], 963 ) -> PolicyDecision: 964 resource_id = str(request.resource.get("name") or request.resource.get("target_id") or "UNKNOWN") 965 event = AuditEvent.create( 966 subject_id=request.subject.subject_id, 967 tenant_id=request.subject.tenant_id, 968 action=request.action, 969 resource=resource_id, 970 decision=decision_status, 971 diagnostic_reason=diagnostic_reason, 972 severity=severity, 973 metadata={"obligations": obligations}, 974 reason_code=reason_code, 975 ) 976 if reason_code == EVENT_TENANT_BOUND_VIOLATION: 977 event.threat_category = "IDENTITY_SPOOFING" 978 event.mitigation_action = "BLOCKED" 979 self._audit_manager.record_event(event) 980 981 dec = PolicyDecision.create( 982 decision=decision_status, 983 diagnostic_reason=diagnostic_reason, 984 obligations=obligations, 985 ) 986 req_config = request.context.get("config") if isinstance(request.context, dict) else None 987 if req_config: 988 cache_key = self._cache.compute_request_hash(request) 989 self._cache.put_invocation_cache(req_config, cache_key, dec) 990 return dec
In-memory PDP evaluating tenant, clearance, roles, and CEL constraints with Deny-Overrides rules.
EmbeddedPolicyEngine( role_graph: RoleGraph, cel_evaluator: CELEvaluator, audit_manager: AuditManager, *args: Any, **kwargs: Any)
596 def __init__( 597 self, 598 role_graph: RoleGraph, 599 cel_evaluator: CELEvaluator, 600 audit_manager: AuditManager, 601 *args: Any, 602 **kwargs: Any, 603 ): 604 """Initialize EmbeddedPolicyEngine. 605 606 Args: 607 role_graph: RoleGraph instance for role inheritance expansion. 608 cel_evaluator: CELEvaluator instance for parameter expression validation. 609 audit_manager: AuditManager instance for security event logging. 610 *args: Optional positional arguments (rules, decision_cache). 611 **kwargs: Optional keyword arguments (rules, decision_cache). 612 """ 613 rules = kwargs.get("rules", args[0] if len(args) > 0 else None) 614 decision_cache = kwargs.get("decision_cache", args[1] if len(args) > 1 else None) 615 self._role_graph = role_graph 616 self._cel_evaluator = cel_evaluator 617 self._audit_manager = audit_manager 618 self._rules = list(rules) if rules is not None else [] 619 self._cache = decision_cache if isinstance(decision_cache, TwoTierDecisionCache) else TwoTierDecisionCache.create() 620 self.enforce_strict_tenant_isolation = True 621 self._invalidation_callbacks: List[Any] = []
Initialize EmbeddedPolicyEngine.
Args: role_graph: RoleGraph instance for role inheritance expansion. cel_evaluator: CELEvaluator instance for parameter expression validation. audit_manager: AuditManager instance for security event logging. *args: Optional positional arguments (rules, decision_cache). **kwargs: Optional keyword arguments (rules, decision_cache).
@classmethod
def
create(cls):
623 @classmethod 624 def create(cls): 625 """Factory method constructing default EmbeddedPolicyEngine. 626 627 Returns: 628 EmbeddedPolicyEngine: Engine initialized with default components. 629 """ 630 return cls( 631 role_graph=RoleGraph.create(), 632 cel_evaluator=CELEvaluator.create(), 633 audit_manager=AuditManager.create(publishers=[]), 634 )
Factory method constructing default EmbeddedPolicyEngine.
Returns: EmbeddedPolicyEngine: Engine initialized with default components.
@classmethod
def
with_components( cls, role_graph: RoleGraph, cel_evaluator: CELEvaluator, audit_manager: AuditManager, *args: Any, **kwargs: Any):
636 @classmethod 637 def with_components( 638 cls, 639 role_graph: RoleGraph, 640 cel_evaluator: CELEvaluator, 641 audit_manager: AuditManager, 642 *args: Any, 643 **kwargs: Any, 644 ): 645 """Factory method constructing EmbeddedPolicyEngine with specified components. 646 647 Args: 648 role_graph: Custom RoleGraph. 649 cel_evaluator: Custom CELEvaluator. 650 audit_manager: Custom AuditManager. 651 *args: Optional positional arguments (rules, decision_cache). 652 **kwargs: Optional keyword arguments (rules, decision_cache). 653 654 Returns: 655 EmbeddedPolicyEngine: Engine instance. 656 """ 657 return cls( 658 role_graph, 659 cel_evaluator, 660 audit_manager, 661 *args, 662 **kwargs, 663 )
Factory method constructing EmbeddedPolicyEngine with specified components.
Args: role_graph: Custom RoleGraph. cel_evaluator: Custom CELEvaluator. audit_manager: Custom AuditManager. *args: Optional positional arguments (rules, decision_cache). **kwargs: Optional keyword arguments (rules, decision_cache).
Returns: EmbeddedPolicyEngine: Engine instance.
role_graph: RoleGraph
665 @property 666 def role_graph(self) -> RoleGraph: 667 """Get the configured RoleGraph.""" 668 return self._role_graph
Get the configured RoleGraph.
audit_manager: AuditManager
670 @property 671 def audit_manager(self) -> AuditManager: 672 """Get the configured AuditManager.""" 673 return self._audit_manager
Get the configured AuditManager.
decision_cache: TwoTierDecisionCache
675 @property 676 def decision_cache(self) -> TwoTierDecisionCache: 677 """Get the configured TwoTierDecisionCache.""" 678 return self._cache
Get the configured TwoTierDecisionCache.
def
register_invalidation_callback(self, callback: Any) -> None:
680 def register_invalidation_callback(self, callback: Any) -> None: 681 """Register a callback to be invoked upon cache eviction or subject invalidation. 682 683 Args: 684 callback: Callable invoked with subject_id on subject eviction, or with no arguments on full cache clear. 685 """ 686 self._invalidation_callbacks.append(callback)
Register a callback to be invoked upon cache eviction or subject invalidation.
Args: callback: Callable invoked with subject_id on subject eviction, or with no arguments on full cache clear.
def
clear_cache(self) -> None:
688 def clear_cache(self) -> None: 689 """Flush all Tier 2 static cache entries and trigger registered invalidation callbacks.""" 690 self._cache.flush() 691 for cb in list(getattr(self, "_invalidation_callbacks", [])): 692 try: 693 cb() 694 except Exception: 695 pass
Flush all Tier 2 static cache entries and trigger registered invalidation callbacks.
def
evict_subject_cache(self, subject_id: str, *args: Any, **kwargs: Any) -> int:
697 def evict_subject_cache(self, subject_id: str, *args: Any, **kwargs: Any) -> int: 698 """Evict cached decisions for a subject upon role revocation and emit an audit event. 699 700 Args: 701 subject_id: Subject identifier whose cached decisions are invalidated. 702 *args: Optional positional config dict. 703 **kwargs: Optional keyword config dict. 704 705 Returns: 706 int: Count of evicted entries. 707 """ 708 evicted_count = self._cache.evict_subject(subject_id, *args, **kwargs) 709 for cb in list(getattr(self, "_invalidation_callbacks", [])): 710 try: 711 cb(subject_id) 712 except Exception: 713 pass 714 event = AuditEvent.create( 715 subject_id=subject_id, 716 tenant_id=str(kwargs.get("tenant_id", "DEFAULT")), 717 action="pdp:cache_evict", 718 resource="TwoTierDecisionCache", 719 decision=DECISION_PERMIT, 720 diagnostic_reason=f"{EVENT_CACHE_EVICTED}: Evicted {evicted_count} cached decisions for {subject_id}", 721 severity=SEVERITY_INFORMATIONAL, 722 metadata={"evicted_count": evicted_count}, 723 reason_code=EVENT_CACHE_EVICTED, 724 ) 725 self._audit_manager.record_event(event) 726 return evicted_count
Evict cached decisions for a subject upon role revocation and emit an audit event.
Args: subject_id: Subject identifier whose cached decisions are invalidated. *args: Optional positional config dict. **kwargs: Optional keyword config dict.
Returns: int: Count of evicted entries.
def
add_rule( self, rule: langchain_security_extension.security_core.pdp.PolicyRule) -> None:
728 def add_rule(self, rule: PolicyRule) -> None: 729 """Add a PolicyRule to engine rule store. 730 731 Args: 732 rule: PolicyRule instance. 733 """ 734 self._rules.append(rule)
Add a PolicyRule to engine rule store.
Args: rule: PolicyRule instance.
736 def evaluate(self, request: PolicyEvaluationRequest) -> PolicyDecision: 737 """Evaluate access request through tenant isolation, quarantine, clearance lattice, roles, and CEL constraints. 738 739 Args: 740 request: Evaluation request envelope. 741 742 Returns: 743 PolicyDecision: Evaluation decision status and obligations. 744 """ 745 subject = request.subject 746 resource = request.resource 747 req_config = request.context.get("config") if isinstance(request.context, dict) else None 748 749 # Check Tier 1 Invocation Cache and Tier 2 Static Cache 750 cache_key = self._cache.compute_request_hash(request) 751 tier1_decision = self._cache.get_invocation_cache(req_config, cache_key) 752 if tier1_decision: 753 if req_config and isinstance(req_config, dict) and isinstance(req_config.get("configurable"), dict): 754 req_config["configurable"]["sec:pdp_cache_hit"] = True 755 req_config["configurable"]["sec:pdp_cache_tier"] = "tier1" 756 return tier1_decision 757 cached_decision = self._cache.get_static(cache_key) 758 if cached_decision: 759 if req_config and isinstance(req_config, dict) and isinstance(req_config.get("configurable"), dict): 760 req_config["configurable"]["sec:pdp_cache_hit"] = True 761 req_config["configurable"]["sec:pdp_cache_tier"] = "tier2" 762 return cached_decision 763 764 # 0. Quarantine Check 765 is_quarantined = bool(resource.get("sec:quarantined") or resource.get("quarantined")) 766 if is_quarantined: 767 expanded_roles = self._role_graph.expand_roles(subject.roles) 768 if "ROLE_ADMIN" not in expanded_roles or subject.clearance_level < 3: 769 decision = self._record_and_return( 770 request=request, 771 decision_status=DECISION_DENY, 772 diagnostic_reason=REASON_QUARANTINED_CHUNK, 773 severity=SEVERITY_WARNING, 774 obligations={}, 775 reason_code=REASON_QUARANTINED_CHUNK, 776 ) 777 self._cache.put_static(cache_key, decision) 778 return decision 779 780 # 0.1 Unauthenticated Subject Check (Fail-closed for anonymous unless explicitly permitted by rule) 781 is_anon = ( 782 subject is None 783 or subject.is_anonymous 784 or subject.subject_id is None 785 or str(subject.subject_id).upper() == ANONYMOUS_SUBJECT_ID 786 ) 787 if is_anon: 788 anon_permitted = False 789 if self._rules: 790 for rule in self._rules: 791 if (rule.target_action == "*" or rule.target_action == request.action) and rule.effect == DECISION_PERMIT: 792 anon_permitted = True 793 break 794 if not anon_permitted: 795 decision = self._record_and_return( 796 request=request, 797 decision_status=DECISION_DENY, 798 diagnostic_reason=REASON_UNAUTHENTICATED, 799 severity=SEVERITY_WARNING, 800 obligations={}, 801 reason_code=REASON_UNAUTHENTICATED, 802 ) 803 self._cache.put_static(cache_key, decision) 804 return decision 805 806 # 1. Multi-Tenancy Check 807 required_tenant = resource.get("tenant_id") 808 if required_tenant: 809 tenant_mismatch = (subject.tenant_id not in required_tenant) if isinstance(required_tenant, (set, list, tuple)) else (required_tenant != subject.tenant_id) 810 if tenant_mismatch: 811 reason_code = EVENT_TENANT_BOUND_VIOLATION if self.enforce_strict_tenant_isolation else None 812 severity = SEVERITY_CRITICAL_ALERT if self.enforce_strict_tenant_isolation else SEVERITY_WARNING 813 decision = self._record_and_return( 814 request=request, 815 decision_status=DECISION_DENY, 816 diagnostic_reason=REASON_TENANT_MISMATCH, 817 severity=severity, 818 obligations={}, 819 reason_code=reason_code, 820 ) 821 self._cache.put_static(cache_key, decision) 822 return decision 823 824 # 2. Clearance Lattice Check (Supports numeric rank or string classification level) 825 req_clr_val = resource.get("min_clearance") or resource.get("clearance_floor") or resource.get("sec:clearance_level") 826 if req_clr_val is not None: 827 req_rank = CLASSIFICATION_TO_RANK.get(str(req_clr_val).upper(), int(req_clr_val) if str(req_clr_val).isdigit() else 0) 828 subj_rank = subject.clearance_level 829 if subj_rank < req_rank: 830 decision = self._record_and_return( 831 request=request, 832 decision_status=DECISION_DENY, 833 diagnostic_reason=REASON_INSUFFICIENT_CLEARANCE, 834 severity=SEVERITY_WARNING, 835 obligations={}, 836 reason_code=None, 837 ) 838 self._cache.put_static(cache_key, decision) 839 return decision 840 841 # 3. Role Inheritance Matching 842 required_roles = resource.get("required_roles") or resource.get("sec:required_roles") or resource.get("sec:allowed_roles") 843 if required_roles: 844 expanded_roles = self._role_graph.expand_roles(subject.roles) 845 if not set(required_roles).intersection(expanded_roles): 846 decision = self._record_and_return( 847 request=request, 848 decision_status=DECISION_DENY, 849 diagnostic_reason=REASON_MISSING_REQUIRED_ROLE, 850 severity=SEVERITY_WARNING, 851 obligations={}, 852 reason_code=None, 853 ) 854 self._cache.put_static(cache_key, decision) 855 return decision 856 857 # 4. CEL Parameter Expression Evaluation 858 cel_expression = resource.get("cel_expression") 859 if cel_expression: 860 runtime_args = resource.get("arguments", {}) 861 tool_meta = resource.get("tool_metadata", {}) 862 passed = self._cel_evaluator.evaluate( 863 expression=cel_expression, 864 subject=subject, 865 args=runtime_args, 866 tool=tool_meta, 867 config=req_config, 868 ) 869 if not passed: 870 strategy = STRATEGY_HARD_ABORT if resource.get("is_mutative") else STRATEGY_RETURN_FEEDBACK 871 diag_reason = REASON_PARAMETER_CONSTRAINT_FAILED 872 if self._cel_evaluator.last_diagnostic_reason: 873 diag_reason = self._cel_evaluator.last_diagnostic_reason 874 elif "entities." in str(cel_expression): 875 diag_reason = "CEL_HYDRATION_FAILED (hydrated owner mismatch)" 876 decision = self._record_and_return( 877 request=request, 878 decision_status=DECISION_DENY, 879 diagnostic_reason=diag_reason, 880 severity=SEVERITY_SECURITY_ALERT, 881 obligations={"strategy": strategy}, 882 reason_code=None, 883 ) 884 self._cache.put_static(cache_key, decision) 885 return decision 886 887 # 5. Declarative Deny-Overrides Evaluation (if custom rules are defined) 888 if self._rules: 889 matched_permit = False 890 for rule in self._rules: 891 if rule.target_action == "*" or rule.target_action == request.action: 892 # Validate rule constraints: roles, clearance, cel_condition 893 rule_matches = True 894 895 if rule.required_roles: 896 expanded_roles = self._role_graph.expand_roles(subject.roles) 897 if not set(rule.required_roles).intersection(expanded_roles): 898 rule_matches = False 899 900 if rule_matches and rule.min_clearance is not None: 901 if subject.clearance_level < int(rule.min_clearance): 902 rule_matches = False 903 904 if rule_matches and rule.cel_condition: 905 runtime_args = resource.get("arguments", {}) 906 tool_meta = resource.get("tool_metadata", {}) 907 if not self._cel_evaluator.evaluate( 908 expression=rule.cel_condition, 909 subject=subject, 910 args=runtime_args, 911 tool=tool_meta, 912 config=req_config, 913 ): 914 rule_matches = False 915 916 if rule_matches: 917 if rule.effect == DECISION_DENY: 918 decision = self._record_and_return( 919 request=request, 920 decision_status=DECISION_DENY, 921 diagnostic_reason=f"Rule {rule.rule_id} DENY", 922 severity=SEVERITY_SECURITY_ALERT, 923 obligations={"strategy": rule.on_denial}, 924 reason_code=None, 925 ) 926 self._cache.put_static(cache_key, decision) 927 return decision 928 elif rule.effect == DECISION_PERMIT: 929 matched_permit = True 930 931 if not matched_permit: 932 decision = self._record_and_return( 933 request=request, 934 decision_status=DECISION_NOT_APPLICABLE, 935 diagnostic_reason="No matching permit policy rule found", 936 severity=SEVERITY_WARNING, 937 obligations={}, 938 reason_code=None, 939 ) 940 self._cache.put_static(cache_key, decision) 941 return decision 942 943 # 6. Permit Default 944 decision = self._record_and_return( 945 request=request, 946 decision_status=DECISION_PERMIT, 947 diagnostic_reason=REASON_AUTHORIZED, 948 severity=SEVERITY_INFORMATIONAL, 949 obligations={}, 950 reason_code=None, 951 ) 952 self._cache.put_static(cache_key, decision) 953 return decision
Evaluate access request through tenant isolation, quarantine, clearance lattice, roles, and CEL constraints.
Args: request: Evaluation request envelope.
Returns: PolicyDecision: Evaluation decision status and obligations.
class
RoleGraph:
274class RoleGraph: 275 """Directed Acyclic Graph (DAG) for hierarchical role inheritance resolution.""" 276 277 def __init__(self): 278 """Initialize empty RoleGraph.""" 279 self._inheritance: Dict[str, Set[str]] = {} 280 281 @classmethod 282 def create(cls): 283 """Factory method creating a RoleGraph. 284 285 Returns: 286 RoleGraph: New RoleGraph instance. 287 """ 288 return cls() 289 290 def add_inheritance(self, parent_role: str, child_role: str) -> None: 291 """Declares that parent_role inherits child_role permissions. 292 293 Args: 294 parent_role: Role that inherits permissions. 295 child_role: Role whose permissions are inherited. 296 297 Raises: 298 ValueError: If adding inheritance creates a cyclic role graph. 299 """ 300 if parent_role not in self._inheritance: 301 self._inheritance[parent_role] = set() 302 self._inheritance[parent_role].add(child_role) 303 self._assert_no_cycles() 304 305 def expand_roles(self, roles: Set[str]) -> Set[str]: 306 """Transitively expands assigned roles through graph hierarchy. 307 308 Args: 309 roles: Base set of assigned caller roles. 310 311 Returns: 312 Set[str]: Expanded set including all inherited roles. 313 """ 314 expanded: Set[str] = set() 315 visited: Set[str] = set() 316 317 def dfs(role: str) -> None: 318 if role in visited: 319 return 320 visited.add(role) 321 expanded.add(role) 322 for child in self._inheritance.get(role, set()): 323 dfs(child) 324 325 for r in roles: 326 dfs(r) 327 return expanded 328 329 def _assert_no_cycles(self) -> None: 330 visited: Set[str] = set() 331 rec_stack: Set[str] = set() 332 333 def has_cycle(node: str) -> bool: 334 visited.add(node) 335 rec_stack.add(node) 336 for neighbor in self._inheritance.get(node, set()): 337 if neighbor not in visited: 338 if has_cycle(neighbor): 339 return True 340 elif neighbor in rec_stack: 341 return True 342 rec_stack.remove(node) 343 return False 344 345 for role in self._inheritance: 346 if role not in visited: 347 if has_cycle(role): 348 raise ValueError(f"Cyclic role inheritance detected at role: '{role}'")
Directed Acyclic Graph (DAG) for hierarchical role inheritance resolution.
RoleGraph()
277 def __init__(self): 278 """Initialize empty RoleGraph.""" 279 self._inheritance: Dict[str, Set[str]] = {}
Initialize empty RoleGraph.
@classmethod
def
create(cls):
281 @classmethod 282 def create(cls): 283 """Factory method creating a RoleGraph. 284 285 Returns: 286 RoleGraph: New RoleGraph instance. 287 """ 288 return cls()
Factory method creating a RoleGraph.
Returns: RoleGraph: New RoleGraph instance.
def
add_inheritance(self, parent_role: str, child_role: str) -> None:
290 def add_inheritance(self, parent_role: str, child_role: str) -> None: 291 """Declares that parent_role inherits child_role permissions. 292 293 Args: 294 parent_role: Role that inherits permissions. 295 child_role: Role whose permissions are inherited. 296 297 Raises: 298 ValueError: If adding inheritance creates a cyclic role graph. 299 """ 300 if parent_role not in self._inheritance: 301 self._inheritance[parent_role] = set() 302 self._inheritance[parent_role].add(child_role) 303 self._assert_no_cycles()
Declares that parent_role inherits child_role permissions.
Args: parent_role: Role that inherits permissions. child_role: Role whose permissions are inherited.
Raises: ValueError: If adding inheritance creates a cyclic role graph.
def
expand_roles(self, roles: Set[str]) -> Set[str]:
305 def expand_roles(self, roles: Set[str]) -> Set[str]: 306 """Transitively expands assigned roles through graph hierarchy. 307 308 Args: 309 roles: Base set of assigned caller roles. 310 311 Returns: 312 Set[str]: Expanded set including all inherited roles. 313 """ 314 expanded: Set[str] = set() 315 visited: Set[str] = set() 316 317 def dfs(role: str) -> None: 318 if role in visited: 319 return 320 visited.add(role) 321 expanded.add(role) 322 for child in self._inheritance.get(role, set()): 323 dfs(child) 324 325 for r in roles: 326 dfs(r) 327 return expanded
Transitively expands assigned roles through graph hierarchy.
Args: roles: Base set of assigned caller roles.
Returns: Set[str]: Expanded set including all inherited roles.
class
TwoTierDecisionCache:
432class TwoTierDecisionCache: 433 """Two-tier caching engine (Tier 1 Invocation-Scoped + Tier 2 Cross-Request Static).""" 434 435 def __init__(self, *args: Any, **kwargs: Any): 436 """Initialize TwoTierDecisionCache. 437 438 Args: 439 *args: Optional positional arguments (ttl_seconds, max_entries, l1_enabled, l2_enabled). 440 **kwargs: Optional keyword arguments. 441 """ 442 ttl_seconds = kwargs.get("ttl_seconds", args[0] if len(args) > 0 else DEFAULT_CACHE_TTL_SECONDS) 443 max_entries = kwargs.get("max_entries", args[1] if len(args) > 1 else DEFAULT_CACHE_MAX_ENTRIES) 444 l1_enabled = kwargs.get("l1_enabled", args[2] if len(args) > 2 else True) 445 l2_enabled = kwargs.get("l2_enabled", args[3] if len(args) > 3 else True) 446 self._tier2_cache: Dict[str, tuple[PolicyDecision, float]] = {} 447 self._ttl_seconds = float(ttl_seconds) 448 self._max_entries = int(max_entries) 449 self._l1_enabled = bool(l1_enabled) 450 self._l2_enabled = bool(l2_enabled) 451 452 @classmethod 453 def create(cls, *args: Any, **kwargs: Any): 454 """Factory method creating TwoTierDecisionCache. 455 456 Args: 457 *args: Optional positional arguments. 458 **kwargs: Optional keyword arguments. 459 460 Returns: 461 TwoTierDecisionCache: New cache instance. 462 """ 463 return cls(*args, **kwargs) 464 465 @property 466 def l1_enabled(self) -> bool: 467 """Check if Tier 1 invocation cache is enabled.""" 468 return self._l1_enabled 469 470 @property 471 def l2_enabled(self) -> bool: 472 """Check if Tier 2 static cache is enabled.""" 473 return self._l2_enabled 474 475 def compute_request_hash(self, request: PolicyEvaluationRequest) -> str: 476 """Compute SHA256 deterministic hash key prefixed by subject_id for an evaluation request. 477 478 Args: 479 request: Policy evaluation request object. 480 481 Returns: 482 str: Subject-prefixed hexadecimal SHA256 digest string. 483 """ 484 res_str = json.dumps(request.resource, sort_keys=True, default=str) 485 roles_str = ",".join(sorted(list(request.subject.roles))) if request.subject.roles else "" 486 s = f"{request.subject.subject_id}:{request.subject.tenant_id}:{request.subject.clearance_level}:{request.subject.department_id}:{roles_str}:{request.action}:{res_str}" 487 digest = hashlib.sha256(s.encode()).hexdigest() 488 return f"{request.subject.subject_id}:{digest}" 489 490 def get_invocation_cache(self, config: Optional[Dict[str, Any]], cache_key: str) -> Optional[PolicyDecision]: 491 """Tier 1: Looks up decision in RunnableConfig['configurable']['sec:decision_cache']. 492 493 Args: 494 config: RunnableConfig dictionary. 495 cache_key: Request hash key. 496 497 Returns: 498 Optional[PolicyDecision]: Cached decision or None if miss. 499 """ 500 if not self._l1_enabled or not config or not isinstance(config, dict): 501 return None 502 configurable = config.get("configurable") 503 if isinstance(configurable, dict): 504 inv_cache = configurable.get(CONFIG_KEY_DECISION_CACHE) 505 if isinstance(inv_cache, dict): 506 return inv_cache.get(cache_key) 507 return None 508 509 def put_invocation_cache(self, config: Optional[Dict[str, Any]], cache_key: str, decision: PolicyDecision) -> None: 510 """Tier 1: Stores decision in RunnableConfig['configurable']['sec:decision_cache']. 511 512 Args: 513 config: Target RunnableConfig dictionary. 514 cache_key: Request hash key. 515 decision: Computed PolicyDecision object to store. 516 """ 517 if not self._l1_enabled or not config or not isinstance(config, dict): 518 return 519 configurable = config.get("configurable") 520 if isinstance(configurable, dict): 521 if CONFIG_KEY_DECISION_CACHE not in configurable or not isinstance(configurable[CONFIG_KEY_DECISION_CACHE], dict): 522 configurable[CONFIG_KEY_DECISION_CACHE] = {} 523 configurable[CONFIG_KEY_DECISION_CACHE][cache_key] = decision 524 525 def get_static(self, cache_key: str) -> Optional[PolicyDecision]: 526 """Tier 2: Looks up decision in process-wide static cache with TTL. 527 528 Args: 529 cache_key: Hash key. 530 531 Returns: 532 Optional[PolicyDecision]: Unexpired cached decision or None. 533 """ 534 if not self._l2_enabled: 535 return None 536 entry = self._tier2_cache.get(cache_key) 537 if entry: 538 decision, ts = entry 539 if time.time() - ts <= self._ttl_seconds: 540 return decision 541 del self._tier2_cache[cache_key] 542 return None 543 544 def put_static(self, cache_key: str, decision: PolicyDecision) -> None: 545 """Tier 2: Stores decision in process-wide static cache with bounded max entries. 546 547 Args: 548 cache_key: Hash key. 549 decision: PolicyDecision to cache. 550 """ 551 if not self._l2_enabled: 552 return 553 if len(self._tier2_cache) >= self._max_entries and cache_key not in self._tier2_cache: 554 oldest_key = min(self._tier2_cache.keys(), key=lambda k: self._tier2_cache[k][1]) 555 del self._tier2_cache[oldest_key] 556 self._tier2_cache[cache_key] = (decision, time.time()) 557 558 def evict_subject(self, subject_id: str, *args: Any, **kwargs: Any) -> int: 559 """Evict all cached policy decisions for a specific subject across L2 and optional L1 config. 560 561 Args: 562 subject_id: Subject identifier whose decisions must be evicted. 563 *args: Optional positional config dict. 564 **kwargs: Optional keyword config dict. 565 566 Returns: 567 int: Count of evicted cache entries. 568 """ 569 prefix = f"{subject_id}:" 570 evicted = 0 571 keys_to_remove = [k for k in self._tier2_cache.keys() if k.startswith(prefix)] 572 for k in keys_to_remove: 573 del self._tier2_cache[k] 574 evicted += 1 575 576 config = kwargs.get("config", args[0] if len(args) > 0 else None) 577 if isinstance(config, dict): 578 configurable = config.get("configurable") 579 if isinstance(configurable, dict): 580 inv_cache = configurable.get(CONFIG_KEY_DECISION_CACHE) 581 if isinstance(inv_cache, dict): 582 l1_keys = [k for k in inv_cache.keys() if k.startswith(prefix)] 583 for k in l1_keys: 584 del inv_cache[k] 585 evicted += 1 586 return evicted 587 588 def flush(self) -> None: 589 """Flush all Tier 2 static cache entries.""" 590 self._tier2_cache.clear()
Two-tier caching engine (Tier 1 Invocation-Scoped + Tier 2 Cross-Request Static).
TwoTierDecisionCache(*args: Any, **kwargs: Any)
435 def __init__(self, *args: Any, **kwargs: Any): 436 """Initialize TwoTierDecisionCache. 437 438 Args: 439 *args: Optional positional arguments (ttl_seconds, max_entries, l1_enabled, l2_enabled). 440 **kwargs: Optional keyword arguments. 441 """ 442 ttl_seconds = kwargs.get("ttl_seconds", args[0] if len(args) > 0 else DEFAULT_CACHE_TTL_SECONDS) 443 max_entries = kwargs.get("max_entries", args[1] if len(args) > 1 else DEFAULT_CACHE_MAX_ENTRIES) 444 l1_enabled = kwargs.get("l1_enabled", args[2] if len(args) > 2 else True) 445 l2_enabled = kwargs.get("l2_enabled", args[3] if len(args) > 3 else True) 446 self._tier2_cache: Dict[str, tuple[PolicyDecision, float]] = {} 447 self._ttl_seconds = float(ttl_seconds) 448 self._max_entries = int(max_entries) 449 self._l1_enabled = bool(l1_enabled) 450 self._l2_enabled = bool(l2_enabled)
Initialize TwoTierDecisionCache.
Args: *args: Optional positional arguments (ttl_seconds, max_entries, l1_enabled, l2_enabled). **kwargs: Optional keyword arguments.
@classmethod
def
create(cls, *args: Any, **kwargs: Any):
452 @classmethod 453 def create(cls, *args: Any, **kwargs: Any): 454 """Factory method creating TwoTierDecisionCache. 455 456 Args: 457 *args: Optional positional arguments. 458 **kwargs: Optional keyword arguments. 459 460 Returns: 461 TwoTierDecisionCache: New cache instance. 462 """ 463 return cls(*args, **kwargs)
Factory method creating TwoTierDecisionCache.
Args: *args: Optional positional arguments. **kwargs: Optional keyword arguments.
Returns: TwoTierDecisionCache: New cache instance.
l1_enabled: bool
465 @property 466 def l1_enabled(self) -> bool: 467 """Check if Tier 1 invocation cache is enabled.""" 468 return self._l1_enabled
Check if Tier 1 invocation cache is enabled.
l2_enabled: bool
470 @property 471 def l2_enabled(self) -> bool: 472 """Check if Tier 2 static cache is enabled.""" 473 return self._l2_enabled
Check if Tier 2 static cache is enabled.
475 def compute_request_hash(self, request: PolicyEvaluationRequest) -> str: 476 """Compute SHA256 deterministic hash key prefixed by subject_id for an evaluation request. 477 478 Args: 479 request: Policy evaluation request object. 480 481 Returns: 482 str: Subject-prefixed hexadecimal SHA256 digest string. 483 """ 484 res_str = json.dumps(request.resource, sort_keys=True, default=str) 485 roles_str = ",".join(sorted(list(request.subject.roles))) if request.subject.roles else "" 486 s = f"{request.subject.subject_id}:{request.subject.tenant_id}:{request.subject.clearance_level}:{request.subject.department_id}:{roles_str}:{request.action}:{res_str}" 487 digest = hashlib.sha256(s.encode()).hexdigest() 488 return f"{request.subject.subject_id}:{digest}"
Compute SHA256 deterministic hash key prefixed by subject_id for an evaluation request.
Args: request: Policy evaluation request object.
Returns: str: Subject-prefixed hexadecimal SHA256 digest string.
def
get_invocation_cache( self, config: Optional[Dict[str, Any]], cache_key: str) -> Optional[PolicyDecision]:
490 def get_invocation_cache(self, config: Optional[Dict[str, Any]], cache_key: str) -> Optional[PolicyDecision]: 491 """Tier 1: Looks up decision in RunnableConfig['configurable']['sec:decision_cache']. 492 493 Args: 494 config: RunnableConfig dictionary. 495 cache_key: Request hash key. 496 497 Returns: 498 Optional[PolicyDecision]: Cached decision or None if miss. 499 """ 500 if not self._l1_enabled or not config or not isinstance(config, dict): 501 return None 502 configurable = config.get("configurable") 503 if isinstance(configurable, dict): 504 inv_cache = configurable.get(CONFIG_KEY_DECISION_CACHE) 505 if isinstance(inv_cache, dict): 506 return inv_cache.get(cache_key) 507 return None
Tier 1: Looks up decision in RunnableConfig['configurable']['sec:decision_cache'].
Args: config: RunnableConfig dictionary. cache_key: Request hash key.
Returns: Optional[PolicyDecision]: Cached decision or None if miss.
def
put_invocation_cache( self, config: Optional[Dict[str, Any]], cache_key: str, decision: PolicyDecision) -> None:
509 def put_invocation_cache(self, config: Optional[Dict[str, Any]], cache_key: str, decision: PolicyDecision) -> None: 510 """Tier 1: Stores decision in RunnableConfig['configurable']['sec:decision_cache']. 511 512 Args: 513 config: Target RunnableConfig dictionary. 514 cache_key: Request hash key. 515 decision: Computed PolicyDecision object to store. 516 """ 517 if not self._l1_enabled or not config or not isinstance(config, dict): 518 return 519 configurable = config.get("configurable") 520 if isinstance(configurable, dict): 521 if CONFIG_KEY_DECISION_CACHE not in configurable or not isinstance(configurable[CONFIG_KEY_DECISION_CACHE], dict): 522 configurable[CONFIG_KEY_DECISION_CACHE] = {} 523 configurable[CONFIG_KEY_DECISION_CACHE][cache_key] = decision
Tier 1: Stores decision in RunnableConfig['configurable']['sec:decision_cache'].
Args: config: Target RunnableConfig dictionary. cache_key: Request hash key. decision: Computed PolicyDecision object to store.
525 def get_static(self, cache_key: str) -> Optional[PolicyDecision]: 526 """Tier 2: Looks up decision in process-wide static cache with TTL. 527 528 Args: 529 cache_key: Hash key. 530 531 Returns: 532 Optional[PolicyDecision]: Unexpired cached decision or None. 533 """ 534 if not self._l2_enabled: 535 return None 536 entry = self._tier2_cache.get(cache_key) 537 if entry: 538 decision, ts = entry 539 if time.time() - ts <= self._ttl_seconds: 540 return decision 541 del self._tier2_cache[cache_key] 542 return None
Tier 2: Looks up decision in process-wide static cache with TTL.
Args: cache_key: Hash key.
Returns: Optional[PolicyDecision]: Unexpired cached decision or None.
544 def put_static(self, cache_key: str, decision: PolicyDecision) -> None: 545 """Tier 2: Stores decision in process-wide static cache with bounded max entries. 546 547 Args: 548 cache_key: Hash key. 549 decision: PolicyDecision to cache. 550 """ 551 if not self._l2_enabled: 552 return 553 if len(self._tier2_cache) >= self._max_entries and cache_key not in self._tier2_cache: 554 oldest_key = min(self._tier2_cache.keys(), key=lambda k: self._tier2_cache[k][1]) 555 del self._tier2_cache[oldest_key] 556 self._tier2_cache[cache_key] = (decision, time.time())
Tier 2: Stores decision in process-wide static cache with bounded max entries.
Args: cache_key: Hash key. decision: PolicyDecision to cache.
def
evict_subject(self, subject_id: str, *args: Any, **kwargs: Any) -> int:
558 def evict_subject(self, subject_id: str, *args: Any, **kwargs: Any) -> int: 559 """Evict all cached policy decisions for a specific subject across L2 and optional L1 config. 560 561 Args: 562 subject_id: Subject identifier whose decisions must be evicted. 563 *args: Optional positional config dict. 564 **kwargs: Optional keyword config dict. 565 566 Returns: 567 int: Count of evicted cache entries. 568 """ 569 prefix = f"{subject_id}:" 570 evicted = 0 571 keys_to_remove = [k for k in self._tier2_cache.keys() if k.startswith(prefix)] 572 for k in keys_to_remove: 573 del self._tier2_cache[k] 574 evicted += 1 575 576 config = kwargs.get("config", args[0] if len(args) > 0 else None) 577 if isinstance(config, dict): 578 configurable = config.get("configurable") 579 if isinstance(configurable, dict): 580 inv_cache = configurable.get(CONFIG_KEY_DECISION_CACHE) 581 if isinstance(inv_cache, dict): 582 l1_keys = [k for k in inv_cache.keys() if k.startswith(prefix)] 583 for k in l1_keys: 584 del inv_cache[k] 585 evicted += 1 586 return evicted
Evict all cached policy decisions for a specific subject across L2 and optional L1 config.
Args: subject_id: Subject identifier whose decisions must be evicted. *args: Optional positional config dict. **kwargs: Optional keyword config dict.
Returns: int: Count of evicted cache entries.
PolicyRequest =
<class 'PolicyEvaluationRequest'>
class
CELEvaluator:
50class CELEvaluator: 51 """Evaluates parameter constraint expressions safely over evaluation context.""" 52 53 def __init__(self, *args: Any, **kwargs: Any): 54 """Initialize CELEvaluator.""" 55 entity_resolver = kwargs.get("entity_resolver", args[0] if len(args) > 0 else DEFAULT_ENTITY_RESOLVER) 56 self._entity_resolver = entity_resolver 57 self._last_diagnostic_reason: Optional[str] = None 58 self._last_hydrated_entities: Dict[str, Dict[str, Any]] = {} 59 60 @classmethod 61 def create(cls, *args: Any, **kwargs: Any): 62 """Factory method to construct a CELEvaluator.""" 63 entity_resolver = kwargs.get("entity_resolver", args[0] if len(args) > 0 else DEFAULT_ENTITY_RESOLVER) 64 return cls(entity_resolver=entity_resolver) 65 66 @property 67 def entity_resolver(self) -> Optional[EntityAttributeResolverSPI]: 68 """Get the configured entity resolver SPI instance.""" 69 return self._entity_resolver 70 71 @entity_resolver.setter 72 def entity_resolver(self, resolver: Optional[EntityAttributeResolverSPI]) -> None: 73 """Set or update the entity resolver SPI instance.""" 74 self._entity_resolver = resolver 75 76 @property 77 def last_diagnostic_reason(self) -> Optional[str]: 78 """Get the diagnostic reason code from the most recent evaluation, if any.""" 79 return self._last_diagnostic_reason 80 81 @property 82 def last_hydrated_entities(self) -> Dict[str, Dict[str, Any]]: 83 """Get the map of entities hydrated during the most recent evaluation.""" 84 return dict(self._last_hydrated_entities) 85 86 def evaluate( 87 self, 88 expression: str, 89 subject: SecurityIdentity, 90 args: Dict[str, Any], 91 tool: Dict[str, Any], 92 *extra_args: Any, 93 **kwargs: Any, 94 ) -> bool: 95 """Evaluate a CEL parameter constraint expression against invocation context. 96 97 Args: 98 expression: Common Expression Language (CEL) boolean condition string. 99 subject: SecurityIdentity envelope of the calling user or agent. 100 args: Map of resolved runtime argument parameters passed to the tool. 101 tool: Metadata attributes dictionary describing the target tool. 102 *extra_args: Optional positional arguments (e.g., config dictionary). 103 **kwargs: Optional keyword arguments (e.g., config dictionary). 104 105 Returns: 106 bool: True if the expression evaluates to truthy; False on failure or syntax error (fail-closed). 107 """ 108 self._last_diagnostic_reason = None 109 self._last_hydrated_entities = {} 110 if not expression or not expression.strip(): 111 return True 112 113 config = kwargs.get("config", extra_args[0] if len(extra_args) > 0 else None) 114 normalized_expr = self._preprocess_cel_syntax(expression) 115 eval_scope = self._build_evaluation_scope(subject, args, tool, config) 116 117 try: 118 tree = ast.parse(normalized_expr, mode="eval") 119 result = self._eval_ast_node(tree.body, eval_scope) 120 return bool(result) 121 except KeyError: 122 self._last_diagnostic_reason = DIAGNOSTIC_FAIL_CLOSED_NULL_ENTITY 123 return False 124 except Exception: 125 # Fail closed to DENY (False) on evaluation errors, missing variables, or syntax failures 126 return False 127 128 def _preprocess_cel_syntax(self, expr: str) -> str: 129 """Translates CEL operator syntax (&&, ||, ternary ?) into Python AST syntax.""" 130 result = expr.replace("&&", " and ").replace("||", " or ") 131 if "?" in result and ":" in result: 132 parts = result.split("?", 1) 133 condition = parts[0].strip() 134 rest = parts[1] 135 true_false = rest.split(":", 1) 136 true_val = true_false[0].strip() 137 false_val = true_false[1].strip() 138 result = f"({true_val}) if ({condition}) else ({false_val})" 139 return result 140 141 def _build_evaluation_scope( 142 self, 143 subject: SecurityIdentity, 144 args: Dict[str, Any], 145 tool: Dict[str, Any], 146 config: Optional[Dict[str, Any]], 147 ) -> Dict[str, Any]: 148 """Constructs attribute dictionary scope for subject, args, tool, and hydrated entities namespaces.""" 149 subject_dict = { 150 "id": subject.subject_id, 151 "tenant_id": subject.tenant_id, 152 "tenantId": subject.tenant_id, 153 "roles": subject.roles, 154 "clearance": subject.clearance_level, 155 "clearance_level": subject.clearance_level, 156 "dept": subject.department_id, 157 "department_id": subject.department_id, 158 "attributes": subject.attributes, 159 } 160 161 entity_cache: Dict[str, Dict[str, Any]] = {} 162 if isinstance(config, dict) and isinstance(config.get("configurable"), dict): 163 entity_cache = config["configurable"].setdefault(CONFIG_KEY_ENTITY_CACHE, {}) 164 165 entities_scope: Dict[str, Any] = {} 166 if self._entity_resolver: 167 evaluator_ref = self 168 169 class _EntityProxy: 170 def __init__(self, resolver: EntityAttributeResolverSPI, etype: str): 171 self._resolver = resolver 172 self._etype = etype 173 174 def _resolve_with_cache(self, eid: str) -> _EntityAttrDict: 175 cache_key = f"{self._etype}:{eid}" 176 if cache_key in entity_cache: 177 res = entity_cache[cache_key] 178 else: 179 res = self._resolver.resolve(self._etype, str(eid)) 180 if res is not None: 181 entity_cache[cache_key] = dict(res) 182 if res is None: 183 raise KeyError(f"Unresolvable entity '{eid}' of type '{self._etype}'") 184 evaluator_ref._last_hydrated_entities[cache_key] = dict(res) 185 return _EntityAttrDict(res) 186 187 def __getitem__(self, eid: str) -> _EntityAttrDict: 188 return self._resolve_with_cache(str(eid)) 189 190 def __getattr__(self, eid: str) -> _EntityAttrDict: 191 return self._resolve_with_cache(str(eid)) 192 193 class _EntitiesDict(dict): 194 def __init__(self, resolver: EntityAttributeResolverSPI): 195 self._resolver = resolver 196 197 def __getitem__(self, etype: str) -> Any: 198 return _EntityProxy(self._resolver, etype) 199 200 def __getattr__(self, etype: str) -> Any: 201 return _EntityProxy(self._resolver, etype) 202 203 entities_scope = _EntitiesDict(self._entity_resolver) 204 205 return { 206 "subject": _AttrDict(subject_dict), 207 "args": _AttrDict(args), 208 "tool": _AttrDict(tool), 209 "entities": entities_scope, 210 "true": True, 211 "false": False, 212 "null": None, 213 "None": None, 214 } 215 216 def _eval_ast_node(self, node: ast.AST, scope: Dict[str, Any]) -> Any: 217 """Safe AST node recursive evaluation.""" 218 if isinstance(node, ast.Expression): 219 return self._eval_ast_node(node.body, scope) 220 221 if isinstance(node, ast.Constant): 222 return node.value 223 224 if isinstance(node, ast.Name): 225 if node.id in scope: 226 return scope[node.id] 227 raise NameError(f"Undefined variable in expression: '{node.id}'") 228 229 if isinstance(node, ast.Attribute): 230 obj = self._eval_ast_node(node.value, scope) 231 if hasattr(obj, node.attr): 232 return getattr(obj, node.attr) 233 if isinstance(obj, str): 234 if node.attr == "startsWith": 235 return obj.startswith 236 elif node.attr == "endsWith": 237 return obj.endswith 238 elif node.attr == "contains": 239 return lambda sub: sub in obj 240 elif node.attr in ("toLowerCase", "lower"): 241 return obj.lower 242 elif node.attr in ("toUpperCase", "upper"): 243 return obj.upper 244 if isinstance(obj, dict) and node.attr in obj: 245 return obj[node.attr] 246 return None 247 248 if isinstance(node, ast.Subscript): 249 value = self._eval_ast_node(node.value, scope) 250 slice_val = self._eval_ast_node(node.slice, scope) 251 return value[slice_val] 252 253 if isinstance(node, ast.UnaryOp): 254 operand = self._eval_ast_node(node.operand, scope) 255 if isinstance(node.op, ast.Not): 256 return not operand 257 if isinstance(node.op, ast.USub): 258 return -operand 259 if isinstance(node.op, ast.UAdd): 260 return +operand 261 262 if isinstance(node, ast.BinOp): 263 left = self._eval_ast_node(node.left, scope) 264 right = self._eval_ast_node(node.right, scope) 265 if isinstance(node.op, ast.Add): 266 return left + right 267 if isinstance(node.op, ast.Sub): 268 return left - right 269 if isinstance(node.op, ast.Mult): 270 return left * right 271 if isinstance(node.op, ast.Div): 272 return left / right 273 if isinstance(node.op, ast.Mod): 274 return left % right 275 276 if isinstance(node, ast.BoolOp): 277 values = [self._eval_ast_node(val, scope) for val in node.values] 278 if isinstance(node.op, ast.And): 279 return all(values) 280 if isinstance(node.op, ast.Or): 281 return any(values) 282 283 if isinstance(node, ast.Compare): 284 left = self._eval_ast_node(node.left, scope) 285 for op, comparator in zip(node.ops, node.comparators): 286 right = self._eval_ast_node(comparator, scope) 287 if isinstance(op, ast.Eq) and not (left == right): 288 return False 289 elif isinstance(op, ast.NotEq) and not (left != right): 290 return False 291 elif isinstance(op, ast.Lt) and not (left < right): 292 return False 293 elif isinstance(op, ast.LtE) and not (left <= right): 294 return False 295 elif isinstance(op, ast.Gt) and not (left > right): 296 return False 297 elif isinstance(op, ast.GtE) and not (left >= right): 298 return False 299 elif isinstance(op, ast.In) and not (left in right): 300 return False 301 elif isinstance(op, ast.NotIn) and not (left not in right): 302 return False 303 left = right 304 return True 305 306 if isinstance(node, ast.IfExp): 307 test = self._eval_ast_node(node.test, scope) 308 if test: 309 return self._eval_ast_node(node.body, scope) 310 else: 311 return self._eval_ast_node(node.orelse, scope) 312 313 if isinstance(node, ast.Call): 314 func = self._eval_ast_node(node.func, scope) 315 args = [self._eval_ast_node(arg, scope) for arg in node.args] 316 if callable(func): 317 return func(*args) 318 return None 319 320 if isinstance(node, ast.List): 321 return [self._eval_ast_node(el, scope) for el in node.elts] 322 323 if isinstance(node, ast.Tuple): 324 return tuple(self._eval_ast_node(el, scope) for el in node.elts) 325 326 if isinstance(node, ast.Set): 327 return set(self._eval_ast_node(el, scope) for el in node.elts) 328 329 if isinstance(node, ast.Dict): 330 keys = [self._eval_ast_node(k, scope) for k in node.keys] 331 vals = [self._eval_ast_node(v, scope) for v in node.values] 332 return dict(zip(keys, vals)) 333 334 raise TypeError(f"Unsupported AST node type: {type(node).__name__}")
Evaluates parameter constraint expressions safely over evaluation context.
CELEvaluator(*args: Any, **kwargs: Any)
53 def __init__(self, *args: Any, **kwargs: Any): 54 """Initialize CELEvaluator.""" 55 entity_resolver = kwargs.get("entity_resolver", args[0] if len(args) > 0 else DEFAULT_ENTITY_RESOLVER) 56 self._entity_resolver = entity_resolver 57 self._last_diagnostic_reason: Optional[str] = None 58 self._last_hydrated_entities: Dict[str, Dict[str, Any]] = {}
Initialize CELEvaluator.
@classmethod
def
create(cls, *args: Any, **kwargs: Any):
60 @classmethod 61 def create(cls, *args: Any, **kwargs: Any): 62 """Factory method to construct a CELEvaluator.""" 63 entity_resolver = kwargs.get("entity_resolver", args[0] if len(args) > 0 else DEFAULT_ENTITY_RESOLVER) 64 return cls(entity_resolver=entity_resolver)
Factory method to construct a CELEvaluator.
entity_resolver: Optional[EntityAttributeResolverSPI]
66 @property 67 def entity_resolver(self) -> Optional[EntityAttributeResolverSPI]: 68 """Get the configured entity resolver SPI instance.""" 69 return self._entity_resolver
Get the configured entity resolver SPI instance.
last_diagnostic_reason: Optional[str]
76 @property 77 def last_diagnostic_reason(self) -> Optional[str]: 78 """Get the diagnostic reason code from the most recent evaluation, if any.""" 79 return self._last_diagnostic_reason
Get the diagnostic reason code from the most recent evaluation, if any.
last_hydrated_entities: Dict[str, Dict[str, Any]]
81 @property 82 def last_hydrated_entities(self) -> Dict[str, Dict[str, Any]]: 83 """Get the map of entities hydrated during the most recent evaluation.""" 84 return dict(self._last_hydrated_entities)
Get the map of entities hydrated during the most recent evaluation.
def
evaluate( self, expression: str, subject: SecurityIdentity, args: Dict[str, Any], tool: Dict[str, Any], *extra_args: Any, **kwargs: Any) -> bool:
86 def evaluate( 87 self, 88 expression: str, 89 subject: SecurityIdentity, 90 args: Dict[str, Any], 91 tool: Dict[str, Any], 92 *extra_args: Any, 93 **kwargs: Any, 94 ) -> bool: 95 """Evaluate a CEL parameter constraint expression against invocation context. 96 97 Args: 98 expression: Common Expression Language (CEL) boolean condition string. 99 subject: SecurityIdentity envelope of the calling user or agent. 100 args: Map of resolved runtime argument parameters passed to the tool. 101 tool: Metadata attributes dictionary describing the target tool. 102 *extra_args: Optional positional arguments (e.g., config dictionary). 103 **kwargs: Optional keyword arguments (e.g., config dictionary). 104 105 Returns: 106 bool: True if the expression evaluates to truthy; False on failure or syntax error (fail-closed). 107 """ 108 self._last_diagnostic_reason = None 109 self._last_hydrated_entities = {} 110 if not expression or not expression.strip(): 111 return True 112 113 config = kwargs.get("config", extra_args[0] if len(extra_args) > 0 else None) 114 normalized_expr = self._preprocess_cel_syntax(expression) 115 eval_scope = self._build_evaluation_scope(subject, args, tool, config) 116 117 try: 118 tree = ast.parse(normalized_expr, mode="eval") 119 result = self._eval_ast_node(tree.body, eval_scope) 120 return bool(result) 121 except KeyError: 122 self._last_diagnostic_reason = DIAGNOSTIC_FAIL_CLOSED_NULL_ENTITY 123 return False 124 except Exception: 125 # Fail closed to DENY (False) on evaluation errors, missing variables, or syntax failures 126 return False
Evaluate a CEL parameter constraint expression against invocation context.
Args: expression: Common Expression Language (CEL) boolean condition string. subject: SecurityIdentity envelope of the calling user or agent. args: Map of resolved runtime argument parameters passed to the tool. tool: Metadata attributes dictionary describing the target tool. *extra_args: Optional positional arguments (e.g., config dictionary). **kwargs: Optional keyword arguments (e.g., config dictionary).
Returns: bool: True if the expression evaluates to truthy; False on failure or syntax error (fail-closed).
class
SecurityRegistry:
14class SecurityRegistry: 15 """Decoupled global SPI registry for host applications.""" 16 17 _default_pdp_engine: Optional[PolicyDecisionEngine] = None 18 _default_entity_resolver: Optional[EntityAttributeResolverSPI] = None 19 20 @classmethod 21 def configure( 22 cls, 23 pdp_engine: Optional[PolicyDecisionEngine] = None, 24 entity_resolver: Optional[EntityAttributeResolverSPI] = None, 25 ) -> None: 26 """Register default engine and resolver instances. 27 28 Args: 29 pdp_engine: Default PolicyDecisionEngine implementation for security evaluations. 30 entity_resolver: Default EntityAttributeResolverSPI for hydrating dynamic entity attributes in CEL. 31 """ 32 if pdp_engine is not None: 33 cls._default_pdp_engine = pdp_engine 34 if entity_resolver is not None: 35 cls._default_entity_resolver = entity_resolver 36 37 @classmethod 38 def get_pdp_engine(cls) -> Optional[PolicyDecisionEngine]: 39 """Retrieve globally registered default PDP engine. 40 41 Returns: 42 Optional[PolicyDecisionEngine]: Globally configured PDP engine, or None if not set. 43 """ 44 return cls._default_pdp_engine 45 46 @classmethod 47 def get_entity_resolver(cls) -> Optional[EntityAttributeResolverSPI]: 48 """Retrieve globally registered default EntityAttributeResolverSPI. 49 50 Returns: 51 Optional[EntityAttributeResolverSPI]: Globally configured entity resolver, or None if not set. 52 """ 53 return cls._default_entity_resolver 54 55 @classmethod 56 def reset(cls) -> None: 57 """Reset registered defaults to None (typically used in test teardowns).""" 58 cls._default_pdp_engine = None 59 cls._default_entity_resolver = None
Decoupled global SPI registry for host applications.
@classmethod
def
configure( cls, pdp_engine: Optional[PolicyDecisionEngine] = None, entity_resolver: Optional[EntityAttributeResolverSPI] = None) -> None:
20 @classmethod 21 def configure( 22 cls, 23 pdp_engine: Optional[PolicyDecisionEngine] = None, 24 entity_resolver: Optional[EntityAttributeResolverSPI] = None, 25 ) -> None: 26 """Register default engine and resolver instances. 27 28 Args: 29 pdp_engine: Default PolicyDecisionEngine implementation for security evaluations. 30 entity_resolver: Default EntityAttributeResolverSPI for hydrating dynamic entity attributes in CEL. 31 """ 32 if pdp_engine is not None: 33 cls._default_pdp_engine = pdp_engine 34 if entity_resolver is not None: 35 cls._default_entity_resolver = entity_resolver
Register default engine and resolver instances.
Args: pdp_engine: Default PolicyDecisionEngine implementation for security evaluations. entity_resolver: Default EntityAttributeResolverSPI for hydrating dynamic entity attributes in CEL.
37 @classmethod 38 def get_pdp_engine(cls) -> Optional[PolicyDecisionEngine]: 39 """Retrieve globally registered default PDP engine. 40 41 Returns: 42 Optional[PolicyDecisionEngine]: Globally configured PDP engine, or None if not set. 43 """ 44 return cls._default_pdp_engine
Retrieve globally registered default PDP engine.
Returns: Optional[PolicyDecisionEngine]: Globally configured PDP engine, or None if not set.
46 @classmethod 47 def get_entity_resolver(cls) -> Optional[EntityAttributeResolverSPI]: 48 """Retrieve globally registered default EntityAttributeResolverSPI. 49 50 Returns: 51 Optional[EntityAttributeResolverSPI]: Globally configured entity resolver, or None if not set. 52 """ 53 return cls._default_entity_resolver
Retrieve globally registered default EntityAttributeResolverSPI.
Returns: Optional[EntityAttributeResolverSPI]: Globally configured entity resolver, or None if not set.
@classmethod
def
reset(cls) -> None:
55 @classmethod 56 def reset(cls) -> None: 57 """Reset registered defaults to None (typically used in test teardowns).""" 58 cls._default_pdp_engine = None 59 cls._default_entity_resolver = None
Reset registered defaults to None (typically used in test teardowns).
class
AuditEvent:
35class AuditEvent: 36 """OCSF-aligned tamper-evident structured audit evaluation record. 37 38 Attributes: 39 event_id (str): Unique UUID string for event identification. 40 schema_version (str): Audit schema version specification. 41 timestamp_utc (float): UTC timestamp in seconds since epoch. 42 invocation_id (str): Unique trace context or invocation ID. 43 trace_id (Optional[str]): OpenTelemetry trace ID. 44 span_id (Optional[str]): OpenTelemetry span ID. 45 subject_id (str): Caller subject identifier. 46 tenant_id (str): Tenant isolation boundary identifier. 47 subject_roles (List[str]): Roles assigned to caller. 48 subject_clearance (int): Clearance level rank. 49 enforcement_point (str): Enforcement point identifier. 50 action (str): Action string requested. 51 target_resource (str): Target resource path or name. 52 decision (str): PDP decision outcome ('PERMIT', 'DENY', 'BLOCKED', etc.). 53 reason_code (str): Standardized security event reason code. 54 diagnostic_reason (str): Human-readable diagnostic description. 55 severity (str): Event severity level string. 56 payload_snapshot (Dict[str, Any]): Sanitized payload snapshot. 57 threat_category (Optional[str]): Threat classification label. 58 threat_score (Optional[float]): Threat confidence score. 59 detected_pattern (Optional[str]): Injection pattern ID. 60 mitigation_action (Optional[str]): Mitigation strategy applied. 61 sequence_number (int): Monotonic audit sequence number. 62 envelope_hmac (Optional[str]): Signature hash for tamper evidence. 63 """ 64 65 def __init__( 66 self, 67 event_id: str, 68 schema_version: str, 69 timestamp_utc: float, 70 invocation_id: str, 71 trace_id: Optional[str], 72 span_id: Optional[str], 73 subject_id: str, 74 tenant_id: str, 75 subject_roles: List[str], 76 subject_clearance: int, 77 enforcement_point: str, 78 action: str, 79 target_resource: str, 80 decision: str, 81 reason_code: str, 82 diagnostic_reason: str, 83 severity: str, 84 payload_snapshot: Dict[str, Any], 85 threat_category: Optional[str] = None, 86 threat_score: Optional[float] = None, 87 detected_pattern: Optional[str] = None, 88 mitigation_action: Optional[str] = None, 89 sequence_number: int = 0, 90 envelope_hmac: Optional[str] = None, 91 ): 92 """Initialize AuditEvent. 93 94 Args: 95 event_id: Unique UUID string for event identification. 96 schema_version: Audit schema version string. 97 timestamp_utc: UTC timestamp in seconds since epoch. 98 invocation_id: Unique trace context or invocation ID. 99 trace_id: OpenTelemetry trace ID if present. 100 span_id: OpenTelemetry span ID if present. 101 subject_id: Caller subject identifier. 102 tenant_id: Multi-tenant boundary identifier. 103 subject_roles: Roles assigned to caller. 104 subject_clearance: Clearance rank assigned to caller. 105 enforcement_point: Enforcement point identifier. 106 action: Requested action identifier. 107 target_resource: Target resource path or name. 108 decision: PDP decision verdict. 109 reason_code: Standardized security event reason code. 110 diagnostic_reason: Human-readable diagnostic description. 111 severity: Event severity level. 112 payload_snapshot: Sanitized payload snapshot dictionary. 113 threat_category: Optional threat classification category. 114 threat_score: Optional threat confidence score. 115 detected_pattern: Optional injection pattern identifier. 116 mitigation_action: Optional mitigation strategy applied. 117 sequence_number: Monotonic audit sequence number. 118 envelope_hmac: Optional HMAC signature for tamper evidence. 119 """ 120 self.event_id = event_id 121 self.schema_version = schema_version 122 self.timestamp_utc = timestamp_utc 123 self.invocation_id = invocation_id 124 self.trace_id = trace_id 125 self.span_id = span_id 126 self.subject_id = subject_id 127 self.tenant_id = tenant_id 128 self.subject_roles = list(subject_roles) 129 self.subject_clearance = subject_clearance 130 self.enforcement_point = enforcement_point 131 self.action = action 132 self.target_resource = target_resource 133 self.decision = decision 134 self.reason_code = reason_code 135 self.diagnostic_reason = diagnostic_reason 136 self.severity = severity 137 self.payload_snapshot = dict(payload_snapshot) 138 self.threat_category = threat_category 139 self.threat_score = threat_score 140 self.detected_pattern = detected_pattern 141 self.mitigation_action = mitigation_action 142 self.sequence_number = sequence_number 143 self.envelope_hmac = envelope_hmac 144 145 @classmethod 146 def create( 147 cls, 148 subject_id: str, 149 tenant_id: str, 150 action: str, 151 resource: str, 152 decision: str, 153 diagnostic_reason: str, 154 severity: str, 155 metadata: Dict[str, Any], 156 enforcement_point: str = "SYSTEM_BOUNDARY", 157 reason_code: str = "POLICY_EVALUATION", 158 ): 159 """Factory method to construct an AuditEvent with OpenTelemetry trace extraction. 160 161 Args: 162 subject_id: Subject identifier. 163 tenant_id: Tenant ID. 164 action: Requested action. 165 resource: Resource path/name. 166 decision: Decision verdict. 167 diagnostic_reason: Rationale description. 168 severity: Severity level. 169 metadata: Payload snapshot dictionary. 170 enforcement_point: PEP point ID. 171 reason_code: Event reason code. 172 173 Returns: 174 AuditEvent: Constructed audit event instance. 175 """ 176 trace_id = metadata.get("trace_id") 177 span_id = metadata.get("span_id") 178 if trace_id is None or span_id is None: 179 try: 180 from opentelemetry import trace 181 current_span = trace.get_current_span() 182 if current_span and current_span.get_span_context().is_valid: 183 ctx = current_span.get_span_context() 184 if trace_id is None: 185 trace_id = f"{ctx.trace_id:032x}" 186 if span_id is None: 187 span_id = f"{ctx.span_id:016x}" 188 except Exception: 189 pass 190 191 return cls( 192 event_id=str(uuid.uuid4()), 193 schema_version=DEFAULT_SCHEMA_VERSION, 194 timestamp_utc=time.time(), 195 invocation_id=metadata.get("invocation_id", str(uuid.uuid4())), 196 trace_id=trace_id, 197 span_id=span_id, 198 subject_id=subject_id, 199 tenant_id=tenant_id, 200 subject_roles=metadata.get("subject_roles", []), 201 subject_clearance=int(metadata.get("subject_clearance", 0)), 202 enforcement_point=enforcement_point, 203 action=action, 204 target_resource=resource, 205 decision=decision, 206 reason_code=reason_code, 207 diagnostic_reason=diagnostic_reason, 208 severity=severity, 209 payload_snapshot=metadata, 210 ) 211 212 @property 213 def resource(self) -> str: 214 """Get target resource identifier.""" 215 return self.target_resource 216 217 @property 218 def metadata(self) -> Dict[str, Any]: 219 """Get payload snapshot dictionary.""" 220 return self.payload_snapshot 221 222 def to_dict(self) -> Dict[str, Any]: 223 """Serialize AuditEvent into dictionary format. 224 225 Returns: 226 Dict[str, Any]: JSON-compatible event dictionary. 227 """ 228 return { 229 "event_id": self.event_id, 230 "schema_version": self.schema_version, 231 "timestamp_utc": self.timestamp_utc, 232 "invocation_id": self.invocation_id, 233 "trace_id": self.trace_id, 234 "span_id": self.span_id, 235 "subject_id": self.subject_id, 236 "tenant_id": self.tenant_id, 237 "subject_roles": self.subject_roles, 238 "subject_clearance": self.subject_clearance, 239 "enforcement_point": self.enforcement_point, 240 "action": self.action, 241 "target_resource": self.target_resource, 242 "decision": self.decision, 243 "reason_code": self.reason_code, 244 "diagnostic_reason": self.diagnostic_reason, 245 "severity": self.severity, 246 "payload_snapshot": self.payload_snapshot, 247 "threat_category": self.threat_category, 248 "threat_score": self.threat_score, 249 "detected_pattern": self.detected_pattern, 250 "mitigation_action": self.mitigation_action, 251 "sequence_number": self.sequence_number, 252 "envelope_hmac": self.envelope_hmac, 253 }
OCSF-aligned tamper-evident structured audit evaluation record.
Attributes: event_id (str): Unique UUID string for event identification. schema_version (str): Audit schema version specification. timestamp_utc (float): UTC timestamp in seconds since epoch. invocation_id (str): Unique trace context or invocation ID. trace_id (Optional[str]): OpenTelemetry trace ID. span_id (Optional[str]): OpenTelemetry span ID. subject_id (str): Caller subject identifier. tenant_id (str): Tenant isolation boundary identifier. subject_roles (List[str]): Roles assigned to caller. subject_clearance (int): Clearance level rank. enforcement_point (str): Enforcement point identifier. action (str): Action string requested. target_resource (str): Target resource path or name. decision (str): PDP decision outcome ('PERMIT', 'DENY', 'BLOCKED', etc.). reason_code (str): Standardized security event reason code. diagnostic_reason (str): Human-readable diagnostic description. severity (str): Event severity level string. payload_snapshot (Dict[str, Any]): Sanitized payload snapshot. threat_category (Optional[str]): Threat classification label. threat_score (Optional[float]): Threat confidence score. detected_pattern (Optional[str]): Injection pattern ID. mitigation_action (Optional[str]): Mitigation strategy applied. sequence_number (int): Monotonic audit sequence number. envelope_hmac (Optional[str]): Signature hash for tamper evidence.
AuditEvent( event_id: str, schema_version: str, timestamp_utc: float, invocation_id: str, trace_id: Optional[str], span_id: Optional[str], subject_id: str, tenant_id: str, subject_roles: List[str], subject_clearance: int, enforcement_point: str, action: str, target_resource: str, decision: str, reason_code: str, diagnostic_reason: str, severity: str, payload_snapshot: Dict[str, Any], threat_category: Optional[str] = None, threat_score: Optional[float] = None, detected_pattern: Optional[str] = None, mitigation_action: Optional[str] = None, sequence_number: int = 0, envelope_hmac: Optional[str] = None)
65 def __init__( 66 self, 67 event_id: str, 68 schema_version: str, 69 timestamp_utc: float, 70 invocation_id: str, 71 trace_id: Optional[str], 72 span_id: Optional[str], 73 subject_id: str, 74 tenant_id: str, 75 subject_roles: List[str], 76 subject_clearance: int, 77 enforcement_point: str, 78 action: str, 79 target_resource: str, 80 decision: str, 81 reason_code: str, 82 diagnostic_reason: str, 83 severity: str, 84 payload_snapshot: Dict[str, Any], 85 threat_category: Optional[str] = None, 86 threat_score: Optional[float] = None, 87 detected_pattern: Optional[str] = None, 88 mitigation_action: Optional[str] = None, 89 sequence_number: int = 0, 90 envelope_hmac: Optional[str] = None, 91 ): 92 """Initialize AuditEvent. 93 94 Args: 95 event_id: Unique UUID string for event identification. 96 schema_version: Audit schema version string. 97 timestamp_utc: UTC timestamp in seconds since epoch. 98 invocation_id: Unique trace context or invocation ID. 99 trace_id: OpenTelemetry trace ID if present. 100 span_id: OpenTelemetry span ID if present. 101 subject_id: Caller subject identifier. 102 tenant_id: Multi-tenant boundary identifier. 103 subject_roles: Roles assigned to caller. 104 subject_clearance: Clearance rank assigned to caller. 105 enforcement_point: Enforcement point identifier. 106 action: Requested action identifier. 107 target_resource: Target resource path or name. 108 decision: PDP decision verdict. 109 reason_code: Standardized security event reason code. 110 diagnostic_reason: Human-readable diagnostic description. 111 severity: Event severity level. 112 payload_snapshot: Sanitized payload snapshot dictionary. 113 threat_category: Optional threat classification category. 114 threat_score: Optional threat confidence score. 115 detected_pattern: Optional injection pattern identifier. 116 mitigation_action: Optional mitigation strategy applied. 117 sequence_number: Monotonic audit sequence number. 118 envelope_hmac: Optional HMAC signature for tamper evidence. 119 """ 120 self.event_id = event_id 121 self.schema_version = schema_version 122 self.timestamp_utc = timestamp_utc 123 self.invocation_id = invocation_id 124 self.trace_id = trace_id 125 self.span_id = span_id 126 self.subject_id = subject_id 127 self.tenant_id = tenant_id 128 self.subject_roles = list(subject_roles) 129 self.subject_clearance = subject_clearance 130 self.enforcement_point = enforcement_point 131 self.action = action 132 self.target_resource = target_resource 133 self.decision = decision 134 self.reason_code = reason_code 135 self.diagnostic_reason = diagnostic_reason 136 self.severity = severity 137 self.payload_snapshot = dict(payload_snapshot) 138 self.threat_category = threat_category 139 self.threat_score = threat_score 140 self.detected_pattern = detected_pattern 141 self.mitigation_action = mitigation_action 142 self.sequence_number = sequence_number 143 self.envelope_hmac = envelope_hmac
Initialize AuditEvent.
Args: event_id: Unique UUID string for event identification. schema_version: Audit schema version string. timestamp_utc: UTC timestamp in seconds since epoch. invocation_id: Unique trace context or invocation ID. trace_id: OpenTelemetry trace ID if present. span_id: OpenTelemetry span ID if present. subject_id: Caller subject identifier. tenant_id: Multi-tenant boundary identifier. subject_roles: Roles assigned to caller. subject_clearance: Clearance rank assigned to caller. enforcement_point: Enforcement point identifier. action: Requested action identifier. target_resource: Target resource path or name. decision: PDP decision verdict. reason_code: Standardized security event reason code. diagnostic_reason: Human-readable diagnostic description. severity: Event severity level. payload_snapshot: Sanitized payload snapshot dictionary. threat_category: Optional threat classification category. threat_score: Optional threat confidence score. detected_pattern: Optional injection pattern identifier. mitigation_action: Optional mitigation strategy applied. sequence_number: Monotonic audit sequence number. envelope_hmac: Optional HMAC signature for tamper evidence.
@classmethod
def
create( cls, subject_id: str, tenant_id: str, action: str, resource: str, decision: str, diagnostic_reason: str, severity: str, metadata: Dict[str, Any], enforcement_point: str = 'SYSTEM_BOUNDARY', reason_code: str = 'POLICY_EVALUATION'):
145 @classmethod 146 def create( 147 cls, 148 subject_id: str, 149 tenant_id: str, 150 action: str, 151 resource: str, 152 decision: str, 153 diagnostic_reason: str, 154 severity: str, 155 metadata: Dict[str, Any], 156 enforcement_point: str = "SYSTEM_BOUNDARY", 157 reason_code: str = "POLICY_EVALUATION", 158 ): 159 """Factory method to construct an AuditEvent with OpenTelemetry trace extraction. 160 161 Args: 162 subject_id: Subject identifier. 163 tenant_id: Tenant ID. 164 action: Requested action. 165 resource: Resource path/name. 166 decision: Decision verdict. 167 diagnostic_reason: Rationale description. 168 severity: Severity level. 169 metadata: Payload snapshot dictionary. 170 enforcement_point: PEP point ID. 171 reason_code: Event reason code. 172 173 Returns: 174 AuditEvent: Constructed audit event instance. 175 """ 176 trace_id = metadata.get("trace_id") 177 span_id = metadata.get("span_id") 178 if trace_id is None or span_id is None: 179 try: 180 from opentelemetry import trace 181 current_span = trace.get_current_span() 182 if current_span and current_span.get_span_context().is_valid: 183 ctx = current_span.get_span_context() 184 if trace_id is None: 185 trace_id = f"{ctx.trace_id:032x}" 186 if span_id is None: 187 span_id = f"{ctx.span_id:016x}" 188 except Exception: 189 pass 190 191 return cls( 192 event_id=str(uuid.uuid4()), 193 schema_version=DEFAULT_SCHEMA_VERSION, 194 timestamp_utc=time.time(), 195 invocation_id=metadata.get("invocation_id", str(uuid.uuid4())), 196 trace_id=trace_id, 197 span_id=span_id, 198 subject_id=subject_id, 199 tenant_id=tenant_id, 200 subject_roles=metadata.get("subject_roles", []), 201 subject_clearance=int(metadata.get("subject_clearance", 0)), 202 enforcement_point=enforcement_point, 203 action=action, 204 target_resource=resource, 205 decision=decision, 206 reason_code=reason_code, 207 diagnostic_reason=diagnostic_reason, 208 severity=severity, 209 payload_snapshot=metadata, 210 )
Factory method to construct an AuditEvent with OpenTelemetry trace extraction.
Args: subject_id: Subject identifier. tenant_id: Tenant ID. action: Requested action. resource: Resource path/name. decision: Decision verdict. diagnostic_reason: Rationale description. severity: Severity level. metadata: Payload snapshot dictionary. enforcement_point: PEP point ID. reason_code: Event reason code.
Returns: AuditEvent: Constructed audit event instance.
resource: str
212 @property 213 def resource(self) -> str: 214 """Get target resource identifier.""" 215 return self.target_resource
Get target resource identifier.
metadata: Dict[str, Any]
217 @property 218 def metadata(self) -> Dict[str, Any]: 219 """Get payload snapshot dictionary.""" 220 return self.payload_snapshot
Get payload snapshot dictionary.
def
to_dict(self) -> Dict[str, Any]:
222 def to_dict(self) -> Dict[str, Any]: 223 """Serialize AuditEvent into dictionary format. 224 225 Returns: 226 Dict[str, Any]: JSON-compatible event dictionary. 227 """ 228 return { 229 "event_id": self.event_id, 230 "schema_version": self.schema_version, 231 "timestamp_utc": self.timestamp_utc, 232 "invocation_id": self.invocation_id, 233 "trace_id": self.trace_id, 234 "span_id": self.span_id, 235 "subject_id": self.subject_id, 236 "tenant_id": self.tenant_id, 237 "subject_roles": self.subject_roles, 238 "subject_clearance": self.subject_clearance, 239 "enforcement_point": self.enforcement_point, 240 "action": self.action, 241 "target_resource": self.target_resource, 242 "decision": self.decision, 243 "reason_code": self.reason_code, 244 "diagnostic_reason": self.diagnostic_reason, 245 "severity": self.severity, 246 "payload_snapshot": self.payload_snapshot, 247 "threat_category": self.threat_category, 248 "threat_score": self.threat_score, 249 "detected_pattern": self.detected_pattern, 250 "mitigation_action": self.mitigation_action, 251 "sequence_number": self.sequence_number, 252 "envelope_hmac": self.envelope_hmac, 253 }
Serialize AuditEvent into dictionary format.
Returns: Dict[str, Any]: JSON-compatible event dictionary.
class
AuditManager:
547class AuditManager: 548 """Central manager for non-blocking asynchronous audit event dispatch and ring buffer management.""" 549 550 def __init__(self, publishers: List[AuditPublisherSPI], *args: Any, **kwargs: Any): 551 """Initialize AuditManager. 552 553 Args: 554 publishers: List of AuditPublisherSPI publisher implementations. 555 *args: Optional positional arguments (capacity). 556 **kwargs: Optional keyword arguments (capacity). 557 """ 558 capacity = int(kwargs.get("capacity", args[0] if len(args) > 0 else DEFAULT_RING_BUFFER_CAPACITY)) 559 self._publishers = list(publishers) 560 self._capacity = capacity 561 self._buffer: deque = deque() 562 self._dropped_count = 0 563 self._async_queue: Optional[Any] = None 564 self._lock = threading.Lock() 565 self._queue_condition = threading.Condition(self._lock) 566 self._stop_event = threading.Event() 567 self._worker_thread = threading.Thread( 568 target=self._worker_loop, 569 name="AuditManagerAsyncWorker", 570 daemon=True, 571 ) 572 self._worker_thread.start() 573 574 @classmethod 575 def create(cls, publishers: List[AuditPublisherSPI], *args: Any, **kwargs: Any): 576 """Factory method to construct AuditManager. 577 578 Args: 579 publishers: List of AuditPublisherSPI publisher implementations. 580 *args: Optional positional arguments (capacity). 581 **kwargs: Optional keyword arguments (capacity). 582 583 Returns: 584 AuditManager: Initialized manager instance with background async worker thread. 585 """ 586 capacity = int(kwargs.get("capacity", args[0] if len(args) > 0 else DEFAULT_RING_BUFFER_CAPACITY)) 587 return cls(publishers=publishers, capacity=capacity) 588 589 @property 590 def publishers(self) -> List[AuditPublisherSPI]: 591 """Get copy of registered audit publisher SPIs.""" 592 return list(self._publishers) 593 594 @property 595 def dropped_count(self) -> int: 596 """Get count of audit events dropped due to buffer overflow.""" 597 with self._lock: 598 return self._dropped_count 599 600 def _worker_loop(self) -> None: 601 """Background worker thread draining events asynchronously from the bounded queue.""" 602 while not self._stop_event.is_set(): 603 event = None 604 with self._queue_condition: 605 while not self._buffer and not self._stop_event.is_set(): 606 self._queue_condition.wait(timeout=0.1) 607 if self._buffer: 608 event = self._buffer.popleft() 609 if event is not None: 610 self._dispatch_to_publishers(event) 611 612 # Drain remaining events on shutdown 613 while True: 614 event = None 615 with self._lock: 616 if self._buffer: 617 event = self._buffer.popleft() 618 if event is None: 619 break 620 self._dispatch_to_publishers(event) 621 622 def _dispatch_to_publishers(self, event: AuditEvent) -> None: 623 """Dispatch event to registered publishers catching exceptions safely.""" 624 for pub in self._publishers: 625 if not getattr(pub, "is_immediate", False): 626 try: 627 pub.publish(event) 628 except Exception: 629 pass 630 631 def record_event(self, event: AuditEvent) -> None: 632 """Execute in-flight sanitization and non-blocking dispatch to ring buffer. 633 634 Args: 635 event: AuditEvent instance to sanitize and record. 636 """ 637 if get_pii_scope() != PII_SCOPE_DISABLED: 638 ctx = PiiContext.create( 639 PII_STAGE_AUDIT_PAYLOAD, 640 event.tenant_id, 641 event.subject_id, 642 event.target_resource, 643 {"action": event.action, "enforcement_point": event.enforcement_point}, 644 ) 645 sanitizer = get_pii_sanitizer() 646 if isinstance(event.payload_snapshot, dict): 647 event.payload_snapshot = sanitizer.sanitize_payload(event.payload_snapshot, ctx) 648 if isinstance(event.diagnostic_reason, str): 649 event.diagnostic_reason = sanitizer.sanitize_text(event.diagnostic_reason, ctx, DEFAULT_PAYLOAD_MAX_CHARS) 650 651 # Immediate dispatch for synchronous publishers (e.g. test collectors) 652 for pub in self._publishers: 653 if getattr(pub, "is_immediate", False): 654 try: 655 pub.publish(event) 656 except Exception: 657 pass 658 659 with self._queue_condition: 660 if len(self._buffer) >= self._capacity: 661 self._buffer.popleft() # DROP_OLDEST_WITH_METRIC 662 self._dropped_count += 1 663 self._buffer.append(event) 664 self._queue_condition.notify() 665 666 if self._async_queue: 667 try: 668 self._async_queue.put_nowait(event) 669 except Exception: 670 pass 671 672 def publish(self, event: AuditEvent) -> None: 673 """Alias for record_event. 674 675 Args: 676 event: AuditEvent instance. 677 """ 678 self.record_event(event) 679 680 def flush(self) -> None: 681 """Flush queued events and all registered audit publishers.""" 682 # Wait until internal bounded buffer is completely drained by worker thread 683 deadline = time.time() + 2.0 684 while time.time() < deadline: 685 with self._lock: 686 if not self._buffer: 687 break 688 time.sleep(0.005) 689 690 for pub in self._publishers: 691 try: 692 pub.flush() 693 except Exception: 694 pass 695 696 def close(self) -> None: 697 """Flush and close all registered audit publishers, stopping worker thread.""" 698 self._stop_event.set() 699 with self._queue_condition: 700 self._queue_condition.notify_all() 701 if self._worker_thread.is_alive(): 702 self._worker_thread.join(timeout=1.0) 703 704 self.flush() 705 for pub in self._publishers: 706 try: 707 pub.close() 708 except Exception: 709 pass
Central manager for non-blocking asynchronous audit event dispatch and ring buffer management.
AuditManager( publishers: List[AuditPublisherSPI], *args: Any, **kwargs: Any)
550 def __init__(self, publishers: List[AuditPublisherSPI], *args: Any, **kwargs: Any): 551 """Initialize AuditManager. 552 553 Args: 554 publishers: List of AuditPublisherSPI publisher implementations. 555 *args: Optional positional arguments (capacity). 556 **kwargs: Optional keyword arguments (capacity). 557 """ 558 capacity = int(kwargs.get("capacity", args[0] if len(args) > 0 else DEFAULT_RING_BUFFER_CAPACITY)) 559 self._publishers = list(publishers) 560 self._capacity = capacity 561 self._buffer: deque = deque() 562 self._dropped_count = 0 563 self._async_queue: Optional[Any] = None 564 self._lock = threading.Lock() 565 self._queue_condition = threading.Condition(self._lock) 566 self._stop_event = threading.Event() 567 self._worker_thread = threading.Thread( 568 target=self._worker_loop, 569 name="AuditManagerAsyncWorker", 570 daemon=True, 571 ) 572 self._worker_thread.start()
Initialize AuditManager.
Args: publishers: List of AuditPublisherSPI publisher implementations. *args: Optional positional arguments (capacity). **kwargs: Optional keyword arguments (capacity).
574 @classmethod 575 def create(cls, publishers: List[AuditPublisherSPI], *args: Any, **kwargs: Any): 576 """Factory method to construct AuditManager. 577 578 Args: 579 publishers: List of AuditPublisherSPI publisher implementations. 580 *args: Optional positional arguments (capacity). 581 **kwargs: Optional keyword arguments (capacity). 582 583 Returns: 584 AuditManager: Initialized manager instance with background async worker thread. 585 """ 586 capacity = int(kwargs.get("capacity", args[0] if len(args) > 0 else DEFAULT_RING_BUFFER_CAPACITY)) 587 return cls(publishers=publishers, capacity=capacity)
Factory method to construct AuditManager.
Args: publishers: List of AuditPublisherSPI publisher implementations. *args: Optional positional arguments (capacity). **kwargs: Optional keyword arguments (capacity).
Returns: AuditManager: Initialized manager instance with background async worker thread.
publishers: List[AuditPublisherSPI]
589 @property 590 def publishers(self) -> List[AuditPublisherSPI]: 591 """Get copy of registered audit publisher SPIs.""" 592 return list(self._publishers)
Get copy of registered audit publisher SPIs.
dropped_count: int
594 @property 595 def dropped_count(self) -> int: 596 """Get count of audit events dropped due to buffer overflow.""" 597 with self._lock: 598 return self._dropped_count
Get count of audit events dropped due to buffer overflow.
631 def record_event(self, event: AuditEvent) -> None: 632 """Execute in-flight sanitization and non-blocking dispatch to ring buffer. 633 634 Args: 635 event: AuditEvent instance to sanitize and record. 636 """ 637 if get_pii_scope() != PII_SCOPE_DISABLED: 638 ctx = PiiContext.create( 639 PII_STAGE_AUDIT_PAYLOAD, 640 event.tenant_id, 641 event.subject_id, 642 event.target_resource, 643 {"action": event.action, "enforcement_point": event.enforcement_point}, 644 ) 645 sanitizer = get_pii_sanitizer() 646 if isinstance(event.payload_snapshot, dict): 647 event.payload_snapshot = sanitizer.sanitize_payload(event.payload_snapshot, ctx) 648 if isinstance(event.diagnostic_reason, str): 649 event.diagnostic_reason = sanitizer.sanitize_text(event.diagnostic_reason, ctx, DEFAULT_PAYLOAD_MAX_CHARS) 650 651 # Immediate dispatch for synchronous publishers (e.g. test collectors) 652 for pub in self._publishers: 653 if getattr(pub, "is_immediate", False): 654 try: 655 pub.publish(event) 656 except Exception: 657 pass 658 659 with self._queue_condition: 660 if len(self._buffer) >= self._capacity: 661 self._buffer.popleft() # DROP_OLDEST_WITH_METRIC 662 self._dropped_count += 1 663 self._buffer.append(event) 664 self._queue_condition.notify() 665 666 if self._async_queue: 667 try: 668 self._async_queue.put_nowait(event) 669 except Exception: 670 pass
Execute in-flight sanitization and non-blocking dispatch to ring buffer.
Args: event: AuditEvent instance to sanitize and record.
672 def publish(self, event: AuditEvent) -> None: 673 """Alias for record_event. 674 675 Args: 676 event: AuditEvent instance. 677 """ 678 self.record_event(event)
Alias for record_event.
Args: event: AuditEvent instance.
def
flush(self) -> None:
680 def flush(self) -> None: 681 """Flush queued events and all registered audit publishers.""" 682 # Wait until internal bounded buffer is completely drained by worker thread 683 deadline = time.time() + 2.0 684 while time.time() < deadline: 685 with self._lock: 686 if not self._buffer: 687 break 688 time.sleep(0.005) 689 690 for pub in self._publishers: 691 try: 692 pub.flush() 693 except Exception: 694 pass
Flush queued events and all registered audit publishers.
def
close(self) -> None:
696 def close(self) -> None: 697 """Flush and close all registered audit publishers, stopping worker thread.""" 698 self._stop_event.set() 699 with self._queue_condition: 700 self._queue_condition.notify_all() 701 if self._worker_thread.is_alive(): 702 self._worker_thread.join(timeout=1.0) 703 704 self.flush() 705 for pub in self._publishers: 706 try: 707 pub.close() 708 except Exception: 709 pass
Flush and close all registered audit publishers, stopping worker thread.
class
AuditPublisherSPI(abc.ABC):
341class AuditPublisherSPI(ABC): 342 """SPI interface for enterprise SIEM and audit publishers.""" 343 344 @property 345 def is_immediate(self) -> bool: 346 """Indicate whether events should be dispatched immediately to this publisher.""" 347 return False 348 349 @abstractmethod 350 def publish(self, event: AuditEvent) -> None: 351 """Publish a single AuditEvent. 352 353 Args: 354 event: AuditEvent to publish. 355 """ 356 pass 357 358 def publish_batch(self, events: List[AuditEvent]) -> None: 359 """Publish a batch of AuditEvents. 360 361 Args: 362 events: List of AuditEvents. 363 """ 364 for event in events: 365 self.publish(event) 366 367 def flush(self) -> None: 368 """Flush buffered log streams.""" 369 pass 370 371 def close(self) -> None: 372 """Release publisher resources.""" 373 pass
SPI interface for enterprise SIEM and audit publishers.
is_immediate: bool
344 @property 345 def is_immediate(self) -> bool: 346 """Indicate whether events should be dispatched immediately to this publisher.""" 347 return False
Indicate whether events should be dispatched immediately to this publisher.
349 @abstractmethod 350 def publish(self, event: AuditEvent) -> None: 351 """Publish a single AuditEvent. 352 353 Args: 354 event: AuditEvent to publish. 355 """ 356 pass
Publish a single AuditEvent.
Args: event: AuditEvent to publish.
358 def publish_batch(self, events: List[AuditEvent]) -> None: 359 """Publish a batch of AuditEvents. 360 361 Args: 362 events: List of AuditEvents. 363 """ 364 for event in events: 365 self.publish(event)
Publish a batch of AuditEvents.
Args: events: List of AuditEvents.
class
AuditSanitizer:
259class AuditSanitizer: 260 """3-stage in-flight PII and secret scrubber pipeline delegating to active PiiSanitizerSPI.""" 261 262 SECRET_REGEX = BuiltInPiiSanitizer.COMBINED_SECRET_REGEX 263 264 @classmethod 265 def create(cls, *args: Any, **kwargs: Any) -> "AuditSanitizer": 266 """Construct an AuditSanitizer and register any provided masked_fields. 267 268 Args: 269 *args: Optional positional arguments (masked_fields). 270 **kwargs: Optional keyword arguments (masked_fields). 271 272 Returns: 273 AuditSanitizer: New sanitizer wrapper instance. 274 """ 275 sanitizer = get_pii_sanitizer() 276 masked_fields = kwargs.get("masked_fields", args[0] if len(args) > 0 else None) 277 if masked_fields and hasattr(sanitizer, "register_masked_field"): 278 for field_name in masked_fields: 279 sanitizer.register_masked_field(str(field_name)) 280 return cls() 281 282 @classmethod 283 def register_masked_field(cls, field_name: str) -> None: 284 """Stage 1: Register a parameter or field name for automatic @Masked annotation redaction. 285 286 Args: 287 field_name: Parameter or field name to register for dictionary masking. 288 """ 289 sanitizer = get_pii_sanitizer() 290 if hasattr(sanitizer, "register_masked_field"): 291 sanitizer.register_masked_field(field_name) 292 293 @classmethod 294 def is_masked_field(cls, field_name: str) -> bool: 295 """Check if a field name is marked as sensitive/masked. 296 297 Args: 298 field_name: Parameter or field name to check. 299 300 Returns: 301 bool: True if field is registered for masking, False otherwise. 302 """ 303 sanitizer = get_pii_sanitizer() 304 if hasattr(sanitizer, "is_masked_field"): 305 return sanitizer.is_masked_field(field_name) 306 return False 307 308 @classmethod 309 def sanitize_string(cls, value: str, max_chars: int) -> str: 310 """Apply Stage 2 regex scrubbing and Stage 3 truncation to a string value. 311 312 Args: 313 value: Input string to sanitize. 314 max_chars: Character length cap before applying truncation suffix. 315 316 Returns: 317 str: Sanitized and optionally truncated string. 318 """ 319 sanitizer = get_pii_sanitizer() 320 context = PiiContext.create(PII_STAGE_AUDIT_PAYLOAD, None, None, None, None) 321 return sanitizer.sanitize_text(value, context, max_chars) 322 323 @classmethod 324 def sanitize_payload(cls, payload: Dict[str, Any], *args: Any, **kwargs: Any) -> Dict[str, Any]: 325 """Scrub secrets, PII, @Masked fields, and truncate oversized strings from payload dictionary. 326 327 Args: 328 payload: Input dictionary containing audit event data or telemetry. 329 *args: Optional positional arguments (max_chars, masked_fields). 330 **kwargs: Optional keyword arguments (max_chars, masked_fields). 331 332 Returns: 333 Dict[str, Any]: Sanitized dictionary payload. 334 """ 335 sanitizer = get_pii_sanitizer() 336 context = PiiContext.create(PII_STAGE_AUDIT_PAYLOAD, None, None, None, None) 337 return sanitizer.sanitize_payload(payload, context, *args, **kwargs)
3-stage in-flight PII and secret scrubber pipeline delegating to active PiiSanitizerSPI.
SECRET_REGEX =
re.compile('(?P<db_uri>(?:postgres(?:ql)?|mysql|mongodb(?:\\+srv)?|redis)://[^:\\s/]+:[^@\\s/]+@[^\\s,\\\'"]+)|(?P<kv_secret>(?:bearer[\\s=]+[^\\s,\\\'"]+|password\\s*[:=]\\s*[^\\s,\\\'"]+|secret\\s*[:=]\\s*[^\\, re.IGNORECASE)
264 @classmethod 265 def create(cls, *args: Any, **kwargs: Any) -> "AuditSanitizer": 266 """Construct an AuditSanitizer and register any provided masked_fields. 267 268 Args: 269 *args: Optional positional arguments (masked_fields). 270 **kwargs: Optional keyword arguments (masked_fields). 271 272 Returns: 273 AuditSanitizer: New sanitizer wrapper instance. 274 """ 275 sanitizer = get_pii_sanitizer() 276 masked_fields = kwargs.get("masked_fields", args[0] if len(args) > 0 else None) 277 if masked_fields and hasattr(sanitizer, "register_masked_field"): 278 for field_name in masked_fields: 279 sanitizer.register_masked_field(str(field_name)) 280 return cls()
Construct an AuditSanitizer and register any provided masked_fields.
Args: *args: Optional positional arguments (masked_fields). **kwargs: Optional keyword arguments (masked_fields).
Returns: AuditSanitizer: New sanitizer wrapper instance.
@classmethod
def
register_masked_field(cls, field_name: str) -> None:
282 @classmethod 283 def register_masked_field(cls, field_name: str) -> None: 284 """Stage 1: Register a parameter or field name for automatic @Masked annotation redaction. 285 286 Args: 287 field_name: Parameter or field name to register for dictionary masking. 288 """ 289 sanitizer = get_pii_sanitizer() 290 if hasattr(sanitizer, "register_masked_field"): 291 sanitizer.register_masked_field(field_name)
Stage 1: Register a parameter or field name for automatic @Masked annotation redaction.
Args: field_name: Parameter or field name to register for dictionary masking.
@classmethod
def
is_masked_field(cls, field_name: str) -> bool:
293 @classmethod 294 def is_masked_field(cls, field_name: str) -> bool: 295 """Check if a field name is marked as sensitive/masked. 296 297 Args: 298 field_name: Parameter or field name to check. 299 300 Returns: 301 bool: True if field is registered for masking, False otherwise. 302 """ 303 sanitizer = get_pii_sanitizer() 304 if hasattr(sanitizer, "is_masked_field"): 305 return sanitizer.is_masked_field(field_name) 306 return False
Check if a field name is marked as sensitive/masked.
Args: field_name: Parameter or field name to check.
Returns: bool: True if field is registered for masking, False otherwise.
@classmethod
def
sanitize_string(cls, value: str, max_chars: int) -> str:
308 @classmethod 309 def sanitize_string(cls, value: str, max_chars: int) -> str: 310 """Apply Stage 2 regex scrubbing and Stage 3 truncation to a string value. 311 312 Args: 313 value: Input string to sanitize. 314 max_chars: Character length cap before applying truncation suffix. 315 316 Returns: 317 str: Sanitized and optionally truncated string. 318 """ 319 sanitizer = get_pii_sanitizer() 320 context = PiiContext.create(PII_STAGE_AUDIT_PAYLOAD, None, None, None, None) 321 return sanitizer.sanitize_text(value, context, max_chars)
Apply Stage 2 regex scrubbing and Stage 3 truncation to a string value.
Args: value: Input string to sanitize. max_chars: Character length cap before applying truncation suffix.
Returns: str: Sanitized and optionally truncated string.
@classmethod
def
sanitize_payload( cls, payload: Dict[str, Any], *args: Any, **kwargs: Any) -> Dict[str, Any]:
323 @classmethod 324 def sanitize_payload(cls, payload: Dict[str, Any], *args: Any, **kwargs: Any) -> Dict[str, Any]: 325 """Scrub secrets, PII, @Masked fields, and truncate oversized strings from payload dictionary. 326 327 Args: 328 payload: Input dictionary containing audit event data or telemetry. 329 *args: Optional positional arguments (max_chars, masked_fields). 330 **kwargs: Optional keyword arguments (max_chars, masked_fields). 331 332 Returns: 333 Dict[str, Any]: Sanitized dictionary payload. 334 """ 335 sanitizer = get_pii_sanitizer() 336 context = PiiContext.create(PII_STAGE_AUDIT_PAYLOAD, None, None, None, None) 337 return sanitizer.sanitize_payload(payload, context, *args, **kwargs)
Scrub secrets, PII, @Masked fields, and truncate oversized strings from payload dictionary.
Args: payload: Input dictionary containing audit event data or telemetry. *args: Optional positional arguments (max_chars, masked_fields). **kwargs: Optional keyword arguments (max_chars, masked_fields).
Returns: Dict[str, Any]: Sanitized dictionary payload.
376class ConsoleAuditLogger(AuditPublisherSPI): 377 """Console logger implementation of AuditPublisherSPI.""" 378 379 def publish(self, event: AuditEvent) -> None: 380 """Print audit log line to standard output. 381 382 Args: 383 event: AuditEvent to publish. 384 """ 385 print(f"[AUDIT LOG] [{event.severity}] {event.action} by {event.subject_id} -> {event.decision} ({event.diagnostic_reason})")
Console logger implementation of AuditPublisherSPI.
379 def publish(self, event: AuditEvent) -> None: 380 """Print audit log line to standard output. 381 382 Args: 383 event: AuditEvent to publish. 384 """ 385 print(f"[AUDIT LOG] [{event.severity}] {event.action} by {event.subject_id} -> {event.decision} ({event.diagnostic_reason})")
Print audit log line to standard output.
Args: event: AuditEvent to publish.
410class JsonFileAuditLogger(AuditPublisherSPI): 411 """File logger writing JSON audit events to specified log path with secret masking.""" 412 413 def __init__(self, file_path: str): 414 """Initialize JsonFileAuditLogger. 415 416 Args: 417 file_path: Output JSON file path. 418 """ 419 self.file_path = file_path 420 421 @classmethod 422 def create(cls, file_path: str): 423 """Factory method creating JsonFileAuditLogger. 424 425 Args: 426 file_path: Target log file path. 427 428 Returns: 429 JsonFileAuditLogger: New logger instance. 430 """ 431 return cls(file_path=file_path) 432 433 def publish(self, event: AuditEvent) -> None: 434 """Append sanitized JSON log line to target log file. 435 436 Args: 437 event: AuditEvent to append. 438 """ 439 sanitized_data = AuditSanitizer.sanitize_payload(event.to_dict()) 440 data_str = json.dumps(sanitized_data) 441 dir_path = os.path.dirname(self.file_path) 442 if dir_path: 443 os.makedirs(dir_path, exist_ok=True) 444 with open(self.file_path, "a", encoding="utf-8") as f: 445 f.write(data_str + "\n")
File logger writing JSON audit events to specified log path with secret masking.
JsonFileAuditLogger(file_path: str)
413 def __init__(self, file_path: str): 414 """Initialize JsonFileAuditLogger. 415 416 Args: 417 file_path: Output JSON file path. 418 """ 419 self.file_path = file_path
Initialize JsonFileAuditLogger.
Args: file_path: Output JSON file path.
@classmethod
def
create(cls, file_path: str):
421 @classmethod 422 def create(cls, file_path: str): 423 """Factory method creating JsonFileAuditLogger. 424 425 Args: 426 file_path: Target log file path. 427 428 Returns: 429 JsonFileAuditLogger: New logger instance. 430 """ 431 return cls(file_path=file_path)
Factory method creating JsonFileAuditLogger.
Args: file_path: Target log file path.
Returns: JsonFileAuditLogger: New logger instance.
433 def publish(self, event: AuditEvent) -> None: 434 """Append sanitized JSON log line to target log file. 435 436 Args: 437 event: AuditEvent to append. 438 """ 439 sanitized_data = AuditSanitizer.sanitize_payload(event.to_dict()) 440 data_str = json.dumps(sanitized_data) 441 dir_path = os.path.dirname(self.file_path) 442 if dir_path: 443 os.makedirs(dir_path, exist_ok=True) 444 with open(self.file_path, "a", encoding="utf-8") as f: 445 f.write(data_str + "\n")
Append sanitized JSON log line to target log file.
Args: event: AuditEvent to append.
389class PythonLoggingAuditPublisher(AuditPublisherSPI): 390 """Publishes single-line JSON log lines to Python logging logger 'audit.security.langchain'.""" 391 392 def __init__(self, logger_name: str = "audit.security.langchain"): 393 """Initialize PythonLoggingAuditPublisher. 394 395 Args: 396 logger_name: Python logger name. 397 """ 398 self.logger = logging.getLogger(logger_name) 399 400 def publish(self, event: AuditEvent) -> None: 401 """Emit sanitized JSON audit record via standard Python logger. 402 403 Args: 404 event: AuditEvent to emit. 405 """ 406 data = AuditSanitizer.sanitize_payload(event.to_dict()) 407 self.logger.info(json.dumps(data))
Publishes single-line JSON log lines to Python logging logger 'audit.security.langchain'.
PythonLoggingAuditPublisher(logger_name: str = 'audit.security.langchain')
392 def __init__(self, logger_name: str = "audit.security.langchain"): 393 """Initialize PythonLoggingAuditPublisher. 394 395 Args: 396 logger_name: Python logger name. 397 """ 398 self.logger = logging.getLogger(logger_name)
Initialize PythonLoggingAuditPublisher.
Args: logger_name: Python logger name.
400 def publish(self, event: AuditEvent) -> None: 401 """Emit sanitized JSON audit record via standard Python logger. 402 403 Args: 404 event: AuditEvent to emit. 405 """ 406 data = AuditSanitizer.sanitize_payload(event.to_dict()) 407 self.logger.info(json.dumps(data))
Emit sanitized JSON audit record via standard Python logger.
Args: event: AuditEvent to emit.
class
PromptInjectionScanner:
57class PromptInjectionScanner: 58 """Rule-based heuristic scanner for prompt injection, jailbreak, and instruction overrides.""" 59 60 INJECTION_PATTERNS = [ 61 ( 62 "JAILBREAK_SYSTEM_OVERRIDE", 63 re.compile( 64 r"(ignore\s+all\s+(previous|prior)\s+instructions|disregard\s+(all\s+)?previous\s+rules|system\s*:\s*you\s+are\s+now|forget\s+your\s+safety\s+guidelines)", 65 re.IGNORECASE, 66 ), 67 0.95, 68 ), 69 ( 70 "ROLEPLAY_DAN_EVASION", 71 re.compile( 72 r"(do\s+anything\s+now|dan\s+mode|jailbroken|unrestricted\s+ai|act\s+as\s+an\s+unfiltered)", 73 re.IGNORECASE, 74 ), 75 0.90, 76 ), 77 ( 78 "INSTRUCTION_HIJACKING", 79 re.compile( 80 r"(\[system\]|###\s*instruction|override_policy\s*=\s*true|<\|im_start\|>system)", 81 re.IGNORECASE, 82 ), 83 0.90, 84 ), 85 ] 86 """Pre-compiled regular expression rules and confidence weights for injection scanning.""" 87 88 def __init__(self, audit_manager: Optional[AuditManager] = None): 89 """Initialize PromptInjectionScanner. 90 91 Args: 92 audit_manager: Optional AuditManager for recording injection alerts. 93 """ 94 self._audit_manager = audit_manager 95 96 @classmethod 97 def create(cls, audit_manager: Optional[AuditManager] = None): 98 """Factory method to construct PromptInjectionScanner. 99 100 Args: 101 audit_manager: Optional AuditManager instance. 102 103 Returns: 104 PromptInjectionScanner: New scanner instance. 105 """ 106 return cls(audit_manager=audit_manager) 107 108 def scan_text(self, text: str) -> PromptInjectionScanResult: 109 """Scan input text string for prompt injection and jailbreak signatures. 110 111 Args: 112 text: Input text string to scan. 113 114 Returns: 115 PromptInjectionScanResult: Scan result containing match status and threat details. 116 """ 117 if not text or not isinstance(text, str): 118 return PromptInjectionScanResult(is_detected=False) 119 120 for pattern_id, regex, score in self.INJECTION_PATTERNS: 121 match = regex.search(text) 122 if match: 123 snippet = match.group(0) 124 category = "INSTRUCTION_OVERRIDE" if "INSTRUCTION" in pattern_id else "PROMPT_INJECTION" 125 return PromptInjectionScanResult( 126 is_detected=True, 127 pattern_id=pattern_id, 128 threat_score=score, 129 threat_category=category, 130 detected_text=snippet, 131 ) 132 133 return PromptInjectionScanResult(is_detected=False) 134 135 def scan_and_audit( 136 self, 137 text: str, 138 subject_id: str, 139 tenant_id: str, 140 enforcement_point: str = PEP_AGENT_GUARD, 141 context_metadata: Optional[Dict[str, Any]] = None, 142 ) -> PromptInjectionScanResult: 143 """Scan input text and publish audit security alert event if threat is detected. 144 145 Args: 146 text: Input text string. 147 subject_id: Subject attempting execution. 148 tenant_id: Tenant identifier. 149 enforcement_point: Target PEP identifier string. 150 context_metadata: Optional additional metadata dictionary. 151 152 Returns: 153 PromptInjectionScanResult: Scan result envelope. 154 """ 155 result = self.scan_text(text) 156 if result.is_detected and self._audit_manager: 157 event_type = ( 158 EVENT_INSTRUCTION_OVERRIDE_ATTEMPT 159 if result.threat_category == "INSTRUCTION_OVERRIDE" 160 else EVENT_PROMPT_INJECTION_DETECTED 161 ) 162 severity = ( 163 SEVERITY_CRITICAL_ALERT 164 if result.threat_score >= 0.9 165 else SEVERITY_SECURITY_ALERT 166 ) 167 168 meta = dict(context_metadata) if context_metadata else {} 169 meta.update({ 170 "detected_snippet": result.detected_text, 171 "threat_score": result.threat_score, 172 "pattern_id": result.pattern_id, 173 }) 174 175 event = AuditEvent.create( 176 subject_id=subject_id, 177 tenant_id=tenant_id, 178 action="prompt:scan", 179 resource=enforcement_point, 180 decision="BLOCKED", 181 diagnostic_reason=f"Prompt injection pattern detected: '{result.pattern_id}'", 182 severity=severity, 183 metadata=meta, 184 enforcement_point=enforcement_point, 185 reason_code=event_type, 186 ) 187 event.threat_category = result.threat_category 188 event.threat_score = result.threat_score 189 event.detected_pattern = result.pattern_id 190 event.mitigation_action = STRATEGY_HARD_ABORT 191 192 self._audit_manager.record_event(event) 193 194 return result
Rule-based heuristic scanner for prompt injection, jailbreak, and instruction overrides.
PromptInjectionScanner( audit_manager: Optional[AuditManager] = None)
88 def __init__(self, audit_manager: Optional[AuditManager] = None): 89 """Initialize PromptInjectionScanner. 90 91 Args: 92 audit_manager: Optional AuditManager for recording injection alerts. 93 """ 94 self._audit_manager = audit_manager
Initialize PromptInjectionScanner.
Args: audit_manager: Optional AuditManager for recording injection alerts.
INJECTION_PATTERNS =
[('JAILBREAK_SYSTEM_OVERRIDE', re.compile('(ignore\\s+all\\s+(previous|prior)\\s+instructions|disregard\\s+(all\\s+)?previous\\s+rules|system\\s*:\\s*you\\s+are\\s+now|forget\\s+your\\s+safety\\s+guidelines)', re.IGNORECASE), 0.95), ('ROLEPLAY_DAN_EVASION', re.compile('(do\\s+anything\\s+now|dan\\s+mode|jailbroken|unrestricted\\s+ai|act\\s+as\\s+an\\s+unfiltered)', re.IGNORECASE), 0.9), ('INSTRUCTION_HIJACKING', re.compile('(\\[system\\]|###\\s*instruction|override_policy\\s*=\\s*true|<\\|im_start\\|>system)', re.IGNORECASE), 0.9)]
Pre-compiled regular expression rules and confidence weights for injection scanning.
96 @classmethod 97 def create(cls, audit_manager: Optional[AuditManager] = None): 98 """Factory method to construct PromptInjectionScanner. 99 100 Args: 101 audit_manager: Optional AuditManager instance. 102 103 Returns: 104 PromptInjectionScanner: New scanner instance. 105 """ 106 return cls(audit_manager=audit_manager)
Factory method to construct PromptInjectionScanner.
Args: audit_manager: Optional AuditManager instance.
Returns: PromptInjectionScanner: New scanner instance.
108 def scan_text(self, text: str) -> PromptInjectionScanResult: 109 """Scan input text string for prompt injection and jailbreak signatures. 110 111 Args: 112 text: Input text string to scan. 113 114 Returns: 115 PromptInjectionScanResult: Scan result containing match status and threat details. 116 """ 117 if not text or not isinstance(text, str): 118 return PromptInjectionScanResult(is_detected=False) 119 120 for pattern_id, regex, score in self.INJECTION_PATTERNS: 121 match = regex.search(text) 122 if match: 123 snippet = match.group(0) 124 category = "INSTRUCTION_OVERRIDE" if "INSTRUCTION" in pattern_id else "PROMPT_INJECTION" 125 return PromptInjectionScanResult( 126 is_detected=True, 127 pattern_id=pattern_id, 128 threat_score=score, 129 threat_category=category, 130 detected_text=snippet, 131 ) 132 133 return PromptInjectionScanResult(is_detected=False)
Scan input text string for prompt injection and jailbreak signatures.
Args: text: Input text string to scan.
Returns: PromptInjectionScanResult: Scan result containing match status and threat details.
def
scan_and_audit( self, text: str, subject_id: str, tenant_id: str, enforcement_point: str = 'AGENT_GUARD', context_metadata: Optional[Dict[str, Any]] = None) -> PromptInjectionScanResult:
135 def scan_and_audit( 136 self, 137 text: str, 138 subject_id: str, 139 tenant_id: str, 140 enforcement_point: str = PEP_AGENT_GUARD, 141 context_metadata: Optional[Dict[str, Any]] = None, 142 ) -> PromptInjectionScanResult: 143 """Scan input text and publish audit security alert event if threat is detected. 144 145 Args: 146 text: Input text string. 147 subject_id: Subject attempting execution. 148 tenant_id: Tenant identifier. 149 enforcement_point: Target PEP identifier string. 150 context_metadata: Optional additional metadata dictionary. 151 152 Returns: 153 PromptInjectionScanResult: Scan result envelope. 154 """ 155 result = self.scan_text(text) 156 if result.is_detected and self._audit_manager: 157 event_type = ( 158 EVENT_INSTRUCTION_OVERRIDE_ATTEMPT 159 if result.threat_category == "INSTRUCTION_OVERRIDE" 160 else EVENT_PROMPT_INJECTION_DETECTED 161 ) 162 severity = ( 163 SEVERITY_CRITICAL_ALERT 164 if result.threat_score >= 0.9 165 else SEVERITY_SECURITY_ALERT 166 ) 167 168 meta = dict(context_metadata) if context_metadata else {} 169 meta.update({ 170 "detected_snippet": result.detected_text, 171 "threat_score": result.threat_score, 172 "pattern_id": result.pattern_id, 173 }) 174 175 event = AuditEvent.create( 176 subject_id=subject_id, 177 tenant_id=tenant_id, 178 action="prompt:scan", 179 resource=enforcement_point, 180 decision="BLOCKED", 181 diagnostic_reason=f"Prompt injection pattern detected: '{result.pattern_id}'", 182 severity=severity, 183 metadata=meta, 184 enforcement_point=enforcement_point, 185 reason_code=event_type, 186 ) 187 event.threat_category = result.threat_category 188 event.threat_score = result.threat_score 189 event.detected_pattern = result.pattern_id 190 event.mitigation_action = STRATEGY_HARD_ABORT 191 192 self._audit_manager.record_event(event) 193 194 return result
Scan input text and publish audit security alert event if threat is detected.
Args: text: Input text string. subject_id: Subject attempting execution. tenant_id: Tenant identifier. enforcement_point: Target PEP identifier string. context_metadata: Optional additional metadata dictionary.
Returns: PromptInjectionScanResult: Scan result envelope.
class
PromptInjectionScanResult:
22class PromptInjectionScanResult: 23 """Scan outcome envelope indicating injection status, confidence score, and pattern ID. 24 25 Attributes: 26 is_detected (bool): True if prompt injection pattern was identified. 27 pattern_id (Optional[str]): Identifier of matched rule pattern. 28 threat_score (float): Confidence score of detected threat (0.0 to 1.0). 29 threat_category (str): Threat classification label. 30 detected_text (Optional[str]): Matched text snippet. 31 """ 32 33 def __init__( 34 self, 35 is_detected: bool, 36 pattern_id: Optional[str] = None, 37 threat_score: float = 0.0, 38 threat_category: str = "PROMPT_INJECTION", 39 detected_text: Optional[str] = None, 40 ): 41 """Initialize PromptInjectionScanResult. 42 43 Args: 44 is_detected: Flag indicating threat detection. 45 pattern_id: Pattern identifier. 46 threat_score: Threat confidence score. 47 threat_category: Threat category string. 48 detected_text: Snippet of matched input text. 49 """ 50 self.is_detected = is_detected 51 self.pattern_id = pattern_id 52 self.threat_score = threat_score 53 self.threat_category = threat_category 54 self.detected_text = detected_text
Scan outcome envelope indicating injection status, confidence score, and pattern ID.
Attributes: is_detected (bool): True if prompt injection pattern was identified. pattern_id (Optional[str]): Identifier of matched rule pattern. threat_score (float): Confidence score of detected threat (0.0 to 1.0). threat_category (str): Threat classification label. detected_text (Optional[str]): Matched text snippet.
PromptInjectionScanResult( is_detected: bool, pattern_id: Optional[str] = None, threat_score: float = 0.0, threat_category: str = 'PROMPT_INJECTION', detected_text: Optional[str] = None)
33 def __init__( 34 self, 35 is_detected: bool, 36 pattern_id: Optional[str] = None, 37 threat_score: float = 0.0, 38 threat_category: str = "PROMPT_INJECTION", 39 detected_text: Optional[str] = None, 40 ): 41 """Initialize PromptInjectionScanResult. 42 43 Args: 44 is_detected: Flag indicating threat detection. 45 pattern_id: Pattern identifier. 46 threat_score: Threat confidence score. 47 threat_category: Threat category string. 48 detected_text: Snippet of matched input text. 49 """ 50 self.is_detected = is_detected 51 self.pattern_id = pattern_id 52 self.threat_score = threat_score 53 self.threat_category = threat_category 54 self.detected_text = detected_text
Initialize PromptInjectionScanResult.
Args: is_detected: Flag indicating threat detection. pattern_id: Pattern identifier. threat_score: Threat confidence score. threat_category: Threat category string. detected_text: Snippet of matched input text.
class
DiagnosticReasonEnum(builtins.str, enum.Enum):
26class DiagnosticReasonEnum(str, Enum): 27 """Structured diagnostic reason enumeration for policy decisions and exceptions.""" 28 AUTHORIZED = REASON_AUTHORIZED 29 TENANT_MISMATCH = REASON_TENANT_MISMATCH 30 INSUFFICIENT_CLEARANCE = REASON_INSUFFICIENT_CLEARANCE 31 MISSING_REQUIRED_ROLE = REASON_MISSING_REQUIRED_ROLE 32 PARAMETER_CONSTRAINT_FAILED = REASON_PARAMETER_CONSTRAINT_FAILED 33 FAIL_CLOSED_NULL_ENTITY = REASON_FAIL_CLOSED_NULL_ENTITY 34 INVALID_ENTITY = REASON_INVALID_ENTITY 35 INTERNAL_ERROR = REASON_INTERNAL_ERROR 36 UNAUTHENTICATED = REASON_UNAUTHENTICATED 37 INVALID_HMAC = REASON_INVALID_HMAC
Structured diagnostic reason enumeration for policy decisions and exceptions.
AUTHORIZED =
<DiagnosticReasonEnum.AUTHORIZED: 'AUTHORIZED'>
TENANT_MISMATCH =
<DiagnosticReasonEnum.TENANT_MISMATCH: 'TENANT_MISMATCH'>
INSUFFICIENT_CLEARANCE =
<DiagnosticReasonEnum.INSUFFICIENT_CLEARANCE: 'INSUFFICIENT_CLEARANCE'>
MISSING_REQUIRED_ROLE =
<DiagnosticReasonEnum.MISSING_REQUIRED_ROLE: 'MISSING_REQUIRED_ROLE'>
PARAMETER_CONSTRAINT_FAILED =
<DiagnosticReasonEnum.PARAMETER_CONSTRAINT_FAILED: 'PARAMETER_CONSTRAINT_FAILED'>
FAIL_CLOSED_NULL_ENTITY =
<DiagnosticReasonEnum.FAIL_CLOSED_NULL_ENTITY: 'FAIL_CLOSED_NULL_ENTITY'>
INVALID_ENTITY =
<DiagnosticReasonEnum.INVALID_ENTITY: 'INVALID_ENTITY'>
INTERNAL_ERROR =
<DiagnosticReasonEnum.INTERNAL_ERROR: 'INTERNAL_ERROR'>
UNAUTHENTICATED =
<DiagnosticReasonEnum.UNAUTHENTICATED: 'UNAUTHENTICATED'>
INVALID_HMAC =
<DiagnosticReasonEnum.INVALID_HMAC: 'INVALID_HMAC'>
class
SecurityAuthorizationError(builtins.Exception):
40class SecurityAuthorizationError(Exception): 41 """Base exception for all security authorization failures across PEP enforcement points. 42 43 Attributes: 44 message (str): Human-readable error description. 45 diagnostic_reason (Optional[str]): Standard diagnostic reason code. 46 obligations (Optional[Dict[str, Any]]): Map of policy enforcement obligations. 47 """ 48 49 def __init__(self, message: str, diagnostic_reason: Optional[str], obligations: Optional[Dict[str, Any]]): 50 """Initialize SecurityAuthorizationError. 51 52 Args: 53 message: Human-readable error description. 54 diagnostic_reason: Diagnostic reason code from PDP evaluation. 55 obligations: Key-value mapping of policy enforcement obligations. 56 """ 57 super().__init__(message) 58 self.message = message 59 self.diagnostic_reason = diagnostic_reason 60 self.obligations = obligations 61 62 @property 63 def is_tenant_violation(self) -> bool: 64 """Check whether this authorization failure is a cross-tenant isolation violation.""" 65 return self.diagnostic_reason == DiagnosticReasonEnum.TENANT_MISMATCH 66 67 @classmethod 68 def create(cls, message: str, diagnostic_reason: Optional[str], obligations: Optional[Dict[str, Any]]): 69 """Factory method to instantiate a SecurityAuthorizationError. 70 71 Args: 72 message: Human-readable error description. 73 diagnostic_reason: Diagnostic reason code. 74 obligations: Dictionary of obligations. 75 76 Returns: 77 SecurityAuthorizationError: New exception instance. 78 """ 79 return cls(message, diagnostic_reason, obligations)
Base exception for all security authorization failures across PEP enforcement points.
Attributes: message (str): Human-readable error description. diagnostic_reason (Optional[str]): Standard diagnostic reason code. obligations (Optional[Dict[str, Any]]): Map of policy enforcement obligations.
SecurityAuthorizationError( message: str, diagnostic_reason: Optional[str], obligations: Optional[Dict[str, Any]])
49 def __init__(self, message: str, diagnostic_reason: Optional[str], obligations: Optional[Dict[str, Any]]): 50 """Initialize SecurityAuthorizationError. 51 52 Args: 53 message: Human-readable error description. 54 diagnostic_reason: Diagnostic reason code from PDP evaluation. 55 obligations: Key-value mapping of policy enforcement obligations. 56 """ 57 super().__init__(message) 58 self.message = message 59 self.diagnostic_reason = diagnostic_reason 60 self.obligations = obligations
Initialize SecurityAuthorizationError.
Args: message: Human-readable error description. diagnostic_reason: Diagnostic reason code from PDP evaluation. obligations: Key-value mapping of policy enforcement obligations.
is_tenant_violation: bool
62 @property 63 def is_tenant_violation(self) -> bool: 64 """Check whether this authorization failure is a cross-tenant isolation violation.""" 65 return self.diagnostic_reason == DiagnosticReasonEnum.TENANT_MISMATCH
Check whether this authorization failure is a cross-tenant isolation violation.
@classmethod
def
create( cls, message: str, diagnostic_reason: Optional[str], obligations: Optional[Dict[str, Any]]):
67 @classmethod 68 def create(cls, message: str, diagnostic_reason: Optional[str], obligations: Optional[Dict[str, Any]]): 69 """Factory method to instantiate a SecurityAuthorizationError. 70 71 Args: 72 message: Human-readable error description. 73 diagnostic_reason: Diagnostic reason code. 74 obligations: Dictionary of obligations. 75 76 Returns: 77 SecurityAuthorizationError: New exception instance. 78 """ 79 return cls(message, diagnostic_reason, obligations)
Factory method to instantiate a SecurityAuthorizationError.
Args: message: Human-readable error description. diagnostic_reason: Diagnostic reason code. obligations: Dictionary of obligations.
Returns: SecurityAuthorizationError: New exception instance.
class
AgentAccessDeniedException(langchain_security_extension.security_core.SecurityAuthorizationError):
82class AgentAccessDeniedException(SecurityAuthorizationError): 83 """Exception raised when entry invocation to an agent or Runnable is denied by PDP. 84 85 Attributes: 86 agent_name (str): Identifier of the agent or Runnable target. 87 """ 88 89 def __init__(self, agent_name: str, message: str, diagnostic_reason: Optional[str]): 90 """Initialize AgentAccessDeniedException. 91 92 Args: 93 agent_name: Name or ID of the protected agent. 94 message: Error description message. 95 diagnostic_reason: Reason code explaining denial. 96 """ 97 super().__init__(message, diagnostic_reason, None) 98 self.agent_name = agent_name 99 100 @classmethod 101 def for_agent(cls, agent_name: str, message: str, diagnostic_reason: Optional[str]): 102 """Factory method creating an AgentAccessDeniedException or TenantIsolationError for a named agent. 103 104 Args: 105 agent_name: Target agent identifier. 106 message: Exception message. 107 diagnostic_reason: Reason code explaining the decision. 108 109 Returns: 110 AgentAccessDeniedException: Constructed exception instance. 111 """ 112 if diagnostic_reason == DiagnosticReasonEnum.TENANT_MISMATCH: 113 return TenantIsolationError(agent_name, message, diagnostic_reason) 114 return cls(agent_name, message, diagnostic_reason)
Exception raised when entry invocation to an agent or Runnable is denied by PDP.
Attributes: agent_name (str): Identifier of the agent or Runnable target.
AgentAccessDeniedException(agent_name: str, message: str, diagnostic_reason: Optional[str])
89 def __init__(self, agent_name: str, message: str, diagnostic_reason: Optional[str]): 90 """Initialize AgentAccessDeniedException. 91 92 Args: 93 agent_name: Name or ID of the protected agent. 94 message: Error description message. 95 diagnostic_reason: Reason code explaining denial. 96 """ 97 super().__init__(message, diagnostic_reason, None) 98 self.agent_name = agent_name
Initialize AgentAccessDeniedException.
Args: agent_name: Name or ID of the protected agent. message: Error description message. diagnostic_reason: Reason code explaining denial.
@classmethod
def
for_agent(cls, agent_name: str, message: str, diagnostic_reason: Optional[str]):
100 @classmethod 101 def for_agent(cls, agent_name: str, message: str, diagnostic_reason: Optional[str]): 102 """Factory method creating an AgentAccessDeniedException or TenantIsolationError for a named agent. 103 104 Args: 105 agent_name: Target agent identifier. 106 message: Exception message. 107 diagnostic_reason: Reason code explaining the decision. 108 109 Returns: 110 AgentAccessDeniedException: Constructed exception instance. 111 """ 112 if diagnostic_reason == DiagnosticReasonEnum.TENANT_MISMATCH: 113 return TenantIsolationError(agent_name, message, diagnostic_reason) 114 return cls(agent_name, message, diagnostic_reason)
Factory method creating an AgentAccessDeniedException or TenantIsolationError for a named agent.
Args: agent_name: Target agent identifier. message: Exception message. diagnostic_reason: Reason code explaining the decision.
Returns: AgentAccessDeniedException: Constructed exception instance.
117class TenantIsolationError(AgentAccessDeniedException): 118 """Typed exception raised when an agent invocation violates tenant boundary isolation.""" 119 pass
Typed exception raised when an agent invocation violates tenant boundary isolation.
class
ToolExecutionDeniedError(langchain_security_extension.security_core.SecurityAuthorizationError):
122class ToolExecutionDeniedError(SecurityAuthorizationError): 123 """Hard abort exception raised when a tool execution is prohibited by policy. 124 125 Attributes: 126 tool_name (str): Name of the tool execution attempt that was aborted. 127 """ 128 129 def __init__(self, tool_name: str, message: str, diagnostic_reason: Optional[str]): 130 """Initialize ToolExecutionDeniedError. 131 132 Args: 133 tool_name: Name of the prohibited tool. 134 message: Error details. 135 diagnostic_reason: Diagnostic reason for denial. 136 """ 137 super().__init__(message, diagnostic_reason, None) 138 self.tool_name = tool_name 139 140 @classmethod 141 def for_tool(cls, tool_name: str, message: str, diagnostic_reason: Optional[str]): 142 """Factory method creating a ToolExecutionDeniedError for a named tool. 143 144 Args: 145 tool_name: Prohibited tool identifier. 146 message: Detailed error message. 147 diagnostic_reason: Diagnostic reason code. 148 149 Returns: 150 ToolExecutionDeniedError: Constructed exception instance. 151 """ 152 return cls(tool_name, message, diagnostic_reason)
Hard abort exception raised when a tool execution is prohibited by policy.
Attributes: tool_name (str): Name of the tool execution attempt that was aborted.
ToolExecutionDeniedError(tool_name: str, message: str, diagnostic_reason: Optional[str])
129 def __init__(self, tool_name: str, message: str, diagnostic_reason: Optional[str]): 130 """Initialize ToolExecutionDeniedError. 131 132 Args: 133 tool_name: Name of the prohibited tool. 134 message: Error details. 135 diagnostic_reason: Diagnostic reason for denial. 136 """ 137 super().__init__(message, diagnostic_reason, None) 138 self.tool_name = tool_name
Initialize ToolExecutionDeniedError.
Args: tool_name: Name of the prohibited tool. message: Error details. diagnostic_reason: Diagnostic reason for denial.
@classmethod
def
for_tool(cls, tool_name: str, message: str, diagnostic_reason: Optional[str]):
140 @classmethod 141 def for_tool(cls, tool_name: str, message: str, diagnostic_reason: Optional[str]): 142 """Factory method creating a ToolExecutionDeniedError for a named tool. 143 144 Args: 145 tool_name: Prohibited tool identifier. 146 message: Detailed error message. 147 diagnostic_reason: Diagnostic reason code. 148 149 Returns: 150 ToolExecutionDeniedError: Constructed exception instance. 151 """ 152 return cls(tool_name, message, diagnostic_reason)
Factory method creating a ToolExecutionDeniedError for a named tool.
Args: tool_name: Prohibited tool identifier. message: Detailed error message. diagnostic_reason: Diagnostic reason code.
Returns: ToolExecutionDeniedError: Constructed exception instance.
class
ToolExecutionFeedbackException(langchain_security_extension.security_core.SecurityAuthorizationError):
160class ToolExecutionFeedbackException(SecurityAuthorizationError): 161 """Exception carrying synthetic model refusal feedback for non-mutative tool violations. 162 163 Attributes: 164 tool_name (str): Tool that encountered constraint violation. 165 refusal_payload (Dict[str, Any]): Structured feedback dictionary returned to LLM. 166 """ 167 168 def __init__(self, tool_name: str, refusal_payload: Dict[str, Any]): 169 """Initialize ToolExecutionFeedbackException. 170 171 Args: 172 tool_name: Name of tool being invoked. 173 refusal_payload: Structured refusal payload for model consumption. 174 """ 175 super().__init__(f"Tool execution denied with feedback: {tool_name}", None, None) 176 self.tool_name = tool_name 177 self.refusal_payload = refusal_payload 178 179 @classmethod 180 def create_payload(cls, tool_name: str, refusal_payload: Dict[str, Any]): 181 """Factory method creating a ToolExecutionFeedbackException with refusal payload. 182 183 Args: 184 tool_name: Name of target tool. 185 refusal_payload: Refusal details for model retry loop. 186 187 Returns: 188 ToolExecutionFeedbackException: Exception instance containing feedback. 189 """ 190 return cls(tool_name, refusal_payload)
Exception carrying synthetic model refusal feedback for non-mutative tool violations.
Attributes: tool_name (str): Tool that encountered constraint violation. refusal_payload (Dict[str, Any]): Structured feedback dictionary returned to LLM.
ToolExecutionFeedbackException(tool_name: str, refusal_payload: Dict[str, Any])
168 def __init__(self, tool_name: str, refusal_payload: Dict[str, Any]): 169 """Initialize ToolExecutionFeedbackException. 170 171 Args: 172 tool_name: Name of tool being invoked. 173 refusal_payload: Structured refusal payload for model consumption. 174 """ 175 super().__init__(f"Tool execution denied with feedback: {tool_name}", None, None) 176 self.tool_name = tool_name 177 self.refusal_payload = refusal_payload
Initialize ToolExecutionFeedbackException.
Args: tool_name: Name of tool being invoked. refusal_payload: Structured refusal payload for model consumption.
@classmethod
def
create_payload(cls, tool_name: str, refusal_payload: Dict[str, Any]):
179 @classmethod 180 def create_payload(cls, tool_name: str, refusal_payload: Dict[str, Any]): 181 """Factory method creating a ToolExecutionFeedbackException with refusal payload. 182 183 Args: 184 tool_name: Name of target tool. 185 refusal_payload: Refusal details for model retry loop. 186 187 Returns: 188 ToolExecutionFeedbackException: Exception instance containing feedback. 189 """ 190 return cls(tool_name, refusal_payload)
Factory method creating a ToolExecutionFeedbackException with refusal payload.
Args: tool_name: Name of target tool. refusal_payload: Refusal details for model retry loop.
Returns: ToolExecutionFeedbackException: Exception instance containing feedback.
252class PolicyViolationError(SecurityAuthorizationError): 253 """Exception raised when policy configuration or static rule validation fails.""" 254 pass
Exception raised when policy configuration or static rule validation fails.
SecurityValidationException =
<class 'langchain_security_extension.security_core.exceptions.IngestionRejectedException'>
DECISION_PERMIT =
'PERMIT'
DECISION_DENY =
'DENY'
ACTION_AGENT_INVOKE =
'agent:invoke'
ACTION_TOOL_EXECUTE =
'tool:execute'
ACTION_CHUNK_RETRIEVE =
'chunk:retrieve'
STRATEGY_HARD_ABORT =
'HARD_ABORT'
STRATEGY_RETURN_FEEDBACK =
'RETURN_FEEDBACK'
class
PiiSanitizerSPI(abc.ABC):
193class PiiSanitizerSPI(ABC): 194 """Abstract Service Provider Interface for PII and Secret Redaction engines.""" 195 196 @abstractmethod 197 def sanitize_text( 198 self, 199 text: str, 200 context: Optional[PiiContext], 201 *args: Any, 202 **kwargs: Any, 203 ) -> str: 204 """Sanitize sensitive entities in raw text strings. 205 206 Args: 207 text: Input string to inspect and sanitize. 208 context: Optional contextual information for the sanitization stage. 209 *args: Optional positional arguments (e.g. max_chars). 210 **kwargs: Optional keyword arguments. 211 212 Returns: 213 Sanitized string with sensitive data redacted. 214 """ 215 pass 216 217 @abstractmethod 218 def sanitize_payload( 219 self, 220 payload: Dict[str, Any], 221 context: Optional[PiiContext], 222 *args: Any, 223 **kwargs: Any, 224 ) -> Dict[str, Any]: 225 """Recursively sanitize a dictionary payload. 226 227 Args: 228 payload: Input dictionary to sanitize. 229 context: Optional contextual information for the sanitization stage. 230 *args: Optional positional overrides (max_chars, masked_fields). 231 **kwargs: Optional keyword overrides. 232 233 Returns: 234 Sanitized dictionary with sensitive values redacted. 235 """ 236 pass
Abstract Service Provider Interface for PII and Secret Redaction engines.
@abstractmethod
def
sanitize_text( self, text: str, context: Optional[PiiContext], *args: Any, **kwargs: Any) -> str:
196 @abstractmethod 197 def sanitize_text( 198 self, 199 text: str, 200 context: Optional[PiiContext], 201 *args: Any, 202 **kwargs: Any, 203 ) -> str: 204 """Sanitize sensitive entities in raw text strings. 205 206 Args: 207 text: Input string to inspect and sanitize. 208 context: Optional contextual information for the sanitization stage. 209 *args: Optional positional arguments (e.g. max_chars). 210 **kwargs: Optional keyword arguments. 211 212 Returns: 213 Sanitized string with sensitive data redacted. 214 """ 215 pass
Sanitize sensitive entities in raw text strings.
Args: text: Input string to inspect and sanitize. context: Optional contextual information for the sanitization stage. *args: Optional positional arguments (e.g. max_chars). **kwargs: Optional keyword arguments.
Returns: Sanitized string with sensitive data redacted.
@abstractmethod
def
sanitize_payload( self, payload: Dict[str, Any], context: Optional[PiiContext], *args: Any, **kwargs: Any) -> Dict[str, Any]:
217 @abstractmethod 218 def sanitize_payload( 219 self, 220 payload: Dict[str, Any], 221 context: Optional[PiiContext], 222 *args: Any, 223 **kwargs: Any, 224 ) -> Dict[str, Any]: 225 """Recursively sanitize a dictionary payload. 226 227 Args: 228 payload: Input dictionary to sanitize. 229 context: Optional contextual information for the sanitization stage. 230 *args: Optional positional overrides (max_chars, masked_fields). 231 **kwargs: Optional keyword overrides. 232 233 Returns: 234 Sanitized dictionary with sensitive values redacted. 235 """ 236 pass
Recursively sanitize a dictionary payload.
Args: payload: Input dictionary to sanitize. context: Optional contextual information for the sanitization stage. *args: Optional positional overrides (max_chars, masked_fields). **kwargs: Optional keyword overrides.
Returns: Sanitized dictionary with sensitive values redacted.
239class BuiltInPiiSanitizer(PiiSanitizerSPI): 240 """Default zero-dependency high-performance PII and credential scrubber. 241 242 Performs 3-stage sanitization: 243 - Stage 1: Key-name masking for sensitive dictionary parameters. 244 - Stage 2: Comprehensive regex matching for emails, phones, SSNs, credit cards (with Luhn), 245 cloud credentials (AWS, Google, GitHub, Slack), JWTs, DB URIs, and private keys. 246 - Stage 3: Payload size truncation to prevent log bloat. 247 """ 248 249 COMBINED_SECRET_REGEX = re.compile( 250 r'(?P<db_uri>(?:postgres(?:ql)?|mysql|mongodb(?:\+srv)?|redis)://[^:\s/]+:[^@\s/]+@[^\s,\'"]+)' 251 r'|(?P<kv_secret>(?:bearer[\s=]+[^\s,\'"]+|password\s*[:=]\s*[^\s,\'"]+|secret\s*[:=]\s*[^\s,\'"]+|token\s*[:=]\s*[^\s,\'"]+|api[_\-]?key\s*[:=]\s*[^\s,\'"]+|aws[_\-]?key\s*[:=]\s*[^\s,\'"]+))' 252 r'|(?P<jwt>eyJ[a-zA-Z0-9_\-]{5,}\.eyJ[a-zA-Z0-9_\-]{5,}\.[a-zA-Z0-9_\-]{5,})' 253 r'|(?P<aws_key>(?:AKIA|ASIA)[0-9A-Z]{16})' 254 r'|(?P<google_key>AIza[-0-9A-Za-z_]{35})' 255 r'|(?P<github_token>gh[pousr]_[A-Za-z0-9_]{36,255})' 256 r'|(?P<slack_token>xox[baprs]-[-0-9A-Za-z]{10,72})' 257 r'|(?P<private_key>-----BEGIN [A-Z\s]+ PRIVATE KEY-----[A-Za-z0-9+/=\s]+-----END [A-Z\s]+ PRIVATE KEY-----)' 258 r'|(?P<email>\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,7}\b)' 259 r'|(?P<ipv4>\b(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\b)' 260 r'|(?P<pan>\b[A-Za-z]{5}[-\s]?[0-9]{4}[-\s]?[A-Za-z]\b)' 261 r'|(?P<card>\b(?:\d{4}[-\s]?){3}\d{1,7}\b|\b\d{13,19}\b)' 262 r'|(?P<aadhaar>(?<!\d)[2-9]\d{3}[-\s]\d{4}[-\s]\d{4}(?![-\s]?\d)|\b[2-9]\d{11}\b|(?<!\d)\d{4}[-\s]\d{4}[-\s]\d{4}(?![-\s]?\d)|\b\d{12}\b)' 263 r'|(?P<ssn>\b\d{3}[-\s]\d{2}[-\s]\d{4}\b|\b\d{9}\b)' 264 r'|(?P<phone>(?:\+\d{1,3}[-.\s]?)?(?:\(\d{3}\)[-.\s]?|\b\d{3}[-.\s])\d{3}[-.\s]\d{4}\b|\b\d{10}\b)', 265 re.IGNORECASE, 266 ) 267 268 PHONE_HINT_REGEX = re.compile(DEFAULT_PHONE_HINT_PATTERN, re.IGNORECASE) 269 SSN_HINT_REGEX = re.compile(DEFAULT_SSN_HINT_PATTERN, re.IGNORECASE) 270 AADHAAR_HINT_REGEX = re.compile(DEFAULT_AADHAAR_HINT_PATTERN, re.IGNORECASE) 271 PAN_HINT_REGEX = re.compile(DEFAULT_PAN_HINT_PATTERN, re.IGNORECASE) 272 273 274 def __init__(self, *args: Any, **kwargs: Any): 275 """Initialize BuiltInPiiSanitizer. 276 277 Args: 278 *args: Optional positional arguments (masked_fields). 279 **kwargs: Optional keyword arguments (masked_fields). 280 """ 281 masked_fields = kwargs.get("masked_fields", args[0] if len(args) > 0 else ()) 282 self._registered_masked_fields: Set[str] = { 283 str(f).lower().strip() for f in DEFAULT_SENSITIVE_FIELD_NAMES 284 } 285 if masked_fields: 286 for field in masked_fields: 287 self.register_masked_field(str(field)) 288 289 @classmethod 290 def create(cls, *args: Any, **kwargs: Any) -> "BuiltInPiiSanitizer": 291 """Factory method to construct a BuiltInPiiSanitizer. 292 293 Args: 294 *args: Optional positional arguments (masked_fields). 295 **kwargs: Optional keyword arguments (masked_fields). 296 297 Returns: 298 BuiltInPiiSanitizer: Configured sanitizer instance. 299 """ 300 return cls(*args, **kwargs) 301 302 def register_masked_field(self, field_name: str) -> None: 303 """Register a parameter or field name for automatic Stage 1 dictionary masking. 304 305 Args: 306 field_name: Name of field or argument to redact when encountered in dictionaries. 307 """ 308 if field_name: 309 self._registered_masked_fields.add(field_name.lower().strip()) 310 311 def is_masked_field(self, field_name: str) -> bool: 312 """Check if a field name is marked as sensitive/masked. 313 314 Args: 315 field_name: Field name to query. 316 317 Returns: 318 bool: True if the field is registered as sensitive/masked, False otherwise. 319 """ 320 return bool(field_name and field_name.lower().strip() in self._registered_masked_fields) 321 322 def _redact_match(self, match: re.Match, full_text: str) -> str: 323 group_dict = match.groupdict() 324 text = match.group(0) 325 326 if group_dict.get("db_uri"): 327 scheme_user = text.split(":", 2) 328 at_part = text.split("@", 1)[1] 329 return f"{scheme_user[0]}:{scheme_user[1]}:{REDACTED_TOKEN}@{at_part}" 330 331 if group_dict.get("kv_secret"): 332 if "=" in text: 333 prefix = text.split("=")[0] 334 return f"{prefix}={REDACTED_TOKEN}" 335 elif ":" in text: 336 prefix = text.split(":")[0] 337 return f"{prefix}: {REDACTED_TOKEN}" 338 elif text.lower().startswith("bearer "): 339 return f"bearer {REDACTED_TOKEN}" 340 return REDACTED_TOKEN 341 342 if group_dict.get("pan"): 343 if is_valid_pan(text): 344 return REDACTED_TOKEN 345 return text 346 347 if group_dict.get("aadhaar"): 348 digits = "".join(c for c in text if c.isdigit()) 349 if len(digits) == 12: 350 start_idx = max(0, match.start() - AADHAAR_CONTEXT_WINDOW_CHARS) 351 end_idx = min(len(full_text), match.end() + AADHAAR_CONTEXT_WINDOW_CHARS) 352 window = full_text[start_idx:end_idx] 353 has_hint = bool(self.AADHAAR_HINT_REGEX.search(window)) 354 if has_hint: 355 return REDACTED_TOKEN 356 if is_valid_aadhaar(digits): 357 if not any(c in text for c in "- "): 358 return text 359 return REDACTED_TOKEN 360 return text 361 362 if group_dict.get("card"): 363 digits = "".join(c for c in text if c.isdigit()) 364 if is_valid_luhn(digits): 365 return REDACTED_TOKEN 366 return text 367 368 if group_dict.get("ssn"): 369 digits = "".join(c for c in text if c.isdigit()) 370 # For 9-digit numbers lacking SSN formatting delimiters, require proximity to SSN keywords 371 if len(digits) == 9 and not any(c in text for c in "- "): 372 start_idx = max(0, match.start() - SSN_CONTEXT_WINDOW_CHARS) 373 end_idx = min(len(full_text), match.end() + SSN_CONTEXT_WINDOW_CHARS) 374 window = full_text[start_idx:end_idx] 375 if not self.SSN_HINT_REGEX.search(window): 376 return text 377 return REDACTED_TOKEN 378 379 if group_dict.get("phone"): 380 digits = "".join(c for c in text if c.isdigit()) 381 # For 10-digit numbers lacking phone formatting delimiters, require proximity to phone/fone keywords 382 if len(digits) == 10 and not any(c in text for c in "-. ()+"): 383 start_idx = max(0, match.start() - PHONE_CONTEXT_WINDOW_CHARS) 384 end_idx = min(len(full_text), match.end() + PHONE_CONTEXT_WINDOW_CHARS) 385 window = full_text[start_idx:end_idx] 386 if not self.PHONE_HINT_REGEX.search(window): 387 return text 388 return REDACTED_TOKEN 389 390 return REDACTED_TOKEN 391 392 def sanitize_text( 393 self, 394 text: str, 395 context: Optional[PiiContext], 396 *args: Any, 397 **kwargs: Any, 398 ) -> str: 399 """Sanitize sensitive entities in raw text strings. 400 401 Args: 402 text: Raw input string to inspect and sanitize. 403 context: Optional PiiContext detailing lifecycle stage and caller identity. 404 *args: Optional positional arguments (e.g. max_chars limit). 405 **kwargs: Optional keyword arguments (e.g. max_chars limit). 406 407 Returns: 408 str: Redacted string with credentials, PII tokens, and secrets obscured. 409 """ 410 max_chars = int(kwargs.get("max_chars", args[0] if len(args) > 0 else DEFAULT_PAYLOAD_MAX_CHARS)) 411 cleaned = self.COMBINED_SECRET_REGEX.sub(lambda m: self._redact_match(m, text), text) 412 if len(cleaned) > max_chars: 413 cleaned = cleaned[:max_chars] + TRUNCATED_SUFFIX 414 return cleaned 415 416 def sanitize_payload( 417 self, 418 payload: Dict[str, Any], 419 context: Optional[PiiContext], 420 *args: Any, 421 **kwargs: Any, 422 ) -> Dict[str, Any]: 423 """Recursively sanitize a dictionary payload. 424 425 Args: 426 payload: Input dictionary containing parameters or telemetry data. 427 context: Optional PiiContext detailing lifecycle stage and caller identity. 428 *args: Optional positional arguments (max_chars, masked_fields). 429 **kwargs: Optional keyword arguments (max_chars, masked_fields). 430 431 Returns: 432 Dict[str, Any]: New dictionary with sensitive fields and in-string PII redacted. 433 """ 434 max_chars = int(kwargs.get("max_chars", args[0] if len(args) > 0 else DEFAULT_PAYLOAD_MAX_CHARS)) 435 extra_masked = set(kwargs.get("masked_fields", args[1] if len(args) > 1 else ())) 436 masked_lower = {str(f).lower() for f in extra_masked} | self._registered_masked_fields 437 438 sanitized: Dict[str, Any] = {} 439 for k, v in payload.items(): 440 if str(k).lower() in masked_lower: 441 sanitized[k] = REDACTED_TOKEN 442 elif isinstance(v, str): 443 sanitized[k] = self.sanitize_text(v, context, max_chars) 444 elif isinstance(v, dict): 445 sanitized[k] = self.sanitize_payload(v, context, max_chars, extra_masked) 446 elif isinstance(v, list): 447 sanitized[k] = [ 448 self.sanitize_payload(item, context, max_chars, extra_masked) 449 if isinstance(item, dict) 450 else (self.sanitize_text(item, context, max_chars) if isinstance(item, str) else item) 451 for item in v 452 ] 453 else: 454 sanitized[k] = v 455 return sanitized
Default zero-dependency high-performance PII and credential scrubber.
Performs 3-stage sanitization:
- Stage 1: Key-name masking for sensitive dictionary parameters.
- Stage 2: Comprehensive regex matching for emails, phones, SSNs, credit cards (with Luhn), cloud credentials (AWS, Google, GitHub, Slack), JWTs, DB URIs, and private keys.
- Stage 3: Payload size truncation to prevent log bloat.
BuiltInPiiSanitizer(*args: Any, **kwargs: Any)
274 def __init__(self, *args: Any, **kwargs: Any): 275 """Initialize BuiltInPiiSanitizer. 276 277 Args: 278 *args: Optional positional arguments (masked_fields). 279 **kwargs: Optional keyword arguments (masked_fields). 280 """ 281 masked_fields = kwargs.get("masked_fields", args[0] if len(args) > 0 else ()) 282 self._registered_masked_fields: Set[str] = { 283 str(f).lower().strip() for f in DEFAULT_SENSITIVE_FIELD_NAMES 284 } 285 if masked_fields: 286 for field in masked_fields: 287 self.register_masked_field(str(field))
Initialize BuiltInPiiSanitizer.
Args: *args: Optional positional arguments (masked_fields). **kwargs: Optional keyword arguments (masked_fields).
COMBINED_SECRET_REGEX =
re.compile('(?P<db_uri>(?:postgres(?:ql)?|mysql|mongodb(?:\\+srv)?|redis)://[^:\\s/]+:[^@\\s/]+@[^\\s,\\\'"]+)|(?P<kv_secret>(?:bearer[\\s=]+[^\\s,\\\'"]+|password\\s*[:=]\\s*[^\\s,\\\'"]+|secret\\s*[:=]\\s*[^\\, re.IGNORECASE)
SSN_HINT_REGEX =
re.compile('(?i)\\b(?:ssn|s\\s*s\\s*n|social\\s+security\\s+(?:number|no|#)?|social\\s+sec\\s*#?)\\b', re.IGNORECASE)
289 @classmethod 290 def create(cls, *args: Any, **kwargs: Any) -> "BuiltInPiiSanitizer": 291 """Factory method to construct a BuiltInPiiSanitizer. 292 293 Args: 294 *args: Optional positional arguments (masked_fields). 295 **kwargs: Optional keyword arguments (masked_fields). 296 297 Returns: 298 BuiltInPiiSanitizer: Configured sanitizer instance. 299 """ 300 return cls(*args, **kwargs)
Factory method to construct a BuiltInPiiSanitizer.
Args: *args: Optional positional arguments (masked_fields). **kwargs: Optional keyword arguments (masked_fields).
Returns: BuiltInPiiSanitizer: Configured sanitizer instance.
def
register_masked_field(self, field_name: str) -> None:
302 def register_masked_field(self, field_name: str) -> None: 303 """Register a parameter or field name for automatic Stage 1 dictionary masking. 304 305 Args: 306 field_name: Name of field or argument to redact when encountered in dictionaries. 307 """ 308 if field_name: 309 self._registered_masked_fields.add(field_name.lower().strip())
Register a parameter or field name for automatic Stage 1 dictionary masking.
Args: field_name: Name of field or argument to redact when encountered in dictionaries.
def
is_masked_field(self, field_name: str) -> bool:
311 def is_masked_field(self, field_name: str) -> bool: 312 """Check if a field name is marked as sensitive/masked. 313 314 Args: 315 field_name: Field name to query. 316 317 Returns: 318 bool: True if the field is registered as sensitive/masked, False otherwise. 319 """ 320 return bool(field_name and field_name.lower().strip() in self._registered_masked_fields)
Check if a field name is marked as sensitive/masked.
Args: field_name: Field name to query.
Returns: bool: True if the field is registered as sensitive/masked, False otherwise.
def
sanitize_text( self, text: str, context: Optional[PiiContext], *args: Any, **kwargs: Any) -> str:
392 def sanitize_text( 393 self, 394 text: str, 395 context: Optional[PiiContext], 396 *args: Any, 397 **kwargs: Any, 398 ) -> str: 399 """Sanitize sensitive entities in raw text strings. 400 401 Args: 402 text: Raw input string to inspect and sanitize. 403 context: Optional PiiContext detailing lifecycle stage and caller identity. 404 *args: Optional positional arguments (e.g. max_chars limit). 405 **kwargs: Optional keyword arguments (e.g. max_chars limit). 406 407 Returns: 408 str: Redacted string with credentials, PII tokens, and secrets obscured. 409 """ 410 max_chars = int(kwargs.get("max_chars", args[0] if len(args) > 0 else DEFAULT_PAYLOAD_MAX_CHARS)) 411 cleaned = self.COMBINED_SECRET_REGEX.sub(lambda m: self._redact_match(m, text), text) 412 if len(cleaned) > max_chars: 413 cleaned = cleaned[:max_chars] + TRUNCATED_SUFFIX 414 return cleaned
Sanitize sensitive entities in raw text strings.
Args: text: Raw input string to inspect and sanitize. context: Optional PiiContext detailing lifecycle stage and caller identity. *args: Optional positional arguments (e.g. max_chars limit). **kwargs: Optional keyword arguments (e.g. max_chars limit).
Returns: str: Redacted string with credentials, PII tokens, and secrets obscured.
def
sanitize_payload( self, payload: Dict[str, Any], context: Optional[PiiContext], *args: Any, **kwargs: Any) -> Dict[str, Any]:
416 def sanitize_payload( 417 self, 418 payload: Dict[str, Any], 419 context: Optional[PiiContext], 420 *args: Any, 421 **kwargs: Any, 422 ) -> Dict[str, Any]: 423 """Recursively sanitize a dictionary payload. 424 425 Args: 426 payload: Input dictionary containing parameters or telemetry data. 427 context: Optional PiiContext detailing lifecycle stage and caller identity. 428 *args: Optional positional arguments (max_chars, masked_fields). 429 **kwargs: Optional keyword arguments (max_chars, masked_fields). 430 431 Returns: 432 Dict[str, Any]: New dictionary with sensitive fields and in-string PII redacted. 433 """ 434 max_chars = int(kwargs.get("max_chars", args[0] if len(args) > 0 else DEFAULT_PAYLOAD_MAX_CHARS)) 435 extra_masked = set(kwargs.get("masked_fields", args[1] if len(args) > 1 else ())) 436 masked_lower = {str(f).lower() for f in extra_masked} | self._registered_masked_fields 437 438 sanitized: Dict[str, Any] = {} 439 for k, v in payload.items(): 440 if str(k).lower() in masked_lower: 441 sanitized[k] = REDACTED_TOKEN 442 elif isinstance(v, str): 443 sanitized[k] = self.sanitize_text(v, context, max_chars) 444 elif isinstance(v, dict): 445 sanitized[k] = self.sanitize_payload(v, context, max_chars, extra_masked) 446 elif isinstance(v, list): 447 sanitized[k] = [ 448 self.sanitize_payload(item, context, max_chars, extra_masked) 449 if isinstance(item, dict) 450 else (self.sanitize_text(item, context, max_chars) if isinstance(item, str) else item) 451 for item in v 452 ] 453 else: 454 sanitized[k] = v 455 return sanitized
Recursively sanitize a dictionary payload.
Args: payload: Input dictionary containing parameters or telemetry data. context: Optional PiiContext detailing lifecycle stage and caller identity. *args: Optional positional arguments (max_chars, masked_fields). **kwargs: Optional keyword arguments (max_chars, masked_fields).
Returns: Dict[str, Any]: New dictionary with sensitive fields and in-string PII redacted.
class
PiiContext:
148class PiiContext: 149 """Contextual metadata passed during PII sanitization lifecycle stages.""" 150 151 def __init__( 152 self, 153 stage: str, 154 tenant_id: Optional[str], 155 subject_id: Optional[str], 156 resource_id: Optional[str], 157 metadata: Optional[Dict[str, Any]], 158 ): 159 """Initialize PiiContext. 160 161 Args: 162 stage: Execution lifecycle stage identifier (e.g. audit_payload, tool_argument, outbound_response). 163 tenant_id: Tenant isolation identifier associated with the request. 164 subject_id: Subject identifier of the caller or initiator. 165 resource_id: Identifier of the target resource, tool, or document. 166 metadata: Additional contextual metadata dictionary. 167 """ 168 self.stage = stage 169 self.tenant_id = tenant_id 170 self.subject_id = subject_id 171 self.resource_id = resource_id 172 self.metadata = dict(metadata) if metadata is not None else {} 173 174 @classmethod 175 def create(cls, stage: str, *args: Any, **kwargs: Any) -> "PiiContext": 176 """Factory method to construct a PiiContext. 177 178 Args: 179 stage: Target execution lifecycle stage. 180 *args: Optional positional arguments (tenant_id, subject_id, resource_id, metadata). 181 **kwargs: Optional keyword arguments. 182 183 Returns: 184 PiiContext: New context instance. 185 """ 186 tenant_id = kwargs.get("tenant_id", args[0] if len(args) > 0 else None) 187 subject_id = kwargs.get("subject_id", args[1] if len(args) > 1 else None) 188 resource_id = kwargs.get("resource_id", args[2] if len(args) > 2 else None) 189 metadata = kwargs.get("metadata", args[3] if len(args) > 3 else None) 190 return cls(stage, tenant_id, subject_id, resource_id, metadata)
Contextual metadata passed during PII sanitization lifecycle stages.
PiiContext( stage: str, tenant_id: Optional[str], subject_id: Optional[str], resource_id: Optional[str], metadata: Optional[Dict[str, Any]])
151 def __init__( 152 self, 153 stage: str, 154 tenant_id: Optional[str], 155 subject_id: Optional[str], 156 resource_id: Optional[str], 157 metadata: Optional[Dict[str, Any]], 158 ): 159 """Initialize PiiContext. 160 161 Args: 162 stage: Execution lifecycle stage identifier (e.g. audit_payload, tool_argument, outbound_response). 163 tenant_id: Tenant isolation identifier associated with the request. 164 subject_id: Subject identifier of the caller or initiator. 165 resource_id: Identifier of the target resource, tool, or document. 166 metadata: Additional contextual metadata dictionary. 167 """ 168 self.stage = stage 169 self.tenant_id = tenant_id 170 self.subject_id = subject_id 171 self.resource_id = resource_id 172 self.metadata = dict(metadata) if metadata is not None else {}
Initialize PiiContext.
Args: stage: Execution lifecycle stage identifier (e.g. audit_payload, tool_argument, outbound_response). tenant_id: Tenant isolation identifier associated with the request. subject_id: Subject identifier of the caller or initiator. resource_id: Identifier of the target resource, tool, or document. metadata: Additional contextual metadata dictionary.
174 @classmethod 175 def create(cls, stage: str, *args: Any, **kwargs: Any) -> "PiiContext": 176 """Factory method to construct a PiiContext. 177 178 Args: 179 stage: Target execution lifecycle stage. 180 *args: Optional positional arguments (tenant_id, subject_id, resource_id, metadata). 181 **kwargs: Optional keyword arguments. 182 183 Returns: 184 PiiContext: New context instance. 185 """ 186 tenant_id = kwargs.get("tenant_id", args[0] if len(args) > 0 else None) 187 subject_id = kwargs.get("subject_id", args[1] if len(args) > 1 else None) 188 resource_id = kwargs.get("resource_id", args[2] if len(args) > 2 else None) 189 metadata = kwargs.get("metadata", args[3] if len(args) > 3 else None) 190 return cls(stage, tenant_id, subject_id, resource_id, metadata)
Factory method to construct a PiiContext.
Args: stage: Target execution lifecycle stage. *args: Optional positional arguments (tenant_id, subject_id, resource_id, metadata). **kwargs: Optional keyword arguments.
Returns: PiiContext: New context instance.
463def get_pii_sanitizer() -> PiiSanitizerSPI: 464 """Retrieve the globally registered PiiSanitizerSPI instance, defaulting to BuiltInPiiSanitizer. 465 466 Returns: 467 PiiSanitizerSPI: Active PII sanitizer instance. 468 """ 469 global _CURRENT_PII_SANITIZER 470 if _CURRENT_PII_SANITIZER is None: 471 _CURRENT_PII_SANITIZER = BuiltInPiiSanitizer() 472 return _CURRENT_PII_SANITIZER
Retrieve the globally registered PiiSanitizerSPI instance, defaulting to BuiltInPiiSanitizer.
Returns: PiiSanitizerSPI: Active PII sanitizer instance.
475def set_pii_sanitizer(sanitizer: PiiSanitizerSPI) -> None: 476 """Register a custom PiiSanitizerSPI instance globally. 477 478 Args: 479 sanitizer: Custom PiiSanitizerSPI implementation to use across the framework. 480 """ 481 global _CURRENT_PII_SANITIZER 482 _CURRENT_PII_SANITIZER = sanitizer
Register a custom PiiSanitizerSPI instance globally.
Args: sanitizer: Custom PiiSanitizerSPI implementation to use across the framework.
def
reset_pii_sanitizer() -> None:
485def reset_pii_sanitizer() -> None: 486 """Reset global PII sanitizer back to the default BuiltInPiiSanitizer instance.""" 487 global _CURRENT_PII_SANITIZER 488 _CURRENT_PII_SANITIZER = None
Reset global PII sanitizer back to the default BuiltInPiiSanitizer instance.
def
get_pii_scope() -> str:
491def get_pii_scope() -> str: 492 """Retrieve active PII redaction scope ('audit_only', 'full_in_flight', or 'disabled'). 493 494 Returns: 495 str: Active PII scope identifier string. 496 """ 497 global _CURRENT_PII_SCOPE 498 return _CURRENT_PII_SCOPE
Retrieve active PII redaction scope ('audit_only', 'full_in_flight', or 'disabled').
Returns: str: Active PII scope identifier string.
def
set_pii_scope(scope: str) -> None:
501def set_pii_scope(scope: str) -> None: 502 """Set active PII redaction scope ('audit_only', 'full_in_flight', or 'disabled'). 503 504 Args: 505 scope: Target PII scope string from PII_SCOPE_* constants. 506 """ 507 global _CURRENT_PII_SCOPE 508 _CURRENT_PII_SCOPE = scope
Set active PII redaction scope ('audit_only', 'full_in_flight', or 'disabled').
Args: scope: Target PII scope string from PII_SCOPE_* constants.
def
reset_pii_scope() -> None:
511def reset_pii_scope() -> None: 512 """Reset active PII redaction scope back to DEFAULT_PII_REDACTION_SCOPE ('audit_only').""" 513 global _CURRENT_PII_SCOPE 514 _CURRENT_PII_SCOPE = DEFAULT_PII_REDACTION_SCOPE
Reset active PII redaction scope back to DEFAULT_PII_REDACTION_SCOPE ('audit_only').
def
is_valid_luhn(card_number_str: str) -> bool:
70def is_valid_luhn(card_number_str: str) -> bool: 71 """Validate candidate credit/debit card number using the Luhn checksum algorithm. 72 73 Args: 74 card_number_str: String containing card number digits with optional delimiters. 75 76 Returns: 77 bool: True if digits satisfy the Luhn mod-10 formula, False otherwise. 78 """ 79 digits = [int(c) for c in card_number_str if c.isdigit()] 80 if len(digits) < 13 or len(digits) > 19: 81 return False 82 checksum = 0 83 reverse_digits = digits[::-1] 84 for i, digit in enumerate(reverse_digits): 85 if i % 2 == 1: 86 doubled = digit * 2 87 checksum += doubled - 9 if doubled > 9 else doubled 88 else: 89 checksum += digit 90 return checksum % 10 == 0
Validate candidate credit/debit card number using the Luhn checksum algorithm.
Args: card_number_str: String containing card number digits with optional delimiters.
Returns: bool: True if digits satisfy the Luhn mod-10 formula, False otherwise.
def
is_valid_verhoeff(number_str: str) -> bool:
93def is_valid_verhoeff(number_str: str) -> bool: 94 """Validate candidate identification number using the Verhoeff checksum algorithm. 95 96 Args: 97 number_str: Numeric string to validate against dihedral group D5 permutations. 98 99 Returns: 100 bool: True if the Verhoeff checksum digit is valid, False otherwise. 101 """ 102 digits = [int(c) for c in number_str if c.isdigit()] 103 if not digits: 104 return False 105 checksum = 0 106 for i, digit in enumerate(reversed(digits)): 107 checksum = VERHOEFF_D_TABLE[checksum][VERHOEFF_P_TABLE[i % 8][digit]] 108 return checksum == 0
Validate candidate identification number using the Verhoeff checksum algorithm.
Args: number_str: Numeric string to validate against dihedral group D5 permutations.
Returns: bool: True if the Verhoeff checksum digit is valid, False otherwise.
def
is_valid_aadhaar(aadhaar_str: str) -> bool:
111def is_valid_aadhaar(aadhaar_str: str) -> bool: 112 """Validate candidate Indian Aadhaar number (12 digits, starts with 2-9, valid Verhoeff checksum). 113 114 Args: 115 aadhaar_str: Candidate Aadhaar number string with optional spaces or hyphens. 116 117 Returns: 118 bool: True if length is 12, first digit is 2-9, and Verhoeff checksum succeeds; False otherwise. 119 """ 120 digits = [c for c in aadhaar_str if c.isdigit()] 121 if len(digits) != 12: 122 return False 123 if digits[0] in ("0", "1"): 124 return False 125 return is_valid_verhoeff("".join(digits))
Validate candidate Indian Aadhaar number (12 digits, starts with 2-9, valid Verhoeff checksum).
Args: aadhaar_str: Candidate Aadhaar number string with optional spaces or hyphens.
Returns: bool: True if length is 12, first digit is 2-9, and Verhoeff checksum succeeds; False otherwise.
def
is_valid_pan(pan_str: str) -> bool:
128def is_valid_pan(pan_str: str) -> bool: 129 """Validate candidate Indian PAN structure (5 letters, 4 digits, 1 letter). 130 131 Args: 132 pan_str: Candidate Permanent Account Number string. 133 134 Returns: 135 bool: True if matching the standard 10-character alphanumeric PAN format, False otherwise. 136 """ 137 stripped = pan_str.strip() 138 digits_part = [c for c in stripped if c.isdigit()] 139 alpha_part = [c for c in stripped if c.isalpha()] 140 if len(digits_part) != 4 or len(alpha_part) != 6: 141 return False 142 clean = re.sub(r"[-\s]", "", stripped) 143 if len(clean) != 10: 144 return False 145 return bool(re.match(r"^[A-Za-z]{5}\d{4}[A-Za-z]$", clean))
Validate candidate Indian PAN structure (5 letters, 4 digits, 1 letter).
Args: pan_str: Candidate Permanent Account Number string.
Returns: bool: True if matching the standard 10-character alphanumeric PAN format, False otherwise.