langchain_security_extension.security_adapter_langchain

LangChain & LangGraph Adapter package exports.

Re-exports integration components including @secured_tool, SecureRunnable, SecureRetriever, SecureCallbackHandler, DocumentSecurityTransformer, and LangGraphNodeGuard.

View Source

 1"""
 2LangChain & LangGraph Adapter package exports.
 3
 4Re-exports integration components including `@secured_tool`, `SecureRunnable`,
 5`SecureRetriever`, `SecureCallbackHandler`, `DocumentSecurityTransformer`, and `LangGraphNodeGuard`.
 6"""
 7
 8from langchain_security_extension.security_adapter_langchain.callback_handler import SecureCallbackHandler
 9from langchain_security_extension.security_adapter_langchain.tool_decorator import secured_tool, SecuredToolConfig, get_tool_schema
10from langchain_security_extension.security_adapter_langchain.retriever_wrapper import SecureRetriever, SecureDocument
11from langchain_security_extension.security_adapter_langchain.ingestion import (
12    DocumentSecurityTransformer,
13    SecureDocumentChunkEnricher,
14    SecureEnvelopeValidator,
15    SecurityEnvelopeValidator,
16    SecureVectorStoreIngestor,
17    SecureTextSplitter,
18    ChunkSecurityEnricherSPI,
19    ChunkEnrichmentContext,
20    HeadingClassificationEnricher,
21    DlpSecretDetectionEnricher,
22)
23from langchain_security_extension.security_adapter_langchain.runnable_guard import SecureRunnable
24from langchain_security_extension.security_adapter_langchain.langgraph_integration import (
25    SecuredAgentState,
26    LangGraphNodeGuard,
27)
28from langchain_security_extension.security_adapter_langchain.violation_handler import (
29    ToolViolationHandler,
30    RetryBudgetTracker,
31    AbuseThrottleTracker,
32    ViolationResolution,
33)
34
35__all__ = [
36    "SecureCallbackHandler",
37    "secured_tool",
38    "SecuredToolConfig",
39    "get_tool_schema",
40    "SecureRetriever",
41    "SecureDocument",
42    "DocumentSecurityTransformer",
43    "SecureDocumentChunkEnricher",
44    "SecureEnvelopeValidator",
45    "SecurityEnvelopeValidator",
46    "SecureVectorStoreIngestor",
47    "SecureTextSplitter",
48    "ChunkSecurityEnricherSPI",
49    "ChunkEnrichmentContext",
50    "HeadingClassificationEnricher",
51    "DlpSecretDetectionEnricher",
52    "SecureRunnable",
53    "SecuredAgentState",
54    "LangGraphNodeGuard",
55    "ToolViolationHandler",
56    "RetryBudgetTracker",
57    "AbuseThrottleTracker",
58    "ViolationResolution",
59]

class SecureCallbackHandler(langchain_core.callbacks.base.BaseCallbackHandler): View Source

 21class SecureCallbackHandler(BaseCallbackHandler):
 22    """LangChain callback handler performing pre-execution PEP evaluations.
 23
 24    Attributes:
 25        pdp_engine (PolicyDecisionEngine): Configured PDP decision engine instance.
 26        security_identity (SecurityIdentity): Active caller security identity envelope.
 27    """
 28
 29    def __init__(
 30        self,
 31        pdp_engine: PolicyDecisionEngine,
 32        security_identity: SecurityIdentity,
 33    ):
 34        """Initialize SecureCallbackHandler.
 35
 36        Args:
 37            pdp_engine: Policy decision engine instance.
 38            security_identity: Caller identity envelope.
 39        """
 40        self._pdp_engine = pdp_engine
 41        self._security_identity = security_identity
 42
 43    @classmethod
 44    def create(
 45        cls,
 46        pdp_engine: PolicyDecisionEngine,
 47        security_identity: SecurityIdentity,
 48    ):
 49        """Factory method to construct SecureCallbackHandler.
 50
 51        Args:
 52            pdp_engine: Policy decision engine.
 53            security_identity: Security identity object.
 54
 55        Returns:
 56            SecureCallbackHandler: Constructed handler instance.
 57        """
 58        return cls(pdp_engine=pdp_engine, security_identity=security_identity)
 59
 60    @property
 61    def pdp_engine(self) -> PolicyDecisionEngine:
 62        """Get the configured PolicyDecisionEngine."""
 63        return self._pdp_engine
 64
 65    @property
 66    def security_identity(self) -> SecurityIdentity:
 67        """Get the bound SecurityIdentity."""
 68        return self._security_identity
 69
 70    def on_chain_start(
 71        self,
 72        serialized: Dict[str, Any],
 73        inputs: Dict[str, Any],
 74        run_id: Any,
 75        parent_run_id: Any,
 76        tags: Optional[List[str]],
 77        metadata: Optional[Dict[str, Any]],
 78        **kwargs: Any,
 79    ) -> None:
 80        """Agent Invocation Guard intercepting chain entry and evaluating access permissions.
 81
 82        Args:
 83            serialized: Serialized chain metadata mapping.
 84            inputs: Chain input arguments mapping.
 85            run_id: Execution run ID.
 86            parent_run_id: Parent run ID if nested.
 87            tags: Operational tags list.
 88            metadata: Additional execution metadata.
 89
 90        Raises:
 91            SecurityAuthorizationError: If PDP denies chain entry.
 92        """
 93        chain_name = str(serialized.get("name") or serialized.get("id") or "LangChainRunnable")
 94        req_roles = serialized.get("required_roles") or (metadata.get("required_roles") if metadata else None)
 95        min_clr = serialized.get("min_clearance") or (metadata.get("min_clearance") if metadata else None)
 96        tenant = (metadata.get("tenant_id") if metadata else None) or serialized.get("tenant_id") or self._security_identity.tenant_id
 97
 98        request = PolicyEvaluationRequest.create(
 99            subject=self._security_identity,
100            action=ACTION_AGENT_INVOKE,
101            resource={
102                "target_id": chain_name,
103                "tenant_id": tenant,
104                "required_roles": req_roles,
105                "min_clearance": min_clr,
106            },
107            context={"run_id": str(run_id), "metadata": metadata or {}},
108        )
109        decision = self._pdp_engine.evaluate(request)
110        if not decision.is_permitted:
111            raise SecurityAuthorizationError.create(
112                message=f"Agent invocation denied for '{chain_name}': {decision.diagnostic_reason}",
113                diagnostic_reason=decision.diagnostic_reason,
114                obligations=decision.obligations,
115            )
116
117    def on_tool_start(
118        self,
119        serialized: Dict[str, Any],
120        input_str: str,
121        run_id: Any,
122        parent_run_id: Any,
123        tags: Optional[List[str]],
124        metadata: Optional[Dict[str, Any]],
125        inputs: Optional[Dict[str, Any]],
126        **kwargs: Any,
127    ) -> None:
128        """Tool Execution Interceptor hook evaluating tool access and CEL parameter constraints.
129
130        Args:
131            serialized: Serialized tool metadata map.
132            input_str: Raw tool input string.
133            run_id: Run ID.
134            parent_run_id: Parent run ID.
135            tags: Event tags.
136            metadata: Metadata dictionary.
137            inputs: Structured tool inputs map.
138
139        Raises:
140            ToolExecutionDeniedError: If PDP denies tool execution.
141        """
142        tool_name = str(serialized.get("name") or "LangChainTool")
143        tool_args = inputs if inputs is not None else {"raw_input": input_str}
144        req_roles = serialized.get("required_roles") or (metadata.get("required_roles") if metadata else None)
145        min_clr = serialized.get("min_clearance") or (metadata.get("min_clearance") if metadata else None)
146        cel_expr = serialized.get("cel_expression") or (metadata.get("cel_expression") if metadata else None)
147        is_mutative = bool(serialized.get("is_mutative") or (metadata.get("is_mutative") if metadata else True))
148        tenant = (metadata.get("tenant_id") if metadata else None) or serialized.get("tenant_id") or self._security_identity.tenant_id
149
150        request = PolicyEvaluationRequest.create(
151            subject=self._security_identity,
152            action=ACTION_TOOL_EXECUTE,
153            resource={
154                "name": tool_name,
155                "target_id": tool_name,
156                "tenant_id": tenant,
157                "required_roles": req_roles,
158                "min_clearance": min_clr,
159                "cel_expression": cel_expr,
160                "is_mutative": is_mutative,
161                "arguments": tool_args,
162                "tool_metadata": {
163                    "name": tool_name,
164                    "isMutative": is_mutative,
165                },
166            },
167            context={"run_id": str(run_id), "metadata": metadata or {}},
168        )
169        decision = self._pdp_engine.evaluate(request)
170        if not decision.is_permitted:
171            strategy = decision.obligations.get("strategy")
172            if strategy == "RETURN_FEEDBACK" and not is_mutative:
173                refusal_payload = {
174                    "status": "ACCESS_DENIED",
175                    "tool": tool_name,
176                    "reason": f"Tool execution denied for '{tool_name}': {decision.diagnostic_reason}",
177                    "diagnostic_reason": decision.diagnostic_reason,
178                }
179                raise ToolExecutionFeedbackException.create_payload(tool_name, refusal_payload)
180            raise ToolExecutionDeniedError.for_tool(
181                tool_name=tool_name,
182                message=f"Tool execution denied for '{tool_name}': {decision.diagnostic_reason}",
183                diagnostic_reason=decision.diagnostic_reason,
184            )

LangChain callback handler performing pre-execution PEP evaluations.

Attributes: pdp_engine (PolicyDecisionEngine): Configured PDP decision engine instance. security_identity (SecurityIdentity): Active caller security identity envelope.

SecureCallbackHandler( pdp_engine: langchain_security_extension.security_core.PolicyDecisionEngine, security_identity: langchain_security_extension.security_core.SecurityIdentity) View Source

29    def __init__(
30        self,
31        pdp_engine: PolicyDecisionEngine,
32        security_identity: SecurityIdentity,
33    ):
34        """Initialize SecureCallbackHandler.
35
36        Args:
37            pdp_engine: Policy decision engine instance.
38            security_identity: Caller identity envelope.
39        """
40        self._pdp_engine = pdp_engine
41        self._security_identity = security_identity

Initialize SecureCallbackHandler.

Args: pdp_engine: Policy decision engine instance. security_identity: Caller identity envelope.

@classmethod

def create( cls, pdp_engine: langchain_security_extension.security_core.PolicyDecisionEngine, security_identity: langchain_security_extension.security_core.SecurityIdentity): View Source

43    @classmethod
44    def create(
45        cls,
46        pdp_engine: PolicyDecisionEngine,
47        security_identity: SecurityIdentity,
48    ):
49        """Factory method to construct SecureCallbackHandler.
50
51        Args:
52            pdp_engine: Policy decision engine.
53            security_identity: Security identity object.
54
55        Returns:
56            SecureCallbackHandler: Constructed handler instance.
57        """
58        return cls(pdp_engine=pdp_engine, security_identity=security_identity)

Factory method to construct SecureCallbackHandler.

Args: pdp_engine: Policy decision engine. security_identity: Security identity object.

Returns: SecureCallbackHandler: Constructed handler instance.

pdp_engine: langchain_security_extension.security_core.PolicyDecisionEngine View Source

60    @property
61    def pdp_engine(self) -> PolicyDecisionEngine:
62        """Get the configured PolicyDecisionEngine."""
63        return self._pdp_engine

Get the configured PolicyDecisionEngine.

security_identity: langchain_security_extension.security_core.SecurityIdentity View Source

65    @property
66    def security_identity(self) -> SecurityIdentity:
67        """Get the bound SecurityIdentity."""
68        return self._security_identity

Get the bound SecurityIdentity.

def on_chain_start( self, serialized: Dict[str, Any], inputs: Dict[str, Any], run_id: Any, parent_run_id: Any, tags: Optional[List[str]], metadata: Optional[Dict[str, Any]], **kwargs: Any) -> None: View Source

 70    def on_chain_start(
 71        self,
 72        serialized: Dict[str, Any],
 73        inputs: Dict[str, Any],
 74        run_id: Any,
 75        parent_run_id: Any,
 76        tags: Optional[List[str]],
 77        metadata: Optional[Dict[str, Any]],
 78        **kwargs: Any,
 79    ) -> None:
 80        """Agent Invocation Guard intercepting chain entry and evaluating access permissions.
 81
 82        Args:
 83            serialized: Serialized chain metadata mapping.
 84            inputs: Chain input arguments mapping.
 85            run_id: Execution run ID.
 86            parent_run_id: Parent run ID if nested.
 87            tags: Operational tags list.
 88            metadata: Additional execution metadata.
 89
 90        Raises:
 91            SecurityAuthorizationError: If PDP denies chain entry.
 92        """
 93        chain_name = str(serialized.get("name") or serialized.get("id") or "LangChainRunnable")
 94        req_roles = serialized.get("required_roles") or (metadata.get("required_roles") if metadata else None)
 95        min_clr = serialized.get("min_clearance") or (metadata.get("min_clearance") if metadata else None)
 96        tenant = (metadata.get("tenant_id") if metadata else None) or serialized.get("tenant_id") or self._security_identity.tenant_id
 97
 98        request = PolicyEvaluationRequest.create(
 99            subject=self._security_identity,
100            action=ACTION_AGENT_INVOKE,
101            resource={
102                "target_id": chain_name,
103                "tenant_id": tenant,
104                "required_roles": req_roles,
105                "min_clearance": min_clr,
106            },
107            context={"run_id": str(run_id), "metadata": metadata or {}},
108        )
109        decision = self._pdp_engine.evaluate(request)
110        if not decision.is_permitted:
111            raise SecurityAuthorizationError.create(
112                message=f"Agent invocation denied for '{chain_name}': {decision.diagnostic_reason}",
113                diagnostic_reason=decision.diagnostic_reason,
114                obligations=decision.obligations,
115            )

Agent Invocation Guard intercepting chain entry and evaluating access permissions.

Args: serialized: Serialized chain metadata mapping. inputs: Chain input arguments mapping. run_id: Execution run ID. parent_run_id: Parent run ID if nested. tags: Operational tags list. metadata: Additional execution metadata.

Raises: SecurityAuthorizationError: If PDP denies chain entry.

def on_tool_start( self, serialized: Dict[str, Any], input_str: str, run_id: Any, parent_run_id: Any, tags: Optional[List[str]], metadata: Optional[Dict[str, Any]], inputs: Optional[Dict[str, Any]], **kwargs: Any) -> None: View Source

117    def on_tool_start(
118        self,
119        serialized: Dict[str, Any],
120        input_str: str,
121        run_id: Any,
122        parent_run_id: Any,
123        tags: Optional[List[str]],
124        metadata: Optional[Dict[str, Any]],
125        inputs: Optional[Dict[str, Any]],
126        **kwargs: Any,
127    ) -> None:
128        """Tool Execution Interceptor hook evaluating tool access and CEL parameter constraints.
129
130        Args:
131            serialized: Serialized tool metadata map.
132            input_str: Raw tool input string.
133            run_id: Run ID.
134            parent_run_id: Parent run ID.
135            tags: Event tags.
136            metadata: Metadata dictionary.
137            inputs: Structured tool inputs map.
138
139        Raises:
140            ToolExecutionDeniedError: If PDP denies tool execution.
141        """
142        tool_name = str(serialized.get("name") or "LangChainTool")
143        tool_args = inputs if inputs is not None else {"raw_input": input_str}
144        req_roles = serialized.get("required_roles") or (metadata.get("required_roles") if metadata else None)
145        min_clr = serialized.get("min_clearance") or (metadata.get("min_clearance") if metadata else None)
146        cel_expr = serialized.get("cel_expression") or (metadata.get("cel_expression") if metadata else None)
147        is_mutative = bool(serialized.get("is_mutative") or (metadata.get("is_mutative") if metadata else True))
148        tenant = (metadata.get("tenant_id") if metadata else None) or serialized.get("tenant_id") or self._security_identity.tenant_id
149
150        request = PolicyEvaluationRequest.create(
151            subject=self._security_identity,
152            action=ACTION_TOOL_EXECUTE,
153            resource={
154                "name": tool_name,
155                "target_id": tool_name,
156                "tenant_id": tenant,
157                "required_roles": req_roles,
158                "min_clearance": min_clr,
159                "cel_expression": cel_expr,
160                "is_mutative": is_mutative,
161                "arguments": tool_args,
162                "tool_metadata": {
163                    "name": tool_name,
164                    "isMutative": is_mutative,
165                },
166            },
167            context={"run_id": str(run_id), "metadata": metadata or {}},
168        )
169        decision = self._pdp_engine.evaluate(request)
170        if not decision.is_permitted:
171            strategy = decision.obligations.get("strategy")
172            if strategy == "RETURN_FEEDBACK" and not is_mutative:
173                refusal_payload = {
174                    "status": "ACCESS_DENIED",
175                    "tool": tool_name,
176                    "reason": f"Tool execution denied for '{tool_name}': {decision.diagnostic_reason}",
177                    "diagnostic_reason": decision.diagnostic_reason,
178                }
179                raise ToolExecutionFeedbackException.create_payload(tool_name, refusal_payload)
180            raise ToolExecutionDeniedError.for_tool(
181                tool_name=tool_name,
182                message=f"Tool execution denied for '{tool_name}': {decision.diagnostic_reason}",
183                diagnostic_reason=decision.diagnostic_reason,
184            )

Tool Execution Interceptor hook evaluating tool access and CEL parameter constraints.

Args: serialized: Serialized tool metadata map. input_str: Raw tool input string. run_id: Run ID. parent_run_id: Parent run ID. tags: Event tags. metadata: Metadata dictionary. inputs: Structured tool inputs map.

Raises: ToolExecutionDeniedError: If PDP denies tool execution.

def secured_tool( required_roles: List[str], min_clearance: int, cel_expression: str, is_mutative: bool, *args: Any, **kwargs: Any) -> Callable: View Source

187def secured_tool(
188    required_roles: List[str],
189    min_clearance: int,
190    cel_expression: str,
191    is_mutative: bool,
192    *args: Any,
193    **kwargs: Any,
194) -> Callable:
195    """Decorator securing tool functions with pre-execution authorization check.
196
197    Args:
198        required_roles: List of mandatory role strings for execution.
199        min_clearance: Clearance floor rank required.
200        cel_expression: CEL condition evaluating caller identity and arguments.
201        is_mutative: True if tool modifies system state, triggering HARD_ABORT on failure.
202        *args: Optional positional arguments (pdp_engine, entity_resolver, masked_params).
203        **kwargs: Optional keyword arguments (pdp_engine, entity_resolver, masked_params).
204
205    Returns:
206        Callable: Wrapped target tool function with PEP interceptor guard.
207    """
208    pdp_engine: Optional[PolicyDecisionEngine] = kwargs.get("pdp_engine", args[0] if len(args) > 0 else None)
209    entity_resolver: Optional[Any] = kwargs.get("entity_resolver", args[1] if len(args) > 1 else None)
210    masked_params: List[str] = list(kwargs.get("masked_params", args[2] if len(args) > 2 else []))
211
212    def decorator(func: Callable) -> Callable:
213        @functools.wraps(func)
214        def wrapper(*args: Any, **kwargs: Any) -> Any:
215            caller_config = kwargs.get("config") or kwargs.get("runnable_config")
216            caller_identity = kwargs.pop("security_identity", None)
217
218            if caller_identity is None and caller_config:
219                caller_identity = IdentityBridge.extract_from_config(caller_config)
220
221            if caller_identity is None and args and isinstance(args[0], SecurityIdentity):
222                caller_identity = args[0]
223                args = args[1:]
224
225            caller_identity = IdentityBridge.resolve_identity(caller_identity)
226
227            if caller_config is None and caller_identity is not None:
228                caller_config = IdentityBridge.create_runnable_config(caller_identity)
229
230            # Resolve PDP engine if None
231            engine = pdp_engine
232            if engine is None and "config" in kwargs and isinstance(kwargs["config"], dict):
233                cfg_conf = kwargs["config"].get("configurable", {})
234                if isinstance(cfg_conf, dict):
235                    engine = cfg_conf.get("sec:pdp_engine") or cfg_conf.get("pdp_engine")
236
237            if engine is None and args and hasattr(args[0], "pdp_engine") and getattr(args[0], "pdp_engine") is not None:
238                engine = getattr(args[0], "pdp_engine")
239
240            if engine is None and "config" in kwargs and isinstance(kwargs["config"], dict):
241                cfg_conf = kwargs["config"].get("configurable", {})
242                if isinstance(cfg_conf, dict):
243                    vh = cfg_conf.get("sec:violation_handler")
244                    if vh and hasattr(vh, "pdp_engine"):
245                        engine = vh.pdp_engine
246                if engine is None:
247                    callbacks_list = kwargs["config"].get("callbacks", [])
248                    if isinstance(callbacks_list, list):
249                        for cb in callbacks_list:
250                            if hasattr(cb, "pdp_engine"):
251                                engine = cb.pdp_engine
252                                break
253
254            if engine is None:
255                from langchain_security_extension.security_core.registry import SecurityRegistry
256                engine = SecurityRegistry.get_pdp_engine()
257
258            if engine is None:
259                from langchain_security_extension.security_core.pdp import EmbeddedPolicyEngine
260                engine = EmbeddedPolicyEngine.create()
261
262            resolved_resolver = entity_resolver
263            if resolved_resolver is None and args and hasattr(args[0], "entity_resolver") and getattr(args[0], "entity_resolver") is not None:
264                resolved_resolver = getattr(args[0], "entity_resolver")
265
266            if resolved_resolver is None and "config" in kwargs and isinstance(kwargs["config"], dict):
267                cfg_conf = kwargs["config"].get("configurable", {})
268                if isinstance(cfg_conf, dict):
269                    resolved_resolver = cfg_conf.get("sec:entity_resolver") or cfg_conf.get("entity_resolver")
270
271            if resolved_resolver is None and engine is not None and hasattr(engine, "_cel_evaluator") and engine._cel_evaluator:
272                resolved_resolver = getattr(engine._cel_evaluator, "entity_resolver", None)
273
274            if resolved_resolver is None:
275                from langchain_security_extension.security_core.registry import SecurityRegistry
276                resolved_resolver = SecurityRegistry.get_entity_resolver()
277
278            if resolved_resolver and hasattr(engine, "_cel_evaluator") and engine._cel_evaluator:
279                engine._cel_evaluator.entity_resolver = resolved_resolver
280
281            import inspect
282
283            # Bind positional and keyword arguments to parameter names
284            bound_args: Dict[str, Any] = {}
285            try:
286                sig = inspect.signature(func)
287                params = list(sig.parameters.keys())
288                if params and params[0] in ("self", "cls"):
289                    if not args or not (hasattr(args[0], "__class__") and args[0].__class__.__name__ in func.__qualname__):
290                        args_to_bind = (None,) + args
291                    else:
292                        args_to_bind = args
293                else:
294                    args_to_bind = args
295
296                clean_kwargs = {k: v for k, v in kwargs.items() if k in sig.parameters}
297                bound = sig.bind_partial(*args_to_bind, **clean_kwargs)
298                bound.apply_defaults()
299                bound_args = dict(bound.arguments)
300            except Exception:
301                bound_args = dict(kwargs)
302
303            bound_args.pop("config", None)
304            bound_args.pop("self", None)
305            bound_args.pop("cls", None)
306
307            aliases: Dict[str, Any] = {}
308            for k, v in list(bound_args.items()):
309                if "_" in k:
310                    parts = k.split("_")
311                    camel = parts[0] + "".join(p.capitalize() for p in parts[1:])
312                    aliases[camel] = v
313                else:
314                    import re
315                    snake = re.sub(r'(?<!^)(?=[A-Z])', '_', k).lower()
316                    aliases[snake] = v
317            bound_args.update(aliases)
318
319            # Resolve target tenant: check decorator kwargs, bound arguments (e.g. tenant_id/tenantId), or default to caller
320            resolved_tenant = kwargs.get("target_tenant") or bound_args.get("tenant_id") or bound_args.get("tenantId") or caller_identity.tenant_id
321
322            tool_name = func.__name__
323            tool_module = getattr(func, "__module__", "")
324
325            resource_descriptor = {
326                "name": tool_name,
327                "target_id": tool_name,
328                "tenant_id": resolved_tenant,
329                "required_roles": required_roles,
330                "min_clearance": min_clearance,
331                "cel_expression": cel_expression,
332                "is_mutative": is_mutative,
333                "arguments": bound_args,
334                "masked_params": masked_params,
335                "tool_metadata": {
336                    "name": tool_name,
337                    "isMutative": is_mutative,
338                    "module": tool_module,
339                    "classificationLevel": min_clearance,
340                },
341            }
342
343            request = PolicyEvaluationRequest.create(
344                subject=caller_identity,
345                action=ACTION_TOOL_EXECUTE,
346                resource=resource_descriptor,
347                context={"config": kwargs.get("config")},
348            )
349
350            decision = engine.evaluate(request)
351
352            if not decision.is_permitted:
353                # Retry budget tracking & parameter escalation checking in config
354                cfg = kwargs.get("config")
355                retries = 0
356                prev_args = None
357                param_escalation_detected = False
358
359                if isinstance(cfg, dict) and isinstance(cfg.get("configurable"), dict):
360                    retries = cfg["configurable"].get("sec:retry_count", 0) + 1
361                    cfg["configurable"]["sec:retry_count"] = retries
362                    prev_args = cfg["configurable"].get("sec:prev_args")
363                    cfg["configurable"]["sec:prev_args"] = bound_args
364
365                    if prev_args is not None and prev_args != bound_args:
366                        param_escalation_detected = True
367
368                strategy = decision.obligations.get("strategy", STRATEGY_HARD_ABORT if is_mutative else STRATEGY_RETURN_FEEDBACK)
369                reason_msg = f"Policy violation: Access denied for tool '{tool_name}' ({decision.diagnostic_reason})"
370
371                if param_escalation_detected and hasattr(engine, "audit_manager") and engine.audit_manager:
372                    from langchain_security_extension.security_core.audit import AuditEvent
373                    from langchain_security_extension.security_core.constants import (
374                        EVENT_PARAMETER_PRIVILEGE_ESCALATION,
375                        PEP_TOOL_INTERCEPTOR,
376                        SEVERITY_SECURITY_ALERT,
377                    )
378                    engine.audit_manager.record_event(AuditEvent.create(
379                        subject_id=caller_identity.subject_id,
380                        tenant_id=caller_identity.tenant_id,
381                        action=ACTION_TOOL_EXECUTE,
382                        resource=tool_name,
383                        decision="DENIED",
384                        diagnostic_reason="Parameter privilege escalation attempt detected: arguments mutated across retries",
385                        severity=SEVERITY_SECURITY_ALERT,
386                        metadata={"previous_args": prev_args, "current_args": bound_args},
387                        enforcement_point=PEP_TOOL_INTERCEPTOR,
388                        reason_code=EVENT_PARAMETER_PRIVILEGE_ESCALATION,
389                    ))
390
391                # Record violation with registered ToolViolationHandler if present in config
392                vh = None
393                if isinstance(cfg, dict) and isinstance(cfg.get("configurable"), dict):
394                    vh = cfg["configurable"].get("sec:violation_handler")
395
396                resolution = None
397                if vh is not None and hasattr(vh, "record_tool_violation"):
398                    resolution = vh.record_tool_violation(
399                        tool_name=tool_name,
400                        diagnostic_reason=str(decision.diagnostic_reason or ""),
401                        is_mutative=is_mutative,
402                        config=cfg,
403                        caller_identity=caller_identity,
404                    )
405                elif isinstance(cfg, dict) and isinstance(cfg.get("configurable"), dict):
406                    from langchain_security_extension.security_adapter_langchain.violation_handler import (
407                        ViolationResolution,
408                        STATUS_BLOCKED_BY_POLICY,
409                        POLICY_STATUS_BLOCKED,
410                        POLICY_STATUS_FEEDBACK,
411                        EXECUTION_ACTION_ABORTED,
412                        EXECUTION_ACTION_FEEDBACK,
413                    )
414                    should_abort_flag = bool(strategy == STRATEGY_HARD_ABORT or is_mutative or retries > 2)
415                    resolution = ViolationResolution.create(
416                        status=STATUS_BLOCKED_BY_POLICY if should_abort_flag else "ACCESS_DENIED",
417                        action_entry=f"Blocked execution of tool '{tool_name}' ({decision.diagnostic_reason})",
418                        root_cause=f"Policy violation: Access denied for tool '{tool_name}' ({decision.diagnostic_reason})",
419                        user_message=reason_msg,
420                        policy_decision=POLICY_STATUS_BLOCKED if should_abort_flag else POLICY_STATUS_FEEDBACK,
421                        execution_action=EXECUTION_ACTION_ABORTED if should_abort_flag else EXECUTION_ACTION_FEEDBACK,
422                        should_abort=should_abort_flag,
423                        diagnostic_reason=decision.diagnostic_reason,
424                        tool_name=tool_name,
425                    )
426                    cfg["configurable"]["sec:last_violation"] = resolution
427
428                should_hard_abort = bool(
429                    strategy == STRATEGY_HARD_ABORT
430                    or is_mutative
431                    or retries > 2
432                    or (resolution and resolution.should_abort)
433                )
434
435                if should_hard_abort:
436                    if retries > 2 and hasattr(engine, "audit_manager") and engine.audit_manager:
437                        from langchain_security_extension.security_core.audit import AuditEvent
438                        from langchain_security_extension.security_core.constants import (
439                            EVENT_POLICY_BRUTE_FORCE_DETECTED,
440                            PEP_TOOL_INTERCEPTOR,
441                            SEVERITY_CRITICAL_ALERT,
442                        )
443                        engine.audit_manager.record_event(AuditEvent.create(
444                            subject_id=caller_identity.subject_id,
445                            tenant_id=caller_identity.tenant_id,
446                            action=ACTION_TOOL_EXECUTE,
447                            resource=tool_name,
448                            decision="HARD_ABORT",
449                            diagnostic_reason="Retry budget exceeded: Policy brute force probe detected",
450                            severity=SEVERITY_CRITICAL_ALERT,
451                            metadata={"retries": retries, "arguments": bound_args},
452                            enforcement_point=PEP_TOOL_INTERCEPTOR,
453                            reason_code=EVENT_POLICY_BRUTE_FORCE_DETECTED,
454                        ))
455                    raise ToolExecutionDeniedError.for_tool(
456                        tool_name=tool_name,
457                        message=reason_msg,
458                        diagnostic_reason=decision.diagnostic_reason,
459                    )
460                else:
461                    refusal_payload = {
462                        "status": "ACCESS_DENIED",
463                        "tool": tool_name,
464                        "reason": reason_msg,
465                        "diagnostic_reason": decision.diagnostic_reason,
466                    }
467                    return refusal_payload
468
469            cfg = kwargs.get("config")
470            if isinstance(cfg, dict) and isinstance(cfg.get("configurable"), dict):
471                cfg["configurable"]["sec:last_violation"] = None
472
473            func_kwargs = dict(kwargs)
474            func_kwargs.pop("config", None)
475            func_kwargs.pop("security_identity", None)
476            func_kwargs.pop("runnable_config", None)
477
478            # Re-inject caller SecurityIdentity and RunnableConfig if target function signature declares them
479            try:
480                sig = inspect.signature(func)
481                if "security_identity" in sig.parameters:
482                    func_kwargs["security_identity"] = caller_identity
483                if "config" in sig.parameters:
484                    func_kwargs["config"] = caller_config
485                elif "runnable_config" in sig.parameters:
486                    func_kwargs["runnable_config"] = caller_config
487            except Exception:
488                pass
489
490            if get_pii_scope() == PII_SCOPE_FULL_IN_FLIGHT:
491                sanitizer = get_pii_sanitizer()
492                ctx_in = PiiContext.create(
493                    PII_STAGE_TOOL_ARGUMENT,
494                    caller_identity.tenant_id,
495                    caller_identity.subject_id,
496                    tool_name,
497                    {"required_roles": required_roles, "min_clearance": min_clearance},
498                )
499                func_kwargs = sanitizer.sanitize_payload(func_kwargs, ctx_in)
500
501            result = func(*args, **func_kwargs)
502
503            if get_pii_scope() == PII_SCOPE_FULL_IN_FLIGHT:
504                sanitizer = get_pii_sanitizer()
505                ctx_out = PiiContext.create(
506                    PII_STAGE_TOOL_OUTPUT,
507                    caller_identity.tenant_id,
508                    caller_identity.subject_id,
509                    tool_name,
510                    {"required_roles": required_roles, "min_clearance": min_clearance},
511                )
512                if isinstance(result, str):
513                    result = sanitizer.sanitize_text(result, ctx_out)
514                elif isinstance(result, dict):
515                    result = sanitizer.sanitize_payload(result, ctx_out)
516                elif isinstance(result, list):
517                    result = [
518                        sanitizer.sanitize_payload(item, ctx_out) if isinstance(item, dict)
519                        else (sanitizer.sanitize_text(item, ctx_out) if isinstance(item, str) else item)
520                        for item in result
521                    ]
522
523            return result
524
525        try:
526            import inspect
527            orig_sig = inspect.signature(func)
528            filtered_params = [
529                p for p in orig_sig.parameters.values()
530                if p.name not in HIDDEN_SECURITY_PARAMS
531            ]
532            wrapper.__signature__ = orig_sig.replace(parameters=filtered_params)
533        except Exception:
534            pass
535
536        return wrapper
537
538    return decorator

Decorator securing tool functions with pre-execution authorization check.

Args: required_roles: List of mandatory role strings for execution. min_clearance: Clearance floor rank required. cel_expression: CEL condition evaluating caller identity and arguments. is_mutative: True if tool modifies system state, triggering HARD_ABORT on failure. *args: Optional positional arguments (pdp_engine, entity_resolver, masked_params). **kwargs: Optional keyword arguments (pdp_engine, entity_resolver, masked_params).

Returns: Callable: Wrapped target tool function with PEP interceptor guard.

class SecuredToolConfig: View Source

 82class SecuredToolConfig:
 83    """Security configuration specification for secured tool functions.
 84
 85    Attributes:
 86        required_roles (List[str]): List of security roles required to execute the tool.
 87        min_clearance (int): Minimum clearance level rank required for tool execution.
 88        cel_expression (str): CEL condition expression string evaluating invocation parameters.
 89        is_mutative (bool): Flag indicating if tool performs state mutation operations.
 90        pdp_engine (PolicyDecisionEngine): PDP engine instance for evaluation.
 91    """
 92
 93    def __init__(
 94        self,
 95        required_roles: List[str],
 96        min_clearance: int,
 97        cel_expression: str,
 98        is_mutative: bool,
 99        pdp_engine: PolicyDecisionEngine,
100        *args: Any,
101        **kwargs: Any,
102    ):
103        """Initialize SecuredToolConfig.
104
105        Args:
106            required_roles: Roles list.
107            min_clearance: Clearance floor.
108            cel_expression: CEL condition string.
109            is_mutative: Mutative action flag.
110            pdp_engine: PDP decision engine instance.
111            *args: Additional arguments (e.g. masked_params).
112            **kwargs: Additional keyword arguments.
113        """
114        self._required_roles = list(required_roles)
115        self._min_clearance = min_clearance
116        self._cel_expression = cel_expression
117        self._is_mutative = is_mutative
118        self._pdp_engine = pdp_engine
119        self._masked_params = list(kwargs.get("masked_params", args[0] if len(args) > 0 else []))
120
121    @classmethod
122    def create(
123        cls,
124        required_roles: List[str],
125        min_clearance: int,
126        cel_expression: str,
127        is_mutative: bool,
128        pdp_engine: PolicyDecisionEngine,
129        *args: Any,
130        **kwargs: Any,
131    ):
132        """Factory method to construct SecuredToolConfig.
133
134        Args:
135            required_roles: List of mandatory role strings for execution.
136            min_clearance: Minimum clearance level rank required.
137            cel_expression: Common Expression Language condition string.
138            is_mutative: True if tool modifies system state.
139            pdp_engine: PDP decision engine instance.
140            *args: Optional positional arguments (masked_params).
141            **kwargs: Optional keyword arguments (masked_params).
142
143        Returns:
144            SecuredToolConfig: New tool security configuration instance.
145        """
146        return cls(
147            required_roles,
148            min_clearance,
149            cel_expression,
150            is_mutative,
151            pdp_engine,
152            *args,
153            **kwargs,
154        )
155
156    @property
157    def required_roles(self) -> List[str]:
158        """Get copy of required roles list."""
159        return list(self._required_roles)
160
161    @property
162    def min_clearance(self) -> int:
163        """Get minimum clearance level rank."""
164        return self._min_clearance
165
166    @property
167    def cel_expression(self) -> str:
168        """Get CEL parameter constraint expression string."""
169        return self._cel_expression
170
171    @property
172    def is_mutative(self) -> bool:
173        """Check if tool action is mutative."""
174        return self._is_mutative
175
176    @property
177    def pdp_engine(self) -> PolicyDecisionEngine:
178        """Get configured PolicyDecisionEngine instance."""
179        return self._pdp_engine
180
181    @property
182    def masked_params(self) -> List[str]:
183        """Get list of masked parameter names."""
184        return list(self._masked_params)

Security configuration specification for secured tool functions.

Attributes: required_roles (List[str]): List of security roles required to execute the tool. min_clearance (int): Minimum clearance level rank required for tool execution. cel_expression (str): CEL condition expression string evaluating invocation parameters. is_mutative (bool): Flag indicating if tool performs state mutation operations. pdp_engine (PolicyDecisionEngine): PDP engine instance for evaluation.

SecuredToolConfig( required_roles: List[str], min_clearance: int, cel_expression: str, is_mutative: bool, pdp_engine: langchain_security_extension.security_core.PolicyDecisionEngine, *args: Any, **kwargs: Any) View Source

 93    def __init__(
 94        self,
 95        required_roles: List[str],
 96        min_clearance: int,
 97        cel_expression: str,
 98        is_mutative: bool,
 99        pdp_engine: PolicyDecisionEngine,
100        *args: Any,
101        **kwargs: Any,
102    ):
103        """Initialize SecuredToolConfig.
104
105        Args:
106            required_roles: Roles list.
107            min_clearance: Clearance floor.
108            cel_expression: CEL condition string.
109            is_mutative: Mutative action flag.
110            pdp_engine: PDP decision engine instance.
111            *args: Additional arguments (e.g. masked_params).
112            **kwargs: Additional keyword arguments.
113        """
114        self._required_roles = list(required_roles)
115        self._min_clearance = min_clearance
116        self._cel_expression = cel_expression
117        self._is_mutative = is_mutative
118        self._pdp_engine = pdp_engine
119        self._masked_params = list(kwargs.get("masked_params", args[0] if len(args) > 0 else []))

Initialize SecuredToolConfig.

Args: required_roles: Roles list. min_clearance: Clearance floor. cel_expression: CEL condition string. is_mutative: Mutative action flag. pdp_engine: PDP decision engine instance. *args: Additional arguments (e.g. masked_params). **kwargs: Additional keyword arguments.

@classmethod

def create( cls, required_roles: List[str], min_clearance: int, cel_expression: str, is_mutative: bool, pdp_engine: langchain_security_extension.security_core.PolicyDecisionEngine, *args: Any, **kwargs: Any): View Source

121    @classmethod
122    def create(
123        cls,
124        required_roles: List[str],
125        min_clearance: int,
126        cel_expression: str,
127        is_mutative: bool,
128        pdp_engine: PolicyDecisionEngine,
129        *args: Any,
130        **kwargs: Any,
131    ):
132        """Factory method to construct SecuredToolConfig.
133
134        Args:
135            required_roles: List of mandatory role strings for execution.
136            min_clearance: Minimum clearance level rank required.
137            cel_expression: Common Expression Language condition string.
138            is_mutative: True if tool modifies system state.
139            pdp_engine: PDP decision engine instance.
140            *args: Optional positional arguments (masked_params).
141            **kwargs: Optional keyword arguments (masked_params).
142
143        Returns:
144            SecuredToolConfig: New tool security configuration instance.
145        """
146        return cls(
147            required_roles,
148            min_clearance,
149            cel_expression,
150            is_mutative,
151            pdp_engine,
152            *args,
153            **kwargs,
154        )

Factory method to construct SecuredToolConfig.

Args: required_roles: List of mandatory role strings for execution. min_clearance: Minimum clearance level rank required. cel_expression: Common Expression Language condition string. is_mutative: True if tool modifies system state. pdp_engine: PDP decision engine instance. *args: Optional positional arguments (masked_params). **kwargs: Optional keyword arguments (masked_params).

Returns: SecuredToolConfig: New tool security configuration instance.

required_roles: List[str] View Source

156    @property
157    def required_roles(self) -> List[str]:
158        """Get copy of required roles list."""
159        return list(self._required_roles)

Get copy of required roles list.

min_clearance: int View Source

161    @property
162    def min_clearance(self) -> int:
163        """Get minimum clearance level rank."""
164        return self._min_clearance

Get minimum clearance level rank.

cel_expression: str View Source

166    @property
167    def cel_expression(self) -> str:
168        """Get CEL parameter constraint expression string."""
169        return self._cel_expression

Get CEL parameter constraint expression string.

is_mutative: bool View Source

171    @property
172    def is_mutative(self) -> bool:
173        """Check if tool action is mutative."""
174        return self._is_mutative

Check if tool action is mutative.

pdp_engine: langchain_security_extension.security_core.PolicyDecisionEngine View Source

176    @property
177    def pdp_engine(self) -> PolicyDecisionEngine:
178        """Get configured PolicyDecisionEngine instance."""
179        return self._pdp_engine

Get configured PolicyDecisionEngine instance.

masked_params: List[str] View Source

181    @property
182    def masked_params(self) -> List[str]:
183        """Get list of masked parameter names."""
184        return list(self._masked_params)

Get list of masked parameter names.

def get_tool_schema(tool: Any) -> Dict[str, Any]: View Source

27def get_tool_schema(tool: Any) -> Dict[str, Any]:
28    """Extract tool schema concealing internal security parameters from model context.
29
30    Inspects tool function signatures or Pydantic `args_schema` and suppresses internal
31    security parameter names (e.g. security_identity, config, runnable_config) from being
32    exposed in model-facing JSON function schemas.
33
34    Args:
35        tool: Target tool callable, BaseTool instance, or structured tool function.
36
37    Returns:
38        Dict[str, Any]: Model-safe JSON Schema dictionary describing tool arguments.
39    """
40    tool_name = getattr(tool, "name", None) or getattr(tool, "__name__", "tool")
41    description = getattr(tool, "description", None) or getattr(tool, "__doc__", "") or ""
42
43    properties: Dict[str, Any] = {}
44    required: List[str] = []
45
46    if hasattr(tool, "args_schema") and tool.args_schema:
47        raw_schema = tool.args_schema.schema() if hasattr(tool.args_schema, "schema") else tool.args_schema
48        if isinstance(raw_schema, dict):
49            props = dict(raw_schema.get("properties", {}))
50            reqs = list(raw_schema.get("required", []))
51            for hidden in HIDDEN_SECURITY_PARAMS:
52                props.pop(hidden, None)
53                if hidden in reqs:
54                    reqs.remove(hidden)
55            properties = props
56            required = reqs
57    else:
58        func = getattr(tool, "func", tool)
59        import inspect
60        try:
61            sig = inspect.signature(func)
62            for param_name, param in sig.parameters.items():
63                if param_name in ("self", "cls") or param_name in HIDDEN_SECURITY_PARAMS:
64                    continue
65                properties[param_name] = {"type": "string", "title": param_name}
66                if param.default is inspect.Parameter.empty:
67                    required.append(param_name)
68        except Exception:
69            pass
70
71    return {
72        "name": tool_name,
73        "description": description.strip(),
74        "parameters": {
75            "type": "object",
76            "properties": properties,
77            "required": required,
78        },
79        "schema_hiding_verified": True,
80    }

Extract tool schema concealing internal security parameters from model context.

Inspects tool function signatures or Pydantic args_schema and suppresses internal security parameter names (e.g. security_identity, config, runnable_config) from being exposed in model-facing JSON function schemas.

Args: tool: Target tool callable, BaseTool instance, or structured tool function.

Returns: Dict[str, Any]: Model-safe JSON Schema dictionary describing tool arguments.

class SecureDocument(langchain_core.documents.base.Document): View Source

53class SecureDocument(Document):
54    """Document container representation for chunk metadata evaluation.
55
56    Attributes:
57        page_content (str): Text content of document chunk.
58        metadata (Dict[str, Any]): Security envelope metadata dictionary.
59    """
60
61    model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)
62    score: Optional[float] = None
63
64    def __init__(self, page_content: str, metadata: Optional[Dict[str, Any]] = None, **kwargs: Any):
65        """Initialize SecureDocument.
66
67        Args:
68            page_content: Chunk page text content.
69            metadata: Metadata envelope mapping.
70            **kwargs: Additional keyword arguments.
71        """
72        super().__init__(page_content=page_content, metadata=dict(metadata or {}), **kwargs)
73
74    @classmethod
75    def create(cls, page_content: str, metadata: Dict[str, Any], **kwargs: Any):
76        """Factory method to construct SecureDocument.
77
78        Args:
79            page_content: Chunk text string.
80            metadata: Metadata mapping.
81            **kwargs: Additional keyword arguments.
82
83        Returns:
84            SecureDocument: Constructed document instance.
85        """
86        return cls(page_content=page_content, metadata=metadata, **kwargs)

Document container representation for chunk metadata evaluation.

Attributes: page_content (str): Text content of document chunk. metadata (Dict[str, Any]): Security envelope metadata dictionary.

score: Optional[float] = None

@classmethod

def create(cls, page_content: str, metadata: Dict[str, Any], **kwargs: Any): View Source

74    @classmethod
75    def create(cls, page_content: str, metadata: Dict[str, Any], **kwargs: Any):
76        """Factory method to construct SecureDocument.
77
78        Args:
79            page_content: Chunk text string.
80            metadata: Metadata mapping.
81            **kwargs: Additional keyword arguments.
82
83        Returns:
84            SecureDocument: Constructed document instance.
85        """
86        return cls(page_content=page_content, metadata=metadata, **kwargs)

Factory method to construct SecureDocument.

Args: page_content: Chunk text string. metadata: Metadata mapping. **kwargs: Additional keyword arguments.

Returns: SecureDocument: Constructed document instance.

class DocumentSecurityTransformer(langchain_core.documents.transformers.BaseDocumentTransformer): View Source

558class DocumentSecurityTransformer(BaseDocumentTransformer):
559    """Stage 1 Pre-Processor stamping baseline document-level security envelope."""
560
561    def __init__(self, *args: Any, **kwargs: Any):
562        """Initialize DocumentSecurityTransformer."""
563        default_classification = kwargs.get("default_classification", args[0] if len(args) > 0 else DEFAULT_CLASSIFICATION_PARAM)
564        default_roles = kwargs.get("default_roles", args[1] if len(args) > 1 else DEFAULT_ROLES_PARAM)
565        default_tenant = kwargs.get("default_tenant", args[2] if len(args) > 2 else DEFAULT_TENANT_PARAM)
566        default_department = kwargs.get("default_department", args[3] if len(args) > 3 else DEFAULT_DEPARTMENT_PARAM)
567        rules = kwargs.get("rules", args[4] if len(args) > 4 else DEFAULT_RULES_PARAM)
568
569        if isinstance(default_classification, int):
570            self._default_clearance = default_classification
571            self._default_classification = CLASSIFICATION_PUBLIC if default_classification == 0 else CLASSIFICATION_INTERNAL
572        else:
573            self._default_classification = str(default_classification)
574            self._default_clearance = CLASSIFICATION_TO_RANK.get(self._default_classification.upper(), 0)
575
576        if "default_clearance" in kwargs:
577            self._default_clearance = int(kwargs["default_clearance"])
578
579        self._default_roles = list(default_roles) if default_roles is not None else []
580        self._default_tenant = default_tenant
581        self._default_department = default_department
582        self._rules = list(rules) if rules is not None else []
583
584    @classmethod
585    def create(
586        cls,
587        *args: Any,
588        **kwargs: Any,
589    ) -> "DocumentSecurityTransformer":
590        """Factory method to construct DocumentSecurityTransformer.
591
592        Args:
593            *args: Optional positional arguments (default_classification, default_roles, default_tenant, default_department, rules).
594            **kwargs: Optional keyword arguments (default_classification, default_roles, default_tenant, default_department, rules).
595
596        Returns:
597            DocumentSecurityTransformer: Configured document transformer instance.
598        """
599        return cls(*args, **kwargs)
600
601    @classmethod
602    def create_default(cls) -> "DocumentSecurityTransformer":
603        """Factory method constructing default DocumentSecurityTransformer.
604
605        Returns:
606            DocumentSecurityTransformer: Default transformer instance.
607        """
608        return cls()
609
610    def transform_documents(self, documents: Sequence[Any], **kwargs: Any) -> Sequence[Any]:
611        """Apply security tags and classification rules to document metadata envelopes.
612
613        Args:
614            documents: Input document sequence.
615
616        Returns:
617            Sequence[Any]: Document sequence enriched with sec:* envelope metadata.
618        """
619        transformed: List[Any] = []
620        for doc in documents:
621            meta = dict(getattr(doc, "metadata", {}))
622            content = getattr(doc, "page_content", str(doc))
623
624            if meta.get("sec:unlabelled_raw") or meta.get("sec:unlabelled") or meta.get("sec:corrupt"):
625                transformed.append(Document(page_content=content, metadata=meta))
626                continue
627
628            classification = meta.get(SEC_CLASSIFICATION_LEVEL, self._default_classification)
629            clearance = meta.get(SEC_CLEARANCE_FLOOR, CLASSIFICATION_TO_RANK.get(str(classification).upper(), self._default_clearance))
630            roles = list(meta.get(SEC_ALLOWED_ROLES, self._default_roles))
631            tenant = meta.get(SEC_TENANT_ID, self._default_tenant)
632            department = meta.get(SEC_DEPARTMENT_ID, self._default_department)
633
634            for rule in self._rules:
635                res = rule(content, meta)
636                if "classification" in res:
637                    classification = res["classification"]
638                    clearance = CLASSIFICATION_TO_RANK.get(str(classification).upper(), clearance)
639                if "clearance" in res:
640                    clearance = res["clearance"]
641                if "roles" in res:
642                    roles = res["roles"]
643                if "tenant" in res:
644                    tenant = res["tenant"]
645                if "department" in res:
646                    department = res["department"]
647
648            doc_id = meta.get(SEC_DOCUMENT_ID, str(uuid.uuid4()))
649            meta[SEC_SCHEMA_VERSION] = DEFAULT_SCHEMA_VERSION
650            meta[SEC_DOCUMENT_ID] = doc_id
651            meta.setdefault(SEC_CHUNK_INDEX, 0)
652            meta[SEC_TENANT_ID] = tenant
653            meta[SEC_CLASSIFICATION_LEVEL] = classification
654            meta["sec:clearance_level"] = clearance
655            meta[SEC_CLEARANCE_FLOOR] = clearance
656            meta[SEC_ALLOWED_ROLES] = roles
657            meta[SEC_REQUIRED_ROLES] = roles  # Dual-compatibility
658            meta[SEC_DEPARTMENT_ID] = department
659            # Dual-serialization for vector stores lacking array index
660            meta[f"{SEC_ALLOWED_ROLES}_str"] = f",{','.join(roles)},"
661
662            if isinstance(doc, SecureDocument):
663                transformed.append(SecureDocument.create(page_content=content, metadata=meta))
664            else:
665                transformed.append(Document(page_content=content, metadata=meta))
666        return transformed
667
668    def transform_document(self, content: str, source_metadata: Dict[str, Any]) -> SecureDocument:
669        """Transform a single document content string and metadata dict.
670
671        Args:
672            content: Document text content.
673            source_metadata: Source metadata map.
674
675        Returns:
676            SecureDocument: Enriched SecureDocument object.
677        """
678        temp_doc = Document(page_content=content, metadata=source_metadata)
679        res = self.transform_documents([temp_doc])[0]
680        return SecureDocument.create(page_content=res.page_content, metadata=res.metadata)

Stage 1 Pre-Processor stamping baseline document-level security envelope.

DocumentSecurityTransformer(*args: Any, **kwargs: Any) View Source

561    def __init__(self, *args: Any, **kwargs: Any):
562        """Initialize DocumentSecurityTransformer."""
563        default_classification = kwargs.get("default_classification", args[0] if len(args) > 0 else DEFAULT_CLASSIFICATION_PARAM)
564        default_roles = kwargs.get("default_roles", args[1] if len(args) > 1 else DEFAULT_ROLES_PARAM)
565        default_tenant = kwargs.get("default_tenant", args[2] if len(args) > 2 else DEFAULT_TENANT_PARAM)
566        default_department = kwargs.get("default_department", args[3] if len(args) > 3 else DEFAULT_DEPARTMENT_PARAM)
567        rules = kwargs.get("rules", args[4] if len(args) > 4 else DEFAULT_RULES_PARAM)
568
569        if isinstance(default_classification, int):
570            self._default_clearance = default_classification
571            self._default_classification = CLASSIFICATION_PUBLIC if default_classification == 0 else CLASSIFICATION_INTERNAL
572        else:
573            self._default_classification = str(default_classification)
574            self._default_clearance = CLASSIFICATION_TO_RANK.get(self._default_classification.upper(), 0)
575
576        if "default_clearance" in kwargs:
577            self._default_clearance = int(kwargs["default_clearance"])
578
579        self._default_roles = list(default_roles) if default_roles is not None else []
580        self._default_tenant = default_tenant
581        self._default_department = default_department
582        self._rules = list(rules) if rules is not None else []

Initialize DocumentSecurityTransformer.

@classmethod

def create( cls, *args: Any, **kwargs: Any) -> DocumentSecurityTransformer: View Source

584    @classmethod
585    def create(
586        cls,
587        *args: Any,
588        **kwargs: Any,
589    ) -> "DocumentSecurityTransformer":
590        """Factory method to construct DocumentSecurityTransformer.
591
592        Args:
593            *args: Optional positional arguments (default_classification, default_roles, default_tenant, default_department, rules).
594            **kwargs: Optional keyword arguments (default_classification, default_roles, default_tenant, default_department, rules).
595
596        Returns:
597            DocumentSecurityTransformer: Configured document transformer instance.
598        """
599        return cls(*args, **kwargs)

Factory method to construct DocumentSecurityTransformer.

Args: *args: Optional positional arguments (default_classification, default_roles, default_tenant, default_department, rules). **kwargs: Optional keyword arguments (default_classification, default_roles, default_tenant, default_department, rules).

Returns: DocumentSecurityTransformer: Configured document transformer instance.

@classmethod

def create_default( cls) -> DocumentSecurityTransformer: View Source

601    @classmethod
602    def create_default(cls) -> "DocumentSecurityTransformer":
603        """Factory method constructing default DocumentSecurityTransformer.
604
605        Returns:
606            DocumentSecurityTransformer: Default transformer instance.
607        """
608        return cls()

Factory method constructing default DocumentSecurityTransformer.

Returns: DocumentSecurityTransformer: Default transformer instance.

def transform_documents(self, documents: Sequence[Any], **kwargs: Any) -> Sequence[Any]: View Source

610    def transform_documents(self, documents: Sequence[Any], **kwargs: Any) -> Sequence[Any]:
611        """Apply security tags and classification rules to document metadata envelopes.
612
613        Args:
614            documents: Input document sequence.
615
616        Returns:
617            Sequence[Any]: Document sequence enriched with sec:* envelope metadata.
618        """
619        transformed: List[Any] = []
620        for doc in documents:
621            meta = dict(getattr(doc, "metadata", {}))
622            content = getattr(doc, "page_content", str(doc))
623
624            if meta.get("sec:unlabelled_raw") or meta.get("sec:unlabelled") or meta.get("sec:corrupt"):
625                transformed.append(Document(page_content=content, metadata=meta))
626                continue
627
628            classification = meta.get(SEC_CLASSIFICATION_LEVEL, self._default_classification)
629            clearance = meta.get(SEC_CLEARANCE_FLOOR, CLASSIFICATION_TO_RANK.get(str(classification).upper(), self._default_clearance))
630            roles = list(meta.get(SEC_ALLOWED_ROLES, self._default_roles))
631            tenant = meta.get(SEC_TENANT_ID, self._default_tenant)
632            department = meta.get(SEC_DEPARTMENT_ID, self._default_department)
633
634            for rule in self._rules:
635                res = rule(content, meta)
636                if "classification" in res:
637                    classification = res["classification"]
638                    clearance = CLASSIFICATION_TO_RANK.get(str(classification).upper(), clearance)
639                if "clearance" in res:
640                    clearance = res["clearance"]
641                if "roles" in res:
642                    roles = res["roles"]
643                if "tenant" in res:
644                    tenant = res["tenant"]
645                if "department" in res:
646                    department = res["department"]
647
648            doc_id = meta.get(SEC_DOCUMENT_ID, str(uuid.uuid4()))
649            meta[SEC_SCHEMA_VERSION] = DEFAULT_SCHEMA_VERSION
650            meta[SEC_DOCUMENT_ID] = doc_id
651            meta.setdefault(SEC_CHUNK_INDEX, 0)
652            meta[SEC_TENANT_ID] = tenant
653            meta[SEC_CLASSIFICATION_LEVEL] = classification
654            meta["sec:clearance_level"] = clearance
655            meta[SEC_CLEARANCE_FLOOR] = clearance
656            meta[SEC_ALLOWED_ROLES] = roles
657            meta[SEC_REQUIRED_ROLES] = roles  # Dual-compatibility
658            meta[SEC_DEPARTMENT_ID] = department
659            # Dual-serialization for vector stores lacking array index
660            meta[f"{SEC_ALLOWED_ROLES}_str"] = f",{','.join(roles)},"
661
662            if isinstance(doc, SecureDocument):
663                transformed.append(SecureDocument.create(page_content=content, metadata=meta))
664            else:
665                transformed.append(Document(page_content=content, metadata=meta))
666        return transformed

Apply security tags and classification rules to document metadata envelopes.

Args: documents: Input document sequence.

Returns: Sequence[Any]: Document sequence enriched with sec:* envelope metadata.

def transform_document( self, content: str, source_metadata: Dict[str, Any]) -> SecureDocument: View Source

668    def transform_document(self, content: str, source_metadata: Dict[str, Any]) -> SecureDocument:
669        """Transform a single document content string and metadata dict.
670
671        Args:
672            content: Document text content.
673            source_metadata: Source metadata map.
674
675        Returns:
676            SecureDocument: Enriched SecureDocument object.
677        """
678        temp_doc = Document(page_content=content, metadata=source_metadata)
679        res = self.transform_documents([temp_doc])[0]
680        return SecureDocument.create(page_content=res.page_content, metadata=res.metadata)

Transform a single document content string and metadata dict.

Args: content: Document text content. source_metadata: Source metadata map.

Returns: SecureDocument: Enriched SecureDocument object.

class SecureDocumentChunkEnricher(langchain_core.documents.transformers.BaseDocumentTransformer): View Source

 865class SecureDocumentChunkEnricher(BaseDocumentTransformer):
 866    """Stage 3 Chunk Security Enricher scanning headers, XML tags, inline overrides, and executing SPI enrichers."""
 867
 868    def __init__(self, *args: Any, **kwargs: Any):
 869        """Initialize SecureDocumentChunkEnricher."""
 870        spi_enrichers = kwargs.get("spi_enrichers", args[0] if len(args) > 0 else DEFAULT_SPI_ENRICHERS_PARAM)
 871        secret_key = kwargs.get("secret_key", args[1] if len(args) > 1 else DEFAULT_SECRET_KEY_PARAM)
 872        audit_manager = kwargs.get("audit_manager", args[2] if len(args) > 2 else None)
 873        self._spi_enrichers = list(spi_enrichers) if spi_enrichers is not None else []
 874        self._secret_key = secret_key
 875        self._audit_manager: Optional[AuditManager] = audit_manager
 876        self._heading_enricher = HeadingClassificationEnricher.create()
 877
 878    @classmethod
 879    def create(
 880        cls,
 881        spi_enrichers: Optional[List[ChunkSecurityEnricherSPI]],
 882        secret_key: Optional[str],
 883        *args: Any,
 884        **kwargs: Any,
 885    ) -> "SecureDocumentChunkEnricher":
 886        """Factory method to construct SecureDocumentChunkEnricher.
 887
 888        Args:
 889            spi_enrichers: Optional list of ChunkSecurityEnricherSPI enrichers.
 890            secret_key: Secret key string for envelope digest signature.
 891            *args: Optional positional arguments.
 892            **kwargs: Optional keyword arguments.
 893
 894        Returns:
 895            SecureDocumentChunkEnricher: New enricher instance.
 896        """
 897        return cls(spi_enrichers, secret_key, *args, **kwargs)
 898
 899    @classmethod
 900    def create_default(cls) -> "SecureDocumentChunkEnricher":
 901        """Factory method creating default SecureDocumentChunkEnricher.
 902
 903        Returns:
 904            SecureDocumentChunkEnricher: Default enricher instance.
 905        """
 906        return cls()
 907
 908    def _apply_monotonic_updates(self, meta: Dict[str, Any], updates: Dict[str, Any], idx: int) -> None:
 909        if not updates:
 910            return
 911        parent_floor = int(meta.get(SEC_CLEARANCE_FLOOR, 0))
 912        de_escalation_detected = False
 913        elevation_detected = False
 914
 915        if SEC_CLEARANCE_FLOOR in updates:
 916            proposed_floor = int(updates[SEC_CLEARANCE_FLOOR])
 917            if proposed_floor < parent_floor:
 918                updates[SEC_CLEARANCE_FLOOR] = parent_floor
 919                de_escalation_detected = True
 920            elif proposed_floor > parent_floor:
 921                meta["sec:clearance_level"] = proposed_floor
 922                elevation_detected = True
 923
 924        if SEC_CLASSIFICATION_LEVEL in updates:
 925            new_rank = CLASSIFICATION_TO_RANK.get(str(updates[SEC_CLASSIFICATION_LEVEL]).upper(), 0)
 926            if new_rank < parent_floor:
 927                del updates[SEC_CLASSIFICATION_LEVEL]
 928                de_escalation_detected = True
 929            elif new_rank > parent_floor:
 930                updates[SEC_CLEARANCE_FLOOR] = new_rank
 931                meta["sec:clearance_level"] = new_rank
 932                elevation_detected = True
 933
 934        meta.update(updates)
 935        if de_escalation_detected:
 936            meta[SEC_MONOTONIC_ELEVATION_ENFORCED] = True
 937            if self._audit_manager:
 938                doc_id = str(meta.get(SEC_DOCUMENT_ID, "UNKNOWN"))
 939                tenant = str(meta.get(SEC_TENANT_ID, DEFAULT_TENANT_ID))
 940                self._audit_manager.record_event(AuditEvent.create(
 941                    subject_id="INGEST_PIPELINE",
 942                    tenant_id=tenant,
 943                    action="ingest:monotonic_elevation",
 944                    resource=f"{doc_id}:{idx}",
 945                    decision="ENFORCED",
 946                    diagnostic_reason=f"{EVENT_MONOTONIC_ELEVATION_ENFORCED} (CLASSIFICATION_DE_ESCALATION_REJECTED)",
 947                    severity=SEVERITY_WARNING,
 948                    metadata={"document_id": doc_id, "chunk_index": idx, "preserved_floor": parent_floor},
 949                    enforcement_point=PEP_INGESTION_PIPELINE,
 950                    reason_code=EVENT_MONOTONIC_ELEVATION_ENFORCED,
 951                ))
 952        elif elevation_detected and self._audit_manager:
 953            doc_id = str(meta.get(SEC_DOCUMENT_ID, "UNKNOWN"))
 954            tenant = str(meta.get(SEC_TENANT_ID, DEFAULT_TENANT_ID))
 955            elevated_cls = str(meta.get(SEC_CLASSIFICATION_LEVEL, CLASSIFICATION_CONFIDENTIAL))
 956            self._audit_manager.record_event(AuditEvent.create(
 957                subject_id="INGEST_PIPELINE",
 958                tenant_id=tenant,
 959                action="ingest:header_elevation",
 960                resource=f"{doc_id}:{idx}",
 961                decision="ELEVATED",
 962                diagnostic_reason=f"Header Elevation: {elevated_cls} sec:clearance_floor elevated",
 963                severity=SEVERITY_NOTICE,
 964                metadata={"document_id": doc_id, "chunk_index": idx, "elevated_classification": elevated_cls},
 965                enforcement_point=PEP_INGESTION_PIPELINE,
 966                reason_code=EVENT_INGEST_HEADER_ELEVATED,
 967            ))
 968
 969    def enrich(self, document: Any) -> Any:
 970        """Enrich a single document chunk.
 971
 972        Args:
 973            document: Document object or SecureDocument to enrich.
 974
 975        Returns:
 976            Any: Enriched document instance with monotonic clearance and metadata tags.
 977        """
 978        return self.transform_documents([document])[0]
 979
 980    def transform_documents(self, documents: Sequence[Any], **kwargs: Any) -> Sequence[Any]:
 981        """Enrich document chunks with monotonic clearance elevation and inline header/XML/SPI tags.
 982
 983        Args:
 984            documents: Input chunk sequence.
 985
 986        Returns:
 987            Sequence[Any]: Enriched chunk sequence.
 988        """
 989        enriched_docs: List[Any] = []
 990        for idx, doc in enumerate(documents):
 991            content = getattr(doc, "page_content", str(doc))
 992            meta = dict(getattr(doc, "metadata", {}))
 993
 994            meta[SEC_CHUNK_INDEX] = idx
 995
 996            # Built-in structural heading & XML tag enrichment with monotonic enforcement
 997            heading_ctx = ChunkEnrichmentContext.create(
 998                chunk_text=content,
 999                parent_document_metadata=dict(meta),
1000                current_security_envelope=dict(meta),
1001                chunk_index=idx,
1002            )
1003            heading_updates = self._heading_enricher.enrich(heading_ctx)
1004            self._apply_monotonic_updates(meta, heading_updates, idx)
1005
1006            # SPI Custom Enricher Strategy (with Monotonic Elevation Enforcement)
1007            for spi in self._spi_enrichers:
1008                if isinstance(spi, HeadingClassificationEnricher):
1009                    continue
1010                ctx = ChunkEnrichmentContext.create(
1011                    chunk_text=content,
1012                    parent_document_metadata=dict(meta),
1013                    current_security_envelope=dict(meta),
1014                    chunk_index=idx,
1015                )
1016                spi_updates = spi.enrich(ctx)
1017                self._apply_monotonic_updates(meta, spi_updates, idx)
1018
1019            # Compute optional SHA-256 HMAC digest signature
1020            if self._secret_key and not (meta.get("sec:unlabelled_raw") or meta.get("sec:unlabelled") or meta.get("sec:corrupt")):
1021                meta[SEC_ENVELOPE_HASH] = compute_envelope_hmac(meta, self._secret_key)
1022
1023            if isinstance(doc, SecureDocument):
1024                enriched_docs.append(SecureDocument.create(page_content=content, metadata=meta))
1025            else:
1026                enriched_docs.append(Document(page_content=content, metadata=meta))
1027
1028        return enriched_docs

Stage 3 Chunk Security Enricher scanning headers, XML tags, inline overrides, and executing SPI enrichers.

SecureDocumentChunkEnricher(*args: Any, **kwargs: Any) View Source

868    def __init__(self, *args: Any, **kwargs: Any):
869        """Initialize SecureDocumentChunkEnricher."""
870        spi_enrichers = kwargs.get("spi_enrichers", args[0] if len(args) > 0 else DEFAULT_SPI_ENRICHERS_PARAM)
871        secret_key = kwargs.get("secret_key", args[1] if len(args) > 1 else DEFAULT_SECRET_KEY_PARAM)
872        audit_manager = kwargs.get("audit_manager", args[2] if len(args) > 2 else None)
873        self._spi_enrichers = list(spi_enrichers) if spi_enrichers is not None else []
874        self._secret_key = secret_key
875        self._audit_manager: Optional[AuditManager] = audit_manager
876        self._heading_enricher = HeadingClassificationEnricher.create()

Initialize SecureDocumentChunkEnricher.

@classmethod

def create( cls, spi_enrichers: Optional[List[ChunkSecurityEnricherSPI]], secret_key: Optional[str], *args: Any, **kwargs: Any) -> SecureDocumentChunkEnricher: View Source

878    @classmethod
879    def create(
880        cls,
881        spi_enrichers: Optional[List[ChunkSecurityEnricherSPI]],
882        secret_key: Optional[str],
883        *args: Any,
884        **kwargs: Any,
885    ) -> "SecureDocumentChunkEnricher":
886        """Factory method to construct SecureDocumentChunkEnricher.
887
888        Args:
889            spi_enrichers: Optional list of ChunkSecurityEnricherSPI enrichers.
890            secret_key: Secret key string for envelope digest signature.
891            *args: Optional positional arguments.
892            **kwargs: Optional keyword arguments.
893
894        Returns:
895            SecureDocumentChunkEnricher: New enricher instance.
896        """
897        return cls(spi_enrichers, secret_key, *args, **kwargs)

Factory method to construct SecureDocumentChunkEnricher.

Args: spi_enrichers: Optional list of ChunkSecurityEnricherSPI enrichers. secret_key: Secret key string for envelope digest signature. *args: Optional positional arguments. **kwargs: Optional keyword arguments.

Returns: SecureDocumentChunkEnricher: New enricher instance.

@classmethod

def create_default( cls) -> SecureDocumentChunkEnricher: View Source

899    @classmethod
900    def create_default(cls) -> "SecureDocumentChunkEnricher":
901        """Factory method creating default SecureDocumentChunkEnricher.
902
903        Returns:
904            SecureDocumentChunkEnricher: Default enricher instance.
905        """
906        return cls()

Factory method creating default SecureDocumentChunkEnricher.

Returns: SecureDocumentChunkEnricher: Default enricher instance.

def enrich(self, document: Any) -> Any: View Source

969    def enrich(self, document: Any) -> Any:
970        """Enrich a single document chunk.
971
972        Args:
973            document: Document object or SecureDocument to enrich.
974
975        Returns:
976            Any: Enriched document instance with monotonic clearance and metadata tags.
977        """
978        return self.transform_documents([document])[0]

Enrich a single document chunk.

Args: document: Document object or SecureDocument to enrich.

Returns: Any: Enriched document instance with monotonic clearance and metadata tags.

def transform_documents(self, documents: Sequence[Any], **kwargs: Any) -> Sequence[Any]: View Source

 980    def transform_documents(self, documents: Sequence[Any], **kwargs: Any) -> Sequence[Any]:
 981        """Enrich document chunks with monotonic clearance elevation and inline header/XML/SPI tags.
 982
 983        Args:
 984            documents: Input chunk sequence.
 985
 986        Returns:
 987            Sequence[Any]: Enriched chunk sequence.
 988        """
 989        enriched_docs: List[Any] = []
 990        for idx, doc in enumerate(documents):
 991            content = getattr(doc, "page_content", str(doc))
 992            meta = dict(getattr(doc, "metadata", {}))
 993
 994            meta[SEC_CHUNK_INDEX] = idx
 995
 996            # Built-in structural heading & XML tag enrichment with monotonic enforcement
 997            heading_ctx = ChunkEnrichmentContext.create(
 998                chunk_text=content,
 999                parent_document_metadata=dict(meta),
1000                current_security_envelope=dict(meta),
1001                chunk_index=idx,
1002            )
1003            heading_updates = self._heading_enricher.enrich(heading_ctx)
1004            self._apply_monotonic_updates(meta, heading_updates, idx)
1005
1006            # SPI Custom Enricher Strategy (with Monotonic Elevation Enforcement)
1007            for spi in self._spi_enrichers:
1008                if isinstance(spi, HeadingClassificationEnricher):
1009                    continue
1010                ctx = ChunkEnrichmentContext.create(
1011                    chunk_text=content,
1012                    parent_document_metadata=dict(meta),
1013                    current_security_envelope=dict(meta),
1014                    chunk_index=idx,
1015                )
1016                spi_updates = spi.enrich(ctx)
1017                self._apply_monotonic_updates(meta, spi_updates, idx)
1018
1019            # Compute optional SHA-256 HMAC digest signature
1020            if self._secret_key and not (meta.get("sec:unlabelled_raw") or meta.get("sec:unlabelled") or meta.get("sec:corrupt")):
1021                meta[SEC_ENVELOPE_HASH] = compute_envelope_hmac(meta, self._secret_key)
1022
1023            if isinstance(doc, SecureDocument):
1024                enriched_docs.append(SecureDocument.create(page_content=content, metadata=meta))
1025            else:
1026                enriched_docs.append(Document(page_content=content, metadata=meta))
1027
1028        return enriched_docs

Enrich document chunks with monotonic clearance elevation and inline header/XML/SPI tags.

Args: documents: Input chunk sequence.

Returns: Sequence[Any]: Enriched chunk sequence.

SecureEnvelopeValidator = <class 'SecurityEnvelopeValidator'>

class SecurityEnvelopeValidator(langchain_core.documents.transformers.BaseDocumentTransformer): View Source

443class SecurityEnvelopeValidator(BaseDocumentTransformer):
444    """Stage 4 Fail-Closed Envelope Validator verifying presence and integrity of sec:* envelope metadata."""
445
446    def __init__(self, *args: Any, **kwargs: Any):
447        """Initialize SecurityEnvelopeValidator."""
448        secret_key = kwargs.get("secret_key", args[0] if len(args) > 0 else DEFAULT_SECRET_KEY_PARAM)
449        self._secret_key = secret_key
450
451    @classmethod
452    def create(cls, secret_key: Optional[str]) -> "SecurityEnvelopeValidator":
453        """Factory method creating SecurityEnvelopeValidator.
454
455        Args:
456            secret_key: Secret key string for HMAC validation.
457
458        Returns:
459            SecurityEnvelopeValidator: New validator instance.
460        """
461        return cls(secret_key=secret_key)
462
463    @classmethod
464    def create_default(cls) -> "SecurityEnvelopeValidator":
465        """Factory method creating default SecurityEnvelopeValidator.
466
467        Returns:
468            SecurityEnvelopeValidator: Default validator instance.
469        """
470        return cls(secret_key=DEFAULT_SECRET_KEY_PARAM)
471
472    @property
473    def secret_key(self) -> Optional[str]:
474        """Get configured HMAC secret key."""
475        return self._secret_key
476
477    @classmethod
478    def has_mandatory_envelope(cls, metadata: Dict[str, Any]) -> bool:
479        """Validate if document metadata contains mandatory sec:* envelope keys according to Spec 1.
480
481        Mandatory keys:
482        - sec:tenant_id
483        - sec:classification_level (or sec:clearance_level)
484        - sec:allowed_roles (or sec:required_roles)
485        """
486        if not isinstance(metadata, dict):
487            return False
488        has_tenant = bool(metadata.get(SEC_TENANT_ID))
489        has_class = bool(metadata.get(SEC_CLASSIFICATION_LEVEL) or metadata.get("sec:clearance_level") is not None)
490        roles = metadata.get(SEC_ALLOWED_ROLES) or metadata.get(SEC_REQUIRED_ROLES)
491        has_roles = bool(roles and len(roles) > 0)
492        return bool(has_tenant and has_class and has_roles)
493
494    @classmethod
495    def validate(cls, document: Any) -> bool:
496        """Validate if document metadata contains mandatory sec:* envelope keys.
497
498        Args:
499            document: Target document or metadata dict.
500
501        Returns:
502            bool: True if required security metadata exists.
503        """
504        meta = getattr(document, "metadata", {}) if not isinstance(document, dict) else document
505        return cls.has_mandatory_envelope(meta)
506
507    def validate_metadata(self, metadata: Dict[str, Any]) -> bool:
508        """Validate metadata dictionary key presence and HMAC signature integrity.
509
510        Args:
511            metadata: Document metadata dictionary.
512
513        Returns:
514            bool: True if valid, False if incomplete or signature mismatched.
515        """
516        if not isinstance(metadata, dict):
517            return False
518        if metadata.get("sec:corrupt") or metadata.get("sec:unlabelled") or metadata.get("sec:unlabelled_raw"):
519            return False
520        if not self.has_mandatory_envelope(metadata):
521            return False
522
523        if self._secret_key and SEC_ENVELOPE_HASH in metadata:
524            expected_hash = compute_envelope_hmac(metadata, self._secret_key)
525            if metadata[SEC_ENVELOPE_HASH] != expected_hash:
526                logger.warning("Security envelope HMAC hash mismatch detected (possible tampering).")
527                return False
528
529        return True
530
531    def transform_documents(self, documents: Sequence[Any], **kwargs: Any) -> Sequence[Any]:
532        """Transform document collection, raising ValueError if any chunk fails validation.
533
534        Args:
535            documents: Sequence of document objects.
536
537        Returns:
538            Sequence[Any]: Validated document sequence.
539
540        Raises:
541            ValueError: If envelope validation fails for any document.
542        """
543        validated: List[Any] = []
544        for doc in documents:
545            meta = getattr(doc, "metadata", {})
546            if not self.validate_metadata(meta):
547                raise ValueError("Security envelope validation failed for document chunk.")
548            validated.append(doc)
549        return validated

Stage 4 Fail-Closed Envelope Validator verifying presence and integrity of sec:* envelope metadata.

SecurityEnvelopeValidator(*args: Any, **kwargs: Any) View Source

446    def __init__(self, *args: Any, **kwargs: Any):
447        """Initialize SecurityEnvelopeValidator."""
448        secret_key = kwargs.get("secret_key", args[0] if len(args) > 0 else DEFAULT_SECRET_KEY_PARAM)
449        self._secret_key = secret_key

Initialize SecurityEnvelopeValidator.

@classmethod

def create( cls, secret_key: Optional[str]) -> SecurityEnvelopeValidator: View Source

451    @classmethod
452    def create(cls, secret_key: Optional[str]) -> "SecurityEnvelopeValidator":
453        """Factory method creating SecurityEnvelopeValidator.
454
455        Args:
456            secret_key: Secret key string for HMAC validation.
457
458        Returns:
459            SecurityEnvelopeValidator: New validator instance.
460        """
461        return cls(secret_key=secret_key)

Factory method creating SecurityEnvelopeValidator.

Args: secret_key: Secret key string for HMAC validation.

Returns: SecurityEnvelopeValidator: New validator instance.

@classmethod

def create_default( cls) -> SecurityEnvelopeValidator: View Source

463    @classmethod
464    def create_default(cls) -> "SecurityEnvelopeValidator":
465        """Factory method creating default SecurityEnvelopeValidator.
466
467        Returns:
468            SecurityEnvelopeValidator: Default validator instance.
469        """
470        return cls(secret_key=DEFAULT_SECRET_KEY_PARAM)

Factory method creating default SecurityEnvelopeValidator.

Returns: SecurityEnvelopeValidator: Default validator instance.

secret_key: Optional[str] View Source

472    @property
473    def secret_key(self) -> Optional[str]:
474        """Get configured HMAC secret key."""
475        return self._secret_key

Get configured HMAC secret key.

@classmethod

def has_mandatory_envelope(cls, metadata: Dict[str, Any]) -> bool: View Source

477    @classmethod
478    def has_mandatory_envelope(cls, metadata: Dict[str, Any]) -> bool:
479        """Validate if document metadata contains mandatory sec:* envelope keys according to Spec 1.
480
481        Mandatory keys:
482        - sec:tenant_id
483        - sec:classification_level (or sec:clearance_level)
484        - sec:allowed_roles (or sec:required_roles)
485        """
486        if not isinstance(metadata, dict):
487            return False
488        has_tenant = bool(metadata.get(SEC_TENANT_ID))
489        has_class = bool(metadata.get(SEC_CLASSIFICATION_LEVEL) or metadata.get("sec:clearance_level") is not None)
490        roles = metadata.get(SEC_ALLOWED_ROLES) or metadata.get(SEC_REQUIRED_ROLES)
491        has_roles = bool(roles and len(roles) > 0)
492        return bool(has_tenant and has_class and has_roles)

Validate if document metadata contains mandatory sec:* envelope keys according to Spec 1.

Mandatory keys:

sec:tenant_id
sec:classification_level (or sec:clearance_level)
sec:allowed_roles (or sec:required_roles)

@classmethod

def validate(cls, document: Any) -> bool: View Source

494    @classmethod
495    def validate(cls, document: Any) -> bool:
496        """Validate if document metadata contains mandatory sec:* envelope keys.
497
498        Args:
499            document: Target document or metadata dict.
500
501        Returns:
502            bool: True if required security metadata exists.
503        """
504        meta = getattr(document, "metadata", {}) if not isinstance(document, dict) else document
505        return cls.has_mandatory_envelope(meta)

Validate if document metadata contains mandatory sec:* envelope keys.

Args: document: Target document or metadata dict.

Returns: bool: True if required security metadata exists.

def validate_metadata(self, metadata: Dict[str, Any]) -> bool: View Source

507    def validate_metadata(self, metadata: Dict[str, Any]) -> bool:
508        """Validate metadata dictionary key presence and HMAC signature integrity.
509
510        Args:
511            metadata: Document metadata dictionary.
512
513        Returns:
514            bool: True if valid, False if incomplete or signature mismatched.
515        """
516        if not isinstance(metadata, dict):
517            return False
518        if metadata.get("sec:corrupt") or metadata.get("sec:unlabelled") or metadata.get("sec:unlabelled_raw"):
519            return False
520        if not self.has_mandatory_envelope(metadata):
521            return False
522
523        if self._secret_key and SEC_ENVELOPE_HASH in metadata:
524            expected_hash = compute_envelope_hmac(metadata, self._secret_key)
525            if metadata[SEC_ENVELOPE_HASH] != expected_hash:
526                logger.warning("Security envelope HMAC hash mismatch detected (possible tampering).")
527                return False
528
529        return True

Validate metadata dictionary key presence and HMAC signature integrity.

Args: metadata: Document metadata dictionary.

Returns: bool: True if valid, False if incomplete or signature mismatched.

def transform_documents(self, documents: Sequence[Any], **kwargs: Any) -> Sequence[Any]: View Source

531    def transform_documents(self, documents: Sequence[Any], **kwargs: Any) -> Sequence[Any]:
532        """Transform document collection, raising ValueError if any chunk fails validation.
533
534        Args:
535            documents: Sequence of document objects.
536
537        Returns:
538            Sequence[Any]: Validated document sequence.
539
540        Raises:
541            ValueError: If envelope validation fails for any document.
542        """
543        validated: List[Any] = []
544        for doc in documents:
545            meta = getattr(doc, "metadata", {})
546            if not self.validate_metadata(meta):
547                raise ValueError("Security envelope validation failed for document chunk.")
548            validated.append(doc)
549        return validated

Transform document collection, raising ValueError if any chunk fails validation.

Args: documents: Sequence of document objects.

Returns: Sequence[Any]: Validated document sequence.

Raises: ValueError: If envelope validation fails for any document.

class SecureVectorStoreIngestor: View Source

1031class SecureVectorStoreIngestor:
1032    """Stage 5 Ingestion orchestrator chunking raw content, enriching with sec:* metadata, and validating modes."""
1033
1034    def __init__(self, *args: Any, **kwargs: Any):
1035        """Initialize SecureVectorStoreIngestor.
1036
1037        Args:
1038            *args: Positional arguments (transformer, enricher, validator, ingestion_mode, text_splitter, audit_manager).
1039            **kwargs: Keyword arguments (transformer, enricher, validator, ingestion_mode, text_splitter, audit_manager).
1040        """
1041        transformer = kwargs.get("transformer", args[0] if len(args) > 0 else None)
1042        enricher = kwargs.get("enricher", args[1] if len(args) > 1 else None)
1043        validator = kwargs.get("validator", args[2] if len(args) > 2 else None)
1044        ingestion_mode = kwargs.get("ingestion_mode", args[3] if len(args) > 3 else DEFAULT_INGESTION_MODE_PARAM)
1045        text_splitter = kwargs.get("text_splitter", args[4] if len(args) > 4 else None)
1046        audit_manager = kwargs.get("audit_manager", args[5] if len(args) > 5 else None)
1047
1048        self._transformer = transformer if transformer is not None else DocumentSecurityTransformer.create_default()
1049        self._enricher = enricher if enricher is not None else SecureDocumentChunkEnricher.create_default()
1050        self._validator = validator if validator is not None else SecurityEnvelopeValidator.create_default()
1051        self._ingestion_mode = ingestion_mode
1052        self._text_splitter = text_splitter
1053        self._audit_manager = audit_manager
1054
1055    @classmethod
1056    def create(
1057        cls,
1058        *args: Any,
1059        **kwargs: Any,
1060    ) -> "SecureVectorStoreIngestor":
1061        """Factory method constructing SecureVectorStoreIngestor.
1062
1063        Args:
1064            *args: Positional arguments (transformer, enricher, validator, ingestion_mode, text_splitter, audit_manager).
1065            **kwargs: Keyword arguments (transformer, enricher, validator, ingestion_mode, text_splitter, audit_manager).
1066
1067        Returns:
1068            SecureVectorStoreIngestor: Newly constructed ingestor orchestrator.
1069        """
1070        return cls(*args, **kwargs)
1071
1072    @classmethod
1073    def create_default(cls) -> "SecureVectorStoreIngestor":
1074        """Factory method constructing default SecureVectorStoreIngestor.
1075
1076        Returns:
1077            SecureVectorStoreIngestor: Default ingestor instance.
1078        """
1079        return cls()
1080
1081    @property
1082    def ingestion_mode(self) -> str:
1083        """Get configured ingestion mode."""
1084        return self._ingestion_mode
1085
1086    def _validate_and_finalize_batch(self, enriched_docs: Sequence[Any], default_doc_id: str) -> List[SecureDocument]:
1087        final_docs: List[SecureDocument] = []
1088        for idx, doc in enumerate(enriched_docs):
1089            meta = dict(getattr(doc, "metadata", {}))
1090            doc_id = str(meta.get(SEC_DOCUMENT_ID, default_doc_id))
1091            meta[SEC_DOCUMENT_ID] = doc_id
1092            if meta.get(SEC_CHUNK_INDEX) is None:
1093                meta[SEC_CHUNK_INDEX] = idx
1094            if not meta.get(SEC_SCHEMA_VERSION):
1095                meta[SEC_SCHEMA_VERSION] = DEFAULT_SCHEMA_VERSION
1096
1097            if not self._validator.validate_metadata(meta):
1098                default_tenant = getattr(self._transformer, "_default_tenant", DEFAULT_TENANT_ID)
1099                tenant = str(meta.get(SEC_TENANT_ID, default_tenant))
1100                if self._ingestion_mode == INGEST_MODE_STRICT_FAIL_CLOSED:
1101                    if self._audit_manager:
1102                        self._audit_manager.record_event(AuditEvent.create(
1103                            subject_id="INGEST_PIPELINE",
1104                            tenant_id=tenant,
1105                            action="ingest:chunk",
1106                            resource=f"{doc_id}:{idx}",
1107                            decision="REJECTED",
1108                            diagnostic_reason=(
1109                                "INGEST_CHUNK_REJECTED | ATOMIC_ROLLBACK | "
1110                                "UNLABELLED_CHUNK_QUARANTINED | MISSING_SECURITY_ENVELOPE: Strict fail-closed validation failure"
1111                            ),
1112                            severity=SEVERITY_WARNING,
1113                            metadata={"document_id": doc_id, "chunk_index": idx},
1114                            enforcement_point=PEP_INGESTION_PIPELINE,
1115                            reason_code=EVENT_INGEST_CHUNK_REJECTED,
1116                        ))
1117                    raise IngestionRejectedException.create_rejection(
1118                        document_id=doc_id,
1119                        chunk_index=idx,
1120                        message=(
1121                            f"INGEST_CHUNK_REJECTED (ATOMIC_ROLLBACK / UNLABELLED_CHUNK_QUARANTINED / MISSING_SECURITY_ENVELOPE): "
1122                            f"Ingestion rejected for document {doc_id} chunk {idx} under Strict Fail-Closed mode."
1123                        ),
1124                    )
1125                elif self._ingestion_mode == INGEST_MODE_EVERYTHING_CLASSIFIED:
1126                    meta[SEC_TENANT_ID] = tenant
1127                    meta[SEC_CLASSIFICATION_LEVEL] = CLASSIFICATION_RESTRICTED
1128                    meta["sec:clearance_level"] = CLEARANCE_RANK_RESTRICTED
1129                    meta[SEC_CLEARANCE_FLOOR] = CLEARANCE_RANK_RESTRICTED
1130                    meta[SEC_QUARANTINED] = True
1131                    if not meta.get(SEC_ALLOWED_ROLES):
1132                        meta[SEC_ALLOWED_ROLES] = ["ROLE_ADMIN"]
1133                        meta[SEC_REQUIRED_ROLES] = ["ROLE_ADMIN"]
1134                    if getattr(self._validator, "secret_key", None):
1135                        meta[SEC_ENVELOPE_HASH] = compute_envelope_hmac(meta, self._validator.secret_key)
1136                    if self._audit_manager:
1137                        self._audit_manager.record_event(AuditEvent.create(
1138                            subject_id="INGEST_PIPELINE",
1139                            tenant_id=tenant,
1140                            action="ingest:chunk",
1141                            resource=f"{doc_id}:{idx}",
1142                            decision="QUARANTINED",
1143                            diagnostic_reason="INGEST_CHUNK_QUARANTINED (sec:quarantined=true): Elevated to RESTRICTED under Everything Classified mode",
1144                            severity=SEVERITY_NOTICE,
1145                            metadata={"document_id": doc_id, "chunk_index": idx, SEC_QUARANTINED: True},
1146                            enforcement_point=PEP_INGESTION_PIPELINE,
1147                            reason_code=EVENT_INGEST_CHUNK_QUARANTINED,
1148                        ))
1149                elif self._ingestion_mode == INGEST_MODE_EVERYTHING_PUBLIC:
1150                    meta[SEC_TENANT_ID] = tenant
1151                    meta[SEC_CLASSIFICATION_LEVEL] = CLASSIFICATION_PUBLIC
1152                    meta["sec:clearance_level"] = CLEARANCE_RANK_PUBLIC
1153                    meta[SEC_CLEARANCE_FLOOR] = CLEARANCE_RANK_PUBLIC
1154                    meta[SEC_ALLOWED_ROLES] = [CLASSIFICATION_PUBLIC]
1155                    meta[SEC_REQUIRED_ROLES] = [CLASSIFICATION_PUBLIC]
1156                    if getattr(self._validator, "secret_key", None):
1157                        meta[SEC_ENVELOPE_HASH] = compute_envelope_hmac(meta, self._validator.secret_key)
1158                    if self._audit_manager:
1159                        self._audit_manager.record_event(AuditEvent.create(
1160                            subject_id="INGEST_PIPELINE",
1161                            tenant_id=tenant,
1162                            action="ingest:chunk",
1163                            resource=f"{doc_id}:{idx}",
1164                            decision="DEFAULT_PUBLIC",
1165                            diagnostic_reason="Defaulted to PUBLIC under Everything Public mode",
1166                            severity=SEVERITY_NOTICE,
1167                            metadata={"document_id": doc_id, "chunk_index": idx},
1168                            enforcement_point=PEP_INGESTION_PIPELINE,
1169                            reason_code=EVENT_INGEST_CHUNK_DEFAULT_PUBLIC,
1170                        ))
1171
1172            final_docs.append(SecureDocument.create(page_content=doc.page_content, metadata=meta))
1173        return final_docs
1174
1175    def _split_documents(self, documents: Sequence[Any]) -> Sequence[Any]:
1176        """Stage 2 TextSplitter execution propagating metadata from parent documents to chunks."""
1177        if self._text_splitter is None:
1178            return documents
1179        if hasattr(self._text_splitter, "transform_documents"):
1180            return self._text_splitter.transform_documents(documents)
1181        if hasattr(self._text_splitter, "split_documents"):
1182            return self._text_splitter.split_documents(documents)
1183        if hasattr(self._text_splitter, "split_text"):
1184            split_docs: List[Any] = []
1185            for doc in documents:
1186                content = getattr(doc, "page_content", str(doc))
1187                meta = dict(getattr(doc, "metadata", {}))
1188                chunks = self._text_splitter.split_text(content)
1189                if not chunks and content:
1190                    chunks = [content]
1191                elif not chunks and content == "":
1192                    chunks = [""]
1193                for chunk_idx, chunk in enumerate(chunks):
1194                    chunk_meta = dict(meta)
1195                    chunk_meta[SEC_CHUNK_INDEX] = chunk_idx
1196                    if isinstance(doc, SecureDocument):
1197                        split_docs.append(SecureDocument.create(page_content=chunk, metadata=chunk_meta))
1198                    else:
1199                        split_docs.append(Document(page_content=chunk, metadata=chunk_meta))
1200            return split_docs
1201        if callable(self._text_splitter):
1202            split_docs = []
1203            for doc in documents:
1204                content = getattr(doc, "page_content", str(doc))
1205                meta = dict(getattr(doc, "metadata", {}))
1206                chunks = self._text_splitter(content)
1207                if not chunks and content:
1208                    chunks = [content]
1209                elif not chunks and content == "":
1210                    chunks = [""]
1211                for chunk_idx, chunk in enumerate(chunks):
1212                    chunk_meta = dict(meta)
1213                    chunk_meta[SEC_CHUNK_INDEX] = chunk_idx
1214                    if isinstance(doc, SecureDocument):
1215                        split_docs.append(SecureDocument.create(page_content=chunk, metadata=chunk_meta))
1216                    else:
1217                        split_docs.append(Document(page_content=chunk, metadata=chunk_meta))
1218            return split_docs
1219        return documents
1220
1221    def ingest_documents(self, documents: Sequence[Any]) -> List[SecureDocument]:
1222        """Atomically ingest a batch of documents through the 5-stage security pipeline.
1223
1224        Document → Stage 1(Transformer) → Stage 2(TextSplitter) → Stage 3(Enricher) → Stage 4(Validator) → Stage 5(VectorStore)
1225
1226        Args:
1227            documents: Sequence of Document objects or dicts with 'page_content'/'content' and 'metadata'.
1228
1229        Returns:
1230            List[SecureDocument]: Validated SecureDocument chunks.
1231
1232        Raises:
1233            IngestionRejectedException: On any chunk validation or enrichment failure in STRICT_FAIL_CLOSED mode.
1234        """
1235        raw_docs: List[Any] = []
1236        for item in documents:
1237            if isinstance(item, dict):
1238                content = str(item.get("page_content") or item.get("content") or "")
1239                meta = dict(item.get("metadata") or {})
1240                raw_docs.append(Document(page_content=content, metadata=meta))
1241            else:
1242                raw_docs.append(item)
1243
1244        try:
1245            # Stage 1: DocumentSecurityTransformer
1246            stamped_docs = self._transformer.transform_documents(raw_docs)
1247            # Stage 2: Standard TextSplitter with metadata propagation
1248            split_docs = self._split_documents(stamped_docs)
1249            # Stage 3: SecureDocumentChunkEnricher
1250            enriched_docs = self._enricher.transform_documents(split_docs)
1251        except Exception as err:
1252            if self._ingestion_mode == INGEST_MODE_STRICT_FAIL_CLOSED:
1253                if self._audit_manager:
1254                    self._audit_manager.record_event(AuditEvent.create(
1255                        subject_id="INGEST_PIPELINE",
1256                        tenant_id=DEFAULT_TENANT_ID,
1257                        action="ingest:batch",
1258                        resource="batch",
1259                        decision="REJECTED",
1260                        diagnostic_reason=f"INGEST_CHUNK_REJECTED | ATOMIC_ROLLBACK: {str(err)}",
1261                        severity=SEVERITY_WARNING,
1262                        metadata={"error": str(err)},
1263                        enforcement_point=PEP_INGESTION_PIPELINE,
1264                        reason_code=EVENT_INGEST_CHUNK_REJECTED,
1265                    ))
1266                raise IngestionRejectedException.create_rejection(
1267                    document_id="batch",
1268                    chunk_index=0,
1269                    message=f"INGEST_CHUNK_REJECTED (ATOMIC_ROLLBACK): {str(err)}",
1270                ) from err
1271            raise
1272
1273        # Stage 4: SecurityEnvelopeValidator and Stage 5 return
1274        return self._validate_and_finalize_batch(enriched_docs, str(uuid.uuid4()))
1275
1276    def ingest_text(
1277        self,
1278        text: str,
1279        source_name: str,
1280        *args: Any,
1281        **kwargs: Any,
1282    ) -> List[SecureDocument]:
1283        """Ingest raw text string into secure document chunks with full envelope validation.
1284
1285        Args:
1286            text: Raw text content string.
1287            source_name: Source document name or identifier.
1288            *args: Optional positional arguments (chunk_delimiter, metadata).
1289            **kwargs: Optional keyword arguments (chunk_delimiter, metadata).
1290
1291        Returns:
1292            List[SecureDocument]: Ingested and validated SecureDocument objects.
1293
1294        Raises:
1295            IngestionRejectedException: If strict validation fails under STRICT_FAIL_CLOSED mode.
1296        """
1297        chunk_delimiter = str(kwargs.get("chunk_delimiter", args[0] if len(args) > 0 else DEFAULT_CHUNK_DELIMITER))
1298        metadata = kwargs.get("metadata", args[1] if len(args) > 1 else None)
1299
1300        base_meta = dict(metadata) if metadata is not None else {}
1301        base_meta["source"] = source_name
1302        doc_id = base_meta.get(SEC_DOCUMENT_ID, str(uuid.uuid4()))
1303        base_meta[SEC_DOCUMENT_ID] = doc_id
1304
1305        if self._text_splitter is not None:
1306            raw_docs = [Document(page_content=text, metadata=dict(base_meta))]
1307            return self.ingest_documents(raw_docs)
1308
1309        raw_chunks = [chunk.strip() for chunk in text.split(chunk_delimiter) if chunk.strip()]
1310        if not raw_chunks and text:
1311            raw_chunks = [text]
1312        raw_docs = [Document(page_content=chunk, metadata=dict(base_meta)) for chunk in raw_chunks]
1313        return self.ingest_documents(raw_docs)

Stage 5 Ingestion orchestrator chunking raw content, enriching with sec:* metadata, and validating modes.

SecureVectorStoreIngestor(*args: Any, **kwargs: Any) View Source

1034    def __init__(self, *args: Any, **kwargs: Any):
1035        """Initialize SecureVectorStoreIngestor.
1036
1037        Args:
1038            *args: Positional arguments (transformer, enricher, validator, ingestion_mode, text_splitter, audit_manager).
1039            **kwargs: Keyword arguments (transformer, enricher, validator, ingestion_mode, text_splitter, audit_manager).
1040        """
1041        transformer = kwargs.get("transformer", args[0] if len(args) > 0 else None)
1042        enricher = kwargs.get("enricher", args[1] if len(args) > 1 else None)
1043        validator = kwargs.get("validator", args[2] if len(args) > 2 else None)
1044        ingestion_mode = kwargs.get("ingestion_mode", args[3] if len(args) > 3 else DEFAULT_INGESTION_MODE_PARAM)
1045        text_splitter = kwargs.get("text_splitter", args[4] if len(args) > 4 else None)
1046        audit_manager = kwargs.get("audit_manager", args[5] if len(args) > 5 else None)
1047
1048        self._transformer = transformer if transformer is not None else DocumentSecurityTransformer.create_default()
1049        self._enricher = enricher if enricher is not None else SecureDocumentChunkEnricher.create_default()
1050        self._validator = validator if validator is not None else SecurityEnvelopeValidator.create_default()
1051        self._ingestion_mode = ingestion_mode
1052        self._text_splitter = text_splitter
1053        self._audit_manager = audit_manager

Initialize SecureVectorStoreIngestor.

Args: *args: Positional arguments (transformer, enricher, validator, ingestion_mode, text_splitter, audit_manager). **kwargs: Keyword arguments (transformer, enricher, validator, ingestion_mode, text_splitter, audit_manager).

@classmethod

def create( cls, *args: Any, **kwargs: Any) -> SecureVectorStoreIngestor: View Source

1055    @classmethod
1056    def create(
1057        cls,
1058        *args: Any,
1059        **kwargs: Any,
1060    ) -> "SecureVectorStoreIngestor":
1061        """Factory method constructing SecureVectorStoreIngestor.
1062
1063        Args:
1064            *args: Positional arguments (transformer, enricher, validator, ingestion_mode, text_splitter, audit_manager).
1065            **kwargs: Keyword arguments (transformer, enricher, validator, ingestion_mode, text_splitter, audit_manager).
1066
1067        Returns:
1068            SecureVectorStoreIngestor: Newly constructed ingestor orchestrator.
1069        """
1070        return cls(*args, **kwargs)

Factory method constructing SecureVectorStoreIngestor.

Returns: SecureVectorStoreIngestor: Newly constructed ingestor orchestrator.

@classmethod

def create_default( cls) -> SecureVectorStoreIngestor: View Source

1072    @classmethod
1073    def create_default(cls) -> "SecureVectorStoreIngestor":
1074        """Factory method constructing default SecureVectorStoreIngestor.
1075
1076        Returns:
1077            SecureVectorStoreIngestor: Default ingestor instance.
1078        """
1079        return cls()

Factory method constructing default SecureVectorStoreIngestor.

Returns: SecureVectorStoreIngestor: Default ingestor instance.

ingestion_mode: str View Source

1081    @property
1082    def ingestion_mode(self) -> str:
1083        """Get configured ingestion mode."""
1084        return self._ingestion_mode

Get configured ingestion mode.

def ingest_documents( self, documents: Sequence[Any]) -> List[SecureDocument]: View Source

1221    def ingest_documents(self, documents: Sequence[Any]) -> List[SecureDocument]:
1222        """Atomically ingest a batch of documents through the 5-stage security pipeline.
1223
1224        Document → Stage 1(Transformer) → Stage 2(TextSplitter) → Stage 3(Enricher) → Stage 4(Validator) → Stage 5(VectorStore)
1225
1226        Args:
1227            documents: Sequence of Document objects or dicts with 'page_content'/'content' and 'metadata'.
1228
1229        Returns:
1230            List[SecureDocument]: Validated SecureDocument chunks.
1231
1232        Raises:
1233            IngestionRejectedException: On any chunk validation or enrichment failure in STRICT_FAIL_CLOSED mode.
1234        """
1235        raw_docs: List[Any] = []
1236        for item in documents:
1237            if isinstance(item, dict):
1238                content = str(item.get("page_content") or item.get("content") or "")
1239                meta = dict(item.get("metadata") or {})
1240                raw_docs.append(Document(page_content=content, metadata=meta))
1241            else:
1242                raw_docs.append(item)
1243
1244        try:
1245            # Stage 1: DocumentSecurityTransformer
1246            stamped_docs = self._transformer.transform_documents(raw_docs)
1247            # Stage 2: Standard TextSplitter with metadata propagation
1248            split_docs = self._split_documents(stamped_docs)
1249            # Stage 3: SecureDocumentChunkEnricher
1250            enriched_docs = self._enricher.transform_documents(split_docs)
1251        except Exception as err:
1252            if self._ingestion_mode == INGEST_MODE_STRICT_FAIL_CLOSED:
1253                if self._audit_manager:
1254                    self._audit_manager.record_event(AuditEvent.create(
1255                        subject_id="INGEST_PIPELINE",
1256                        tenant_id=DEFAULT_TENANT_ID,
1257                        action="ingest:batch",
1258                        resource="batch",
1259                        decision="REJECTED",
1260                        diagnostic_reason=f"INGEST_CHUNK_REJECTED | ATOMIC_ROLLBACK: {str(err)}",
1261                        severity=SEVERITY_WARNING,
1262                        metadata={"error": str(err)},
1263                        enforcement_point=PEP_INGESTION_PIPELINE,
1264                        reason_code=EVENT_INGEST_CHUNK_REJECTED,
1265                    ))
1266                raise IngestionRejectedException.create_rejection(
1267                    document_id="batch",
1268                    chunk_index=0,
1269                    message=f"INGEST_CHUNK_REJECTED (ATOMIC_ROLLBACK): {str(err)}",
1270                ) from err
1271            raise
1272
1273        # Stage 4: SecurityEnvelopeValidator and Stage 5 return
1274        return self._validate_and_finalize_batch(enriched_docs, str(uuid.uuid4()))

Atomically ingest a batch of documents through the 5-stage security pipeline.

Document → Stage 1(Transformer) → Stage 2(TextSplitter) → Stage 3(Enricher) → Stage 4(Validator) → Stage 5(VectorStore)

Args: documents: Sequence of Document objects or dicts with 'page_content'/'content' and 'metadata'.

Returns: List[SecureDocument]: Validated SecureDocument chunks.

Raises: IngestionRejectedException: On any chunk validation or enrichment failure in STRICT_FAIL_CLOSED mode.

def ingest_text( self, text: str, source_name: str, *args: Any, **kwargs: Any) -> List[SecureDocument]: View Source

1276    def ingest_text(
1277        self,
1278        text: str,
1279        source_name: str,
1280        *args: Any,
1281        **kwargs: Any,
1282    ) -> List[SecureDocument]:
1283        """Ingest raw text string into secure document chunks with full envelope validation.
1284
1285        Args:
1286            text: Raw text content string.
1287            source_name: Source document name or identifier.
1288            *args: Optional positional arguments (chunk_delimiter, metadata).
1289            **kwargs: Optional keyword arguments (chunk_delimiter, metadata).
1290
1291        Returns:
1292            List[SecureDocument]: Ingested and validated SecureDocument objects.
1293
1294        Raises:
1295            IngestionRejectedException: If strict validation fails under STRICT_FAIL_CLOSED mode.
1296        """
1297        chunk_delimiter = str(kwargs.get("chunk_delimiter", args[0] if len(args) > 0 else DEFAULT_CHUNK_DELIMITER))
1298        metadata = kwargs.get("metadata", args[1] if len(args) > 1 else None)
1299
1300        base_meta = dict(metadata) if metadata is not None else {}
1301        base_meta["source"] = source_name
1302        doc_id = base_meta.get(SEC_DOCUMENT_ID, str(uuid.uuid4()))
1303        base_meta[SEC_DOCUMENT_ID] = doc_id
1304
1305        if self._text_splitter is not None:
1306            raw_docs = [Document(page_content=text, metadata=dict(base_meta))]
1307            return self.ingest_documents(raw_docs)
1308
1309        raw_chunks = [chunk.strip() for chunk in text.split(chunk_delimiter) if chunk.strip()]
1310        if not raw_chunks and text:
1311            raw_chunks = [text]
1312        raw_docs = [Document(page_content=chunk, metadata=dict(base_meta)) for chunk in raw_chunks]
1313        return self.ingest_documents(raw_docs)

Ingest raw text string into secure document chunks with full envelope validation.

Args: text: Raw text content string. source_name: Source document name or identifier. *args: Optional positional arguments (chunk_delimiter, metadata). **kwargs: Optional keyword arguments (chunk_delimiter, metadata).

Returns: List[SecureDocument]: Ingested and validated SecureDocument objects.

Raises: IngestionRejectedException: If strict validation fails under STRICT_FAIL_CLOSED mode.

class SecureTextSplitter(langchain_core.documents.transformers.BaseDocumentTransformer): View Source

683class SecureTextSplitter(BaseDocumentTransformer):
684    """Stage 2 TextSplitter splitting documents into bounded chunks while propagating security metadata envelopes."""
685
686    def __init__(self, *args: Any, **kwargs: Any):
687        """Initialize SecureTextSplitter.
688
689        Args:
690            *args: Optional positional arguments (chunk_size, chunk_overlap, separators).
691            **kwargs: Optional keyword arguments (chunk_size, chunk_overlap, separators).
692        """
693        chunk_size = kwargs.get("chunk_size", args[0] if len(args) > 0 else DEFAULT_CHUNK_SIZE)
694        chunk_overlap = kwargs.get("chunk_overlap", args[1] if len(args) > 1 else DEFAULT_CHUNK_OVERLAP)
695        separators = kwargs.get("separators", args[2] if len(args) > 2 else DEFAULT_TEXT_SEPARATORS)
696
697        if int(chunk_size) <= 0:
698            raise ValueError("chunk_size must be positive")
699        if int(chunk_overlap) < 0:
700            raise ValueError("chunk_overlap must be non-negative")
701        if int(chunk_overlap) >= int(chunk_size):
702            raise ValueError("chunk_overlap must be smaller than chunk_size")
703
704        self._chunk_size = int(chunk_size)
705        self._chunk_overlap = int(chunk_overlap)
706        self._separators = list(separators)
707
708    @classmethod
709    def create(cls, *args: Any, **kwargs: Any) -> "SecureTextSplitter":
710        """Factory method constructing SecureTextSplitter.
711
712        Args:
713            *args: Optional positional arguments (chunk_size, chunk_overlap, separators).
714            **kwargs: Optional keyword arguments (chunk_size, chunk_overlap, separators).
715
716        Returns:
717            SecureTextSplitter: Newly constructed text splitter instance.
718        """
719        return cls(*args, **kwargs)
720
721    @classmethod
722    def create_default(cls) -> "SecureTextSplitter":
723        """Factory method constructing default SecureTextSplitter."""
724        return cls()
725
726    @property
727    def chunk_size(self) -> int:
728        """Get configured chunk size."""
729        return self._chunk_size
730
731    @property
732    def chunk_overlap(self) -> int:
733        """Get configured chunk overlap."""
734        return self._chunk_overlap
735
736    @property
737    def separators(self) -> List[str]:
738        """Get list of text separators."""
739        return list(self._separators)
740
741    def _merge_splits(self, splits: List[str], separator: str) -> List[str]:
742        docs: List[str] = []
743        current_doc: List[str] = []
744        total = 0
745        sep_len = len(separator) if separator else 0
746
747        for d in splits:
748            _len = len(d)
749            if total + _len + (sep_len if current_doc else 0) > self._chunk_size:
750                if total > 0:
751                    doc = separator.join(current_doc).strip() if separator else "".join(current_doc)
752                    if doc:
753                        docs.append(doc)
754                    while current_doc and (total > self._chunk_overlap or (total + _len + (sep_len if current_doc else 0) > self._chunk_size and total > 0)):
755                        popped = current_doc.pop(0)
756                        total -= len(popped) + (sep_len if current_doc else 0)
757                if _len > self._chunk_size:
758                    start = 0
759                    step = self._chunk_size - self._chunk_overlap if self._chunk_size > self._chunk_overlap else self._chunk_size
760                    while start < _len:
761                        end = min(start + self._chunk_size, _len)
762                        slice_chunk = d[start:end].strip() if separator else d[start:end]
763                        if slice_chunk:
764                            docs.append(slice_chunk)
765                        start += step
766                    current_doc = []
767                    total = 0
768                    continue
769
770            current_doc.append(d)
771            total += _len + (sep_len if len(current_doc) > 1 else 0)
772
773        if current_doc:
774            doc = separator.join(current_doc).strip() if separator else "".join(current_doc)
775            if doc:
776                docs.append(doc)
777
778        return docs
779
780    def _split_text_recursive(self, text: str, separators: Sequence[str]) -> List[str]:
781        if not text:
782            return []
783        chosen_separator: str = ""
784        new_separators: List[str] = []
785        for i, sep in enumerate(separators):
786            if sep == "":
787                chosen_separator = ""
788                break
789            if sep in text:
790                chosen_separator = sep
791                new_separators = list(separators[i + 1:])
792                break
793
794        if chosen_separator == "":
795            splits = list(text)
796        else:
797            splits = text.split(chosen_separator)
798
799        good_splits: List[str] = []
800        for s in splits:
801            if not s and chosen_separator != "":
802                continue
803            if len(s) > self._chunk_size and new_separators:
804                sub_splits = self._split_text_recursive(s, new_separators)
805                good_splits.extend(sub_splits)
806            else:
807                good_splits.append(s)
808
809        return self._merge_splits(good_splits, chosen_separator)
810
811    def split_text(self, text: str) -> List[str]:
812        """Split incoming text into chunks.
813
814        Args:
815            text: Raw input text.
816
817        Returns:
818            List[str]: List of text chunks.
819        """
820        if not text:
821            return []
822        if len(text) <= self._chunk_size:
823            return [text.strip()] if text.strip() else []
824        return self._split_text_recursive(text, self._separators)
825
826    def split_documents(self, documents: Sequence[Any]) -> List[Any]:
827        """Split documents into chunks while propagating metadata envelopes.
828
829        Args:
830            documents: Input sequence of document objects.
831
832        Returns:
833            List[Any]: Chunks with parent metadata propagated.
834        """
835        chunks: List[Any] = []
836        for doc in documents:
837            text = getattr(doc, "page_content", str(doc))
838            metadata = dict(getattr(doc, "metadata", {}))
839            text_splits = self.split_text(text)
840            if not text_splits and text:
841                text_splits = [text]
842            elif not text_splits and text == "":
843                text_splits = [""]
844            for chunk_idx, split_chunk in enumerate(text_splits):
845                chunk_meta = dict(metadata)
846                chunk_meta[SEC_CHUNK_INDEX] = chunk_idx
847                if isinstance(doc, SecureDocument):
848                    chunks.append(SecureDocument.create(page_content=split_chunk, metadata=chunk_meta))
849                else:
850                    chunks.append(Document(page_content=split_chunk, metadata=chunk_meta))
851        return chunks
852
853    def transform_documents(self, documents: Sequence[Any], **kwargs: Any) -> Sequence[Any]:
854        """Transform documents by splitting them into chunks with metadata propagation.
855
856        Args:
857            documents: Documents to transform.
858
859        Returns:
860            Sequence[Any]: Split document chunks.
861        """
862        return self.split_documents(documents)

Stage 2 TextSplitter splitting documents into bounded chunks while propagating security metadata envelopes.

SecureTextSplitter(*args: Any, **kwargs: Any) View Source

686    def __init__(self, *args: Any, **kwargs: Any):
687        """Initialize SecureTextSplitter.
688
689        Args:
690            *args: Optional positional arguments (chunk_size, chunk_overlap, separators).
691            **kwargs: Optional keyword arguments (chunk_size, chunk_overlap, separators).
692        """
693        chunk_size = kwargs.get("chunk_size", args[0] if len(args) > 0 else DEFAULT_CHUNK_SIZE)
694        chunk_overlap = kwargs.get("chunk_overlap", args[1] if len(args) > 1 else DEFAULT_CHUNK_OVERLAP)
695        separators = kwargs.get("separators", args[2] if len(args) > 2 else DEFAULT_TEXT_SEPARATORS)
696
697        if int(chunk_size) <= 0:
698            raise ValueError("chunk_size must be positive")
699        if int(chunk_overlap) < 0:
700            raise ValueError("chunk_overlap must be non-negative")
701        if int(chunk_overlap) >= int(chunk_size):
702            raise ValueError("chunk_overlap must be smaller than chunk_size")
703
704        self._chunk_size = int(chunk_size)
705        self._chunk_overlap = int(chunk_overlap)
706        self._separators = list(separators)

Initialize SecureTextSplitter.

Args: *args: Optional positional arguments (chunk_size, chunk_overlap, separators). **kwargs: Optional keyword arguments (chunk_size, chunk_overlap, separators).

@classmethod

def create( cls, *args: Any, **kwargs: Any) -> SecureTextSplitter: View Source

708    @classmethod
709    def create(cls, *args: Any, **kwargs: Any) -> "SecureTextSplitter":
710        """Factory method constructing SecureTextSplitter.
711
712        Args:
713            *args: Optional positional arguments (chunk_size, chunk_overlap, separators).
714            **kwargs: Optional keyword arguments (chunk_size, chunk_overlap, separators).
715
716        Returns:
717            SecureTextSplitter: Newly constructed text splitter instance.
718        """
719        return cls(*args, **kwargs)

Factory method constructing SecureTextSplitter.

Args: *args: Optional positional arguments (chunk_size, chunk_overlap, separators). **kwargs: Optional keyword arguments (chunk_size, chunk_overlap, separators).

Returns: SecureTextSplitter: Newly constructed text splitter instance.

@classmethod

def create_default( cls) -> SecureTextSplitter: View Source

721    @classmethod
722    def create_default(cls) -> "SecureTextSplitter":
723        """Factory method constructing default SecureTextSplitter."""
724        return cls()

Factory method constructing default SecureTextSplitter.

chunk_size: int View Source

726    @property
727    def chunk_size(self) -> int:
728        """Get configured chunk size."""
729        return self._chunk_size

Get configured chunk size.

chunk_overlap: int View Source

731    @property
732    def chunk_overlap(self) -> int:
733        """Get configured chunk overlap."""
734        return self._chunk_overlap

Get configured chunk overlap.

separators: List[str] View Source

736    @property
737    def separators(self) -> List[str]:
738        """Get list of text separators."""
739        return list(self._separators)

Get list of text separators.

def split_text(self, text: str) -> List[str]: View Source

811    def split_text(self, text: str) -> List[str]:
812        """Split incoming text into chunks.
813
814        Args:
815            text: Raw input text.
816
817        Returns:
818            List[str]: List of text chunks.
819        """
820        if not text:
821            return []
822        if len(text) <= self._chunk_size:
823            return [text.strip()] if text.strip() else []
824        return self._split_text_recursive(text, self._separators)

Split incoming text into chunks.

Args: text: Raw input text.

Returns: List[str]: List of text chunks.

def split_documents(self, documents: Sequence[Any]) -> List[Any]: View Source

826    def split_documents(self, documents: Sequence[Any]) -> List[Any]:
827        """Split documents into chunks while propagating metadata envelopes.
828
829        Args:
830            documents: Input sequence of document objects.
831
832        Returns:
833            List[Any]: Chunks with parent metadata propagated.
834        """
835        chunks: List[Any] = []
836        for doc in documents:
837            text = getattr(doc, "page_content", str(doc))
838            metadata = dict(getattr(doc, "metadata", {}))
839            text_splits = self.split_text(text)
840            if not text_splits and text:
841                text_splits = [text]
842            elif not text_splits and text == "":
843                text_splits = [""]
844            for chunk_idx, split_chunk in enumerate(text_splits):
845                chunk_meta = dict(metadata)
846                chunk_meta[SEC_CHUNK_INDEX] = chunk_idx
847                if isinstance(doc, SecureDocument):
848                    chunks.append(SecureDocument.create(page_content=split_chunk, metadata=chunk_meta))
849                else:
850                    chunks.append(Document(page_content=split_chunk, metadata=chunk_meta))
851        return chunks

Split documents into chunks while propagating metadata envelopes.

Args: documents: Input sequence of document objects.

Returns: List[Any]: Chunks with parent metadata propagated.

def transform_documents(self, documents: Sequence[Any], **kwargs: Any) -> Sequence[Any]: View Source

853    def transform_documents(self, documents: Sequence[Any], **kwargs: Any) -> Sequence[Any]:
854        """Transform documents by splitting them into chunks with metadata propagation.
855
856        Args:
857            documents: Documents to transform.
858
859        Returns:
860            Sequence[Any]: Split document chunks.
861        """
862        return self.split_documents(documents)

Transform documents by splitting them into chunks with metadata propagation.

Args: documents: Documents to transform.

Returns: Sequence[Any]: Split document chunks.

class ChunkSecurityEnricherSPI(abc.ABC): View Source

328class ChunkSecurityEnricherSPI(ABC):
329    """Extensibility interface for enterprise DLP, secret scanners, or regex classifiers."""
330
331    @abstractmethod
332    def enrich(self, context: ChunkEnrichmentContext) -> Dict[str, Any]:
333        """Returns updated security envelope metadata dictionary.
334
335        Args:
336            context: Chunk enrichment context container.
337
338        Returns:
339            Dict[str, Any]: Security envelope metadata key-value updates.
340        """
341        pass

Extensibility interface for enterprise DLP, secret scanners, or regex classifiers.

@abstractmethod

def enrich( self, context: ChunkEnrichmentContext) -> Dict[str, Any]: View Source

331    @abstractmethod
332    def enrich(self, context: ChunkEnrichmentContext) -> Dict[str, Any]:
333        """Returns updated security envelope metadata dictionary.
334
335        Args:
336            context: Chunk enrichment context container.
337
338        Returns:
339            Dict[str, Any]: Security envelope metadata key-value updates.
340        """
341        pass

Returns updated security envelope metadata dictionary.

Args: context: Chunk enrichment context container.

Returns: Dict[str, Any]: Security envelope metadata key-value updates.

@dataclass(frozen=True)

class ChunkEnrichmentContext: View Source

286@dataclass(frozen=True)
287class ChunkEnrichmentContext:
288    """Read-only evaluation context supplied to ChunkSecurityEnricherSPI implementations.
289
290    Attributes:
291        chunk_text (str): Content of the chunk being enriched.
292        parent_document_metadata (Dict[str, Any]): Parent document security envelope dictionary.
293        current_security_envelope (Dict[str, Any]): Active security metadata envelope.
294        chunk_index (int): Zero-based chunk index.
295    """
296    chunk_text: str
297    parent_document_metadata: Dict[str, Any]
298    current_security_envelope: Dict[str, Any]
299    chunk_index: int
300
301    @classmethod
302    def create(
303        cls,
304        chunk_text: str,
305        parent_document_metadata: Dict[str, Any],
306        current_security_envelope: Dict[str, Any],
307        chunk_index: int,
308    ) -> "ChunkEnrichmentContext":
309        """Factory method to construct ChunkEnrichmentContext.
310
311        Args:
312            chunk_text: Chunk text content.
313            parent_document_metadata: Parent metadata map.
314            current_security_envelope: Current security envelope map.
315            chunk_index: Index of chunk.
316
317        Returns:
318            ChunkEnrichmentContext: New context instance.
319        """
320        return cls(
321            chunk_text=chunk_text,
322            parent_document_metadata=parent_document_metadata,
323            current_security_envelope=current_security_envelope,
324            chunk_index=chunk_index,
325        )

Read-only evaluation context supplied to ChunkSecurityEnricherSPI implementations.

Attributes: chunk_text (str): Content of the chunk being enriched. parent_document_metadata (Dict[str, Any]): Parent document security envelope dictionary. current_security_envelope (Dict[str, Any]): Active security metadata envelope. chunk_index (int): Zero-based chunk index.

ChunkEnrichmentContext( chunk_text: str, parent_document_metadata: Dict[str, Any], current_security_envelope: Dict[str, Any], chunk_index: int)

chunk_text: str

parent_document_metadata: Dict[str, Any]

current_security_envelope: Dict[str, Any]

chunk_index: int

@classmethod

def create( cls, chunk_text: str, parent_document_metadata: Dict[str, Any], current_security_envelope: Dict[str, Any], chunk_index: int) -> ChunkEnrichmentContext: View Source

301    @classmethod
302    def create(
303        cls,
304        chunk_text: str,
305        parent_document_metadata: Dict[str, Any],
306        current_security_envelope: Dict[str, Any],
307        chunk_index: int,
308    ) -> "ChunkEnrichmentContext":
309        """Factory method to construct ChunkEnrichmentContext.
310
311        Args:
312            chunk_text: Chunk text content.
313            parent_document_metadata: Parent metadata map.
314            current_security_envelope: Current security envelope map.
315            chunk_index: Index of chunk.
316
317        Returns:
318            ChunkEnrichmentContext: New context instance.
319        """
320        return cls(
321            chunk_text=chunk_text,
322            parent_document_metadata=parent_document_metadata,
323            current_security_envelope=current_security_envelope,
324            chunk_index=chunk_index,
325        )

Factory method to construct ChunkEnrichmentContext.

Args: chunk_text: Chunk text content. parent_document_metadata: Parent metadata map. current_security_envelope: Current security envelope map. chunk_index: Index of chunk.

Returns: ChunkEnrichmentContext: New context instance.

class HeadingClassificationEnricher(langchain_security_extension.security_adapter_langchain.ChunkSecurityEnricherSPI): View Source

344class HeadingClassificationEnricher(ChunkSecurityEnricherSPI):
345    """Built-in ChunkSecurityEnricherSPI parsing structural Markdown headings, XML security tags, and inline role overrides."""
346
347    @classmethod
348    def create(cls) -> "HeadingClassificationEnricher":
349        """Construct a HeadingClassificationEnricher instance.
350
351        Returns:
352            HeadingClassificationEnricher: New enricher instance.
353        """
354        return cls()
355
356    def enrich(self, context: ChunkEnrichmentContext) -> Dict[str, Any]:
357        """Parse structural headers and inline tags from chunk text to produce security envelope updates.
358
359        Args:
360            context: ChunkEnrichmentContext providing chunk text and metadata.
361
362        Returns:
363            Dict[str, Any]: Security envelope updates parsed from headings and inline tags.
364        """
365        updates: Dict[str, Any] = {}
366        content = context.chunk_text
367
368        header_match = HEADER_CLASSIFICATION_PATTERN.search(content)
369        if header_match:
370            header_cls = header_match.group(1).upper()
371            updates[SEC_CLASSIFICATION_LEVEL] = header_cls
372            updates[SEC_CLEARANCE_FLOOR] = CLASSIFICATION_TO_RANK.get(header_cls, 0)
373
374        xml_match = XML_CLASSIFICATION_PATTERN.search(content)
375        if xml_match:
376            xml_cls = xml_match.group(1).upper()
377            xml_rank = CLASSIFICATION_TO_RANK.get(xml_cls, 0)
378            if SEC_CLEARANCE_FLOOR not in updates or xml_rank > updates[SEC_CLEARANCE_FLOOR] or xml_rank < context.current_security_envelope.get(SEC_CLEARANCE_FLOOR, 0):
379                updates[SEC_CLASSIFICATION_LEVEL] = xml_cls
380                updates[SEC_CLEARANCE_FLOOR] = xml_rank
381
382        inline_roles_match = INLINE_ROLES_PATTERN.search(content)
383        if inline_roles_match:
384            parsed_roles = [r.strip() for r in inline_roles_match.group(1).split(",") if r.strip()]
385            if parsed_roles:
386                updates[SEC_ALLOWED_ROLES] = parsed_roles
387                updates[SEC_REQUIRED_ROLES] = parsed_roles
388                updates[f"{SEC_ALLOWED_ROLES}_str"] = f",{','.join(parsed_roles)},"
389
390        return updates

Built-in ChunkSecurityEnricherSPI parsing structural Markdown headings, XML security tags, and inline role overrides.

@classmethod

def create( cls) -> HeadingClassificationEnricher: View Source

347    @classmethod
348    def create(cls) -> "HeadingClassificationEnricher":
349        """Construct a HeadingClassificationEnricher instance.
350
351        Returns:
352            HeadingClassificationEnricher: New enricher instance.
353        """
354        return cls()

Construct a HeadingClassificationEnricher instance.

Returns: HeadingClassificationEnricher: New enricher instance.

def enrich( self, context: ChunkEnrichmentContext) -> Dict[str, Any]: View Source

356    def enrich(self, context: ChunkEnrichmentContext) -> Dict[str, Any]:
357        """Parse structural headers and inline tags from chunk text to produce security envelope updates.
358
359        Args:
360            context: ChunkEnrichmentContext providing chunk text and metadata.
361
362        Returns:
363            Dict[str, Any]: Security envelope updates parsed from headings and inline tags.
364        """
365        updates: Dict[str, Any] = {}
366        content = context.chunk_text
367
368        header_match = HEADER_CLASSIFICATION_PATTERN.search(content)
369        if header_match:
370            header_cls = header_match.group(1).upper()
371            updates[SEC_CLASSIFICATION_LEVEL] = header_cls
372            updates[SEC_CLEARANCE_FLOOR] = CLASSIFICATION_TO_RANK.get(header_cls, 0)
373
374        xml_match = XML_CLASSIFICATION_PATTERN.search(content)
375        if xml_match:
376            xml_cls = xml_match.group(1).upper()
377            xml_rank = CLASSIFICATION_TO_RANK.get(xml_cls, 0)
378            if SEC_CLEARANCE_FLOOR not in updates or xml_rank > updates[SEC_CLEARANCE_FLOOR] or xml_rank < context.current_security_envelope.get(SEC_CLEARANCE_FLOOR, 0):
379                updates[SEC_CLASSIFICATION_LEVEL] = xml_cls
380                updates[SEC_CLEARANCE_FLOOR] = xml_rank
381
382        inline_roles_match = INLINE_ROLES_PATTERN.search(content)
383        if inline_roles_match:
384            parsed_roles = [r.strip() for r in inline_roles_match.group(1).split(",") if r.strip()]
385            if parsed_roles:
386                updates[SEC_ALLOWED_ROLES] = parsed_roles
387                updates[SEC_REQUIRED_ROLES] = parsed_roles
388                updates[f"{SEC_ALLOWED_ROLES}_str"] = f",{','.join(parsed_roles)},"
389
390        return updates

Parse structural headers and inline tags from chunk text to produce security envelope updates.

Args: context: ChunkEnrichmentContext providing chunk text and metadata.

Returns: Dict[str, Any]: Security envelope updates parsed from headings and inline tags.

class DlpSecretDetectionEnricher(langchain_security_extension.security_adapter_langchain.ChunkSecurityEnricherSPI): View Source

393class DlpSecretDetectionEnricher(ChunkSecurityEnricherSPI):
394    """Built-in ChunkSecurityEnricherSPI scanning chunk text for credentials, secrets, and PII to elevate classification."""
395
396    @classmethod
397    def create(cls) -> "DlpSecretDetectionEnricher":
398        """Construct a DlpSecretDetectionEnricher instance.
399
400        Returns:
401            DlpSecretDetectionEnricher: New DLP enricher instance.
402        """
403        return cls()
404
405    def enrich(self, context: ChunkEnrichmentContext) -> Dict[str, Any]:
406        """Scan chunk text for DLP patterns and elevate to RESTRICTED if detected.
407
408        Args:
409            context: ChunkEnrichmentContext providing chunk text and metadata.
410
411        Returns:
412            Dict[str, Any]: Security envelope updates if secrets/PII are detected, or empty dict.
413        """
414        content = context.chunk_text
415        for pattern in DLP_SECRET_PATTERNS:
416            if pattern.search(content):
417                return {
418                    SEC_CLASSIFICATION_LEVEL: CLASSIFICATION_RESTRICTED,
419                    SEC_CLEARANCE_FLOOR: CLEARANCE_RANK_RESTRICTED,
420                    SEC_DLP_FLAGGED: True,
421                }
422        return {}

Built-in ChunkSecurityEnricherSPI scanning chunk text for credentials, secrets, and PII to elevate classification.

@classmethod

def create( cls) -> DlpSecretDetectionEnricher: View Source

396    @classmethod
397    def create(cls) -> "DlpSecretDetectionEnricher":
398        """Construct a DlpSecretDetectionEnricher instance.
399
400        Returns:
401            DlpSecretDetectionEnricher: New DLP enricher instance.
402        """
403        return cls()

Construct a DlpSecretDetectionEnricher instance.

Returns: DlpSecretDetectionEnricher: New DLP enricher instance.

def enrich( self, context: ChunkEnrichmentContext) -> Dict[str, Any]: View Source

405    def enrich(self, context: ChunkEnrichmentContext) -> Dict[str, Any]:
406        """Scan chunk text for DLP patterns and elevate to RESTRICTED if detected.
407
408        Args:
409            context: ChunkEnrichmentContext providing chunk text and metadata.
410
411        Returns:
412            Dict[str, Any]: Security envelope updates if secrets/PII are detected, or empty dict.
413        """
414        content = context.chunk_text
415        for pattern in DLP_SECRET_PATTERNS:
416            if pattern.search(content):
417                return {
418                    SEC_CLASSIFICATION_LEVEL: CLASSIFICATION_RESTRICTED,
419                    SEC_CLEARANCE_FLOOR: CLEARANCE_RANK_RESTRICTED,
420                    SEC_DLP_FLAGGED: True,
421                }
422        return {}

Scan chunk text for DLP patterns and elevate to RESTRICTED if detected.

Args: context: ChunkEnrichmentContext providing chunk text and metadata.

Returns: Dict[str, Any]: Security envelope updates if secrets/PII are detected, or empty dict.

class SecuredAgentState(typing_extensions.TypedDict): View Source

17class SecuredAgentState(TypedDict):
18    """LangGraph state schema incorporating in-band security envelope.
19
20    Attributes:
21        messages (Sequence[Any]): Sequence of chat message objects.
22        sec_invocation_parameters (Dict[str, Any]): Embedded security invocation parameters state dictionary.
23    """
24    messages: Sequence[Any]
25    sec_invocation_parameters: Dict[str, Any]

LangGraph state schema incorporating in-band security envelope.

Attributes: messages (Sequence[Any]): Sequence of chat message objects. sec_invocation_parameters (Dict[str, Any]): Embedded security invocation parameters state dictionary.

messages: Sequence[Any]

sec_invocation_parameters: Dict[str, Any]

class LangGraphNodeGuard: View Source

 28class LangGraphNodeGuard:
 29    """Interceptor for LangGraph graph nodes enforcing security context propagation and authorization.
 30
 31    Attributes:
 32        node_name (str): Identifier of graph node target.
 33        required_roles (Sequence[str]): Sequence of required role strings.
 34        min_clearance (int): Minimum required clearance level rank.
 35        pdp_engine (PolicyDecisionEngine): PDP decision engine instance.
 36    """
 37
 38    def __init__(
 39        self,
 40        node_name: str,
 41        required_roles: Sequence[str],
 42        min_clearance: int,
 43        pdp_engine: PolicyDecisionEngine,
 44    ):
 45        """Initialize LangGraphNodeGuard.
 46
 47        Args:
 48            node_name: Target node identifier.
 49            required_roles: Roles sequence.
 50            min_clearance: Clearance level floor.
 51            pdp_engine: PDP decision engine instance.
 52        """
 53        self._node_name = node_name
 54        self._required_roles = list(required_roles)
 55        self._min_clearance = min_clearance
 56        self._pdp_engine = pdp_engine
 57
 58    @classmethod
 59    def create(
 60        cls,
 61        node_name: str,
 62        required_roles: Sequence[str],
 63        min_clearance: int,
 64        pdp_engine: PolicyDecisionEngine,
 65    ):
 66        """Factory method constructing a LangGraphNodeGuard.
 67
 68        Args:
 69            node_name: Target graph node name.
 70            required_roles: Required role strings sequence.
 71            min_clearance: Clearance floor rank.
 72            pdp_engine: PDP decision engine.
 73
 74        Returns:
 75            LangGraphNodeGuard: New guard instance.
 76        """
 77        return cls(
 78            node_name=node_name,
 79            required_roles=required_roles,
 80            min_clearance=min_clearance,
 81            pdp_engine=pdp_engine,
 82        )
 83
 84    @property
 85    def node_name(self) -> str:
 86        """Get the guarded node identifier string."""
 87        return self._node_name
 88
 89    @property
 90    def required_roles(self) -> Sequence[str]:
 91        """Get copy of required role identifiers."""
 92        return list(self._required_roles)
 93
 94    @property
 95    def min_clearance(self) -> int:
 96        """Get minimum clearance level rank."""
 97        return self._min_clearance
 98
 99    @property
100    def pdp_engine(self) -> PolicyDecisionEngine:
101        """Get the configured PolicyDecisionEngine instance."""
102        return self._pdp_engine
103
104    def wrap_node(self, node_fn: Callable[[Dict[str, Any]], Dict[str, Any]]) -> Callable[[Dict[str, Any]], Dict[str, Any]]:
105        """Wraps a LangGraph node function with security evaluation.
106
107        Args:
108            node_fn: Target graph node callable accepting state dictionary and returning updated state.
109
110        Returns:
111            Callable[[Dict[str, Any]], Dict[str, Any]]: Guarded node function.
112        """
113
114        def guarded_node(state: Dict[str, Any]) -> Dict[str, Any]:
115            inv_params = InvocationParameters.deserialize_from_state(state)
116            identity = inv_params.security_identity
117
118            request = PolicyEvaluationRequest.create(
119                subject=identity,
120                action=ACTION_AGENT_INVOKE,
121                resource={
122                    "target_id": self._node_name,
123                    "tenant_id": identity.tenant_id,
124                    "required_roles": self._required_roles,
125                    "min_clearance": self._min_clearance,
126                },
127                context={"node_name": self._node_name},
128            )
129
130            decision = self._pdp_engine.evaluate(request)
131            if not decision.is_permitted:
132                raise SecurityAuthorizationError.create(
133                    message=f"LangGraph Node '{self._node_name}' authorization failed: {decision.diagnostic_reason}",
134                    diagnostic_reason=decision.diagnostic_reason,
135                    obligations=decision.obligations,
136                )
137
138            # Ensure security envelope persists in returned state
139            output_state = node_fn(state)
140            if isinstance(output_state, dict) and SECURITY_INVOCATION_PARAMS_KEY not in output_state:
141                output_state[SECURITY_INVOCATION_PARAMS_KEY] = state.get(
142                    SECURITY_INVOCATION_PARAMS_KEY,
143                    inv_params.serialize_to_state()[SECURITY_INVOCATION_PARAMS_KEY],
144                )
145            return output_state
146
147        return guarded_node

Interceptor for LangGraph graph nodes enforcing security context propagation and authorization.

Attributes: node_name (str): Identifier of graph node target. required_roles (Sequence[str]): Sequence of required role strings. min_clearance (int): Minimum required clearance level rank. pdp_engine (PolicyDecisionEngine): PDP decision engine instance.

LangGraphNodeGuard( node_name: str, required_roles: Sequence[str], min_clearance: int, pdp_engine: langchain_security_extension.security_core.PolicyDecisionEngine) View Source

38    def __init__(
39        self,
40        node_name: str,
41        required_roles: Sequence[str],
42        min_clearance: int,
43        pdp_engine: PolicyDecisionEngine,
44    ):
45        """Initialize LangGraphNodeGuard.
46
47        Args:
48            node_name: Target node identifier.
49            required_roles: Roles sequence.
50            min_clearance: Clearance level floor.
51            pdp_engine: PDP decision engine instance.
52        """
53        self._node_name = node_name
54        self._required_roles = list(required_roles)
55        self._min_clearance = min_clearance
56        self._pdp_engine = pdp_engine

Initialize LangGraphNodeGuard.

Args: node_name: Target node identifier. required_roles: Roles sequence. min_clearance: Clearance level floor. pdp_engine: PDP decision engine instance.

@classmethod

def create( cls, node_name: str, required_roles: Sequence[str], min_clearance: int, pdp_engine: langchain_security_extension.security_core.PolicyDecisionEngine): View Source

58    @classmethod
59    def create(
60        cls,
61        node_name: str,
62        required_roles: Sequence[str],
63        min_clearance: int,
64        pdp_engine: PolicyDecisionEngine,
65    ):
66        """Factory method constructing a LangGraphNodeGuard.
67
68        Args:
69            node_name: Target graph node name.
70            required_roles: Required role strings sequence.
71            min_clearance: Clearance floor rank.
72            pdp_engine: PDP decision engine.
73
74        Returns:
75            LangGraphNodeGuard: New guard instance.
76        """
77        return cls(
78            node_name=node_name,
79            required_roles=required_roles,
80            min_clearance=min_clearance,
81            pdp_engine=pdp_engine,
82        )

Factory method constructing a LangGraphNodeGuard.

Args: node_name: Target graph node name. required_roles: Required role strings sequence. min_clearance: Clearance floor rank. pdp_engine: PDP decision engine.

Returns: LangGraphNodeGuard: New guard instance.

node_name: str View Source

84    @property
85    def node_name(self) -> str:
86        """Get the guarded node identifier string."""
87        return self._node_name

Get the guarded node identifier string.

required_roles: Sequence[str] View Source

89    @property
90    def required_roles(self) -> Sequence[str]:
91        """Get copy of required role identifiers."""
92        return list(self._required_roles)

Get copy of required role identifiers.

min_clearance: int View Source

94    @property
95    def min_clearance(self) -> int:
96        """Get minimum clearance level rank."""
97        return self._min_clearance

Get minimum clearance level rank.

pdp_engine: langchain_security_extension.security_core.PolicyDecisionEngine View Source

 99    @property
100    def pdp_engine(self) -> PolicyDecisionEngine:
101        """Get the configured PolicyDecisionEngine instance."""
102        return self._pdp_engine

Get the configured PolicyDecisionEngine instance.

def wrap_node( self, node_fn: Callable[[Dict[str, Any]], Dict[str, Any]]) -> Callable[[Dict[str, Any]], Dict[str, Any]]: View Source

104    def wrap_node(self, node_fn: Callable[[Dict[str, Any]], Dict[str, Any]]) -> Callable[[Dict[str, Any]], Dict[str, Any]]:
105        """Wraps a LangGraph node function with security evaluation.
106
107        Args:
108            node_fn: Target graph node callable accepting state dictionary and returning updated state.
109
110        Returns:
111            Callable[[Dict[str, Any]], Dict[str, Any]]: Guarded node function.
112        """
113
114        def guarded_node(state: Dict[str, Any]) -> Dict[str, Any]:
115            inv_params = InvocationParameters.deserialize_from_state(state)
116            identity = inv_params.security_identity
117
118            request = PolicyEvaluationRequest.create(
119                subject=identity,
120                action=ACTION_AGENT_INVOKE,
121                resource={
122                    "target_id": self._node_name,
123                    "tenant_id": identity.tenant_id,
124                    "required_roles": self._required_roles,
125                    "min_clearance": self._min_clearance,
126                },
127                context={"node_name": self._node_name},
128            )
129
130            decision = self._pdp_engine.evaluate(request)
131            if not decision.is_permitted:
132                raise SecurityAuthorizationError.create(
133                    message=f"LangGraph Node '{self._node_name}' authorization failed: {decision.diagnostic_reason}",
134                    diagnostic_reason=decision.diagnostic_reason,
135                    obligations=decision.obligations,
136                )
137
138            # Ensure security envelope persists in returned state
139            output_state = node_fn(state)
140            if isinstance(output_state, dict) and SECURITY_INVOCATION_PARAMS_KEY not in output_state:
141                output_state[SECURITY_INVOCATION_PARAMS_KEY] = state.get(
142                    SECURITY_INVOCATION_PARAMS_KEY,
143                    inv_params.serialize_to_state()[SECURITY_INVOCATION_PARAMS_KEY],
144                )
145            return output_state
146
147        return guarded_node

Wraps a LangGraph node function with security evaluation.

Args: node_fn: Target graph node callable accepting state dictionary and returning updated state.

Returns: Callable[[Dict[str, Any]], Dict[str, Any]]: Guarded node function.

class RetryBudgetTracker: View Source

175class RetryBudgetTracker:
176    """Invocation-scoped retry budget tracker preventing recursive LLM retry loops."""
177
178    def __init__(self, max_feedback_violations: int):
179        """Initialize RetryBudgetTracker.
180
181        Args:
182            max_feedback_violations: Maximum feedback violation retry allowance before escalating to hard abort.
183        """
184        self._max_feedback_violations = max_feedback_violations
185
186    @classmethod
187    def create(cls, max_feedback_violations: int) -> "RetryBudgetTracker":
188        """Factory method to construct RetryBudgetTracker.
189
190        Args:
191            max_feedback_violations: Maximum retry threshold integer.
192
193        Returns:
194            RetryBudgetTracker: Configured tracker instance.
195        """
196        return cls(max_feedback_violations=max_feedback_violations)
197
198    @property
199    def max_feedback_violations(self) -> int:
200        """Get maximum feedback violation retry allowance."""
201        return self._max_feedback_violations
202
203    def get_retry_count(self, config: Optional[Dict[str, Any]]) -> int:
204        """Retrieve the current retry violation count from invocation configuration.
205
206        Args:
207            config: RunnableConfig dictionary containing configurable metadata.
208
209        Returns:
210            int: Current integer retry count.
211        """
212        if isinstance(config, dict) and isinstance(config.get("configurable"), dict):
213            return int(config["configurable"].get(CONFIG_RETRY_COUNT_KEY, 0))
214        return 0
215
216    def increment_retry_count(self, config: Optional[Dict[str, Any]]) -> int:
217        """Increment and persist the retry count inside the invocation configuration.
218
219        Args:
220            config: Target RunnableConfig dictionary.
221
222        Returns:
223            int: Newly updated retry count integer.
224        """
225        if isinstance(config, dict) and isinstance(config.get("configurable"), dict):
226            count = int(config["configurable"].get(CONFIG_RETRY_COUNT_KEY, 0)) + 1
227            config["configurable"][CONFIG_RETRY_COUNT_KEY] = count
228            return count
229        return 0
230
231    def is_budget_exceeded(self, config: Optional[Dict[str, Any]]) -> bool:
232        """Check if invocation-scoped retries have exceeded the maximum feedback limit.
233
234        Args:
235            config: Target RunnableConfig dictionary.
236
237        Returns:
238            bool: True if retry budget is exhausted, False otherwise.
239        """
240        return self.get_retry_count(config) > self._max_feedback_violations
241
242    def reset(self, config: Optional[Dict[str, Any]]) -> None:
243        """Reset the invocation retry counter to zero.
244
245        Args:
246            config: Target RunnableConfig dictionary.
247        """
248        if isinstance(config, dict) and isinstance(config.get("configurable"), dict):
249            config["configurable"][CONFIG_RETRY_COUNT_KEY] = 0

Invocation-scoped retry budget tracker preventing recursive LLM retry loops.

RetryBudgetTracker(max_feedback_violations: int) View Source

178    def __init__(self, max_feedback_violations: int):
179        """Initialize RetryBudgetTracker.
180
181        Args:
182            max_feedback_violations: Maximum feedback violation retry allowance before escalating to hard abort.
183        """
184        self._max_feedback_violations = max_feedback_violations

Initialize RetryBudgetTracker.

Args: max_feedback_violations: Maximum feedback violation retry allowance before escalating to hard abort.

@classmethod

def create( cls, max_feedback_violations: int) -> RetryBudgetTracker: View Source

186    @classmethod
187    def create(cls, max_feedback_violations: int) -> "RetryBudgetTracker":
188        """Factory method to construct RetryBudgetTracker.
189
190        Args:
191            max_feedback_violations: Maximum retry threshold integer.
192
193        Returns:
194            RetryBudgetTracker: Configured tracker instance.
195        """
196        return cls(max_feedback_violations=max_feedback_violations)

Factory method to construct RetryBudgetTracker.

Args: max_feedback_violations: Maximum retry threshold integer.

Returns: RetryBudgetTracker: Configured tracker instance.

max_feedback_violations: int View Source

198    @property
199    def max_feedback_violations(self) -> int:
200        """Get maximum feedback violation retry allowance."""
201        return self._max_feedback_violations

Get maximum feedback violation retry allowance.

def get_retry_count(self, config: Optional[Dict[str, Any]]) -> int: View Source

203    def get_retry_count(self, config: Optional[Dict[str, Any]]) -> int:
204        """Retrieve the current retry violation count from invocation configuration.
205
206        Args:
207            config: RunnableConfig dictionary containing configurable metadata.
208
209        Returns:
210            int: Current integer retry count.
211        """
212        if isinstance(config, dict) and isinstance(config.get("configurable"), dict):
213            return int(config["configurable"].get(CONFIG_RETRY_COUNT_KEY, 0))
214        return 0

Retrieve the current retry violation count from invocation configuration.

Args: config: RunnableConfig dictionary containing configurable metadata.

Returns: int: Current integer retry count.

def increment_retry_count(self, config: Optional[Dict[str, Any]]) -> int: View Source

216    def increment_retry_count(self, config: Optional[Dict[str, Any]]) -> int:
217        """Increment and persist the retry count inside the invocation configuration.
218
219        Args:
220            config: Target RunnableConfig dictionary.
221
222        Returns:
223            int: Newly updated retry count integer.
224        """
225        if isinstance(config, dict) and isinstance(config.get("configurable"), dict):
226            count = int(config["configurable"].get(CONFIG_RETRY_COUNT_KEY, 0)) + 1
227            config["configurable"][CONFIG_RETRY_COUNT_KEY] = count
228            return count
229        return 0

Increment and persist the retry count inside the invocation configuration.

Args: config: Target RunnableConfig dictionary.

Returns: int: Newly updated retry count integer.

def is_budget_exceeded(self, config: Optional[Dict[str, Any]]) -> bool: View Source

231    def is_budget_exceeded(self, config: Optional[Dict[str, Any]]) -> bool:
232        """Check if invocation-scoped retries have exceeded the maximum feedback limit.
233
234        Args:
235            config: Target RunnableConfig dictionary.
236
237        Returns:
238            bool: True if retry budget is exhausted, False otherwise.
239        """
240        return self.get_retry_count(config) > self._max_feedback_violations

Check if invocation-scoped retries have exceeded the maximum feedback limit.

Args: config: Target RunnableConfig dictionary.

Returns: bool: True if retry budget is exhausted, False otherwise.

def reset(self, config: Optional[Dict[str, Any]]) -> None: View Source

242    def reset(self, config: Optional[Dict[str, Any]]) -> None:
243        """Reset the invocation retry counter to zero.
244
245        Args:
246            config: Target RunnableConfig dictionary.
247        """
248        if isinstance(config, dict) and isinstance(config.get("configurable"), dict):
249            config["configurable"][CONFIG_RETRY_COUNT_KEY] = 0

Reset the invocation retry counter to zero.

Args: config: Target RunnableConfig dictionary.

class AbuseThrottleTracker: View Source

252class AbuseThrottleTracker:
253    """Sliding-window abuse throttle tracker detecting rapid cross-invocation denial floods."""
254
255    def __init__(self, window_seconds: float, max_denials: int):
256        """Initialize AbuseThrottleTracker.
257
258        Args:
259            window_seconds: Duration of sliding evaluation window in seconds.
260            max_denials: Maximum allowed denial events within window before throttling.
261        """
262        self._window_seconds = window_seconds
263        self._max_denials = max_denials
264        self._denial_history: Dict[str, List[float]] = {}
265
266    @classmethod
267    def create(cls, window_seconds: float, max_denials: int) -> "AbuseThrottleTracker":
268        """Factory method to construct AbuseThrottleTracker.
269
270        Args:
271            window_seconds: Window duration in seconds.
272            max_denials: Threshold count of denials.
273
274        Returns:
275            AbuseThrottleTracker: Configured tracker instance.
276        """
277        return cls(window_seconds=window_seconds, max_denials=max_denials)
278
279    @property
280    def window_seconds(self) -> float:
281        """Get sliding window duration in seconds."""
282        return self._window_seconds
283
284    @property
285    def max_denials(self) -> int:
286        """Get maximum allowed denials threshold."""
287        return self._max_denials
288
289    def _prune(self, subject_id: str, current_time: float) -> List[float]:
290        cutoff = current_time - self._window_seconds
291        timestamps = [t for t in self._denial_history.get(subject_id, []) if t >= cutoff]
292        self._denial_history[subject_id] = timestamps
293        return timestamps
294
295    def record_denial(self, subject_id: str, timestamp: float) -> bool:
296        """Record a denial occurrence for a subject and check if abuse threshold is breached.
297
298        Args:
299            subject_id: Subject identifier.
300            timestamp: Epoch timestamp in seconds.
301
302        Returns:
303            bool: True if subject exceeds abuse threshold (is abusing), False otherwise.
304        """
305        timestamps = self._prune(subject_id, timestamp)
306        timestamps.append(timestamp)
307        self._denial_history[subject_id] = timestamps
308        return len(timestamps) >= self._max_denials
309
310    def is_throttled(self, subject_id: str, timestamp: float) -> bool:
311        """Check if subject is currently throttled due to recent denial density.
312
313        Args:
314            subject_id: Subject identifier.
315            timestamp: Epoch timestamp in seconds.
316
317        Returns:
318            bool: True if subject is throttled, False otherwise.
319        """
320        timestamps = self._prune(subject_id, timestamp)
321        return len(timestamps) >= self._max_denials
322
323    def reset(self, *args: Any, **kwargs: Any) -> None:
324        """Reset denial history for a specific subject or all subjects.
325
326        Args:
327            *args: Optional positional subject_id.
328            **kwargs: Optional keyword subject_id.
329        """
330        subject_id = args[0] if len(args) > 0 else kwargs.get("subject_id")
331        if subject_id is not None:
332            self._denial_history.pop(str(subject_id), None)
333        else:
334            self._denial_history.clear()

Sliding-window abuse throttle tracker detecting rapid cross-invocation denial floods.

AbuseThrottleTracker(window_seconds: float, max_denials: int) View Source

255    def __init__(self, window_seconds: float, max_denials: int):
256        """Initialize AbuseThrottleTracker.
257
258        Args:
259            window_seconds: Duration of sliding evaluation window in seconds.
260            max_denials: Maximum allowed denial events within window before throttling.
261        """
262        self._window_seconds = window_seconds
263        self._max_denials = max_denials
264        self._denial_history: Dict[str, List[float]] = {}

Initialize AbuseThrottleTracker.

Args: window_seconds: Duration of sliding evaluation window in seconds. max_denials: Maximum allowed denial events within window before throttling.

@classmethod

def create( cls, window_seconds: float, max_denials: int) -> AbuseThrottleTracker: View Source

266    @classmethod
267    def create(cls, window_seconds: float, max_denials: int) -> "AbuseThrottleTracker":
268        """Factory method to construct AbuseThrottleTracker.
269
270        Args:
271            window_seconds: Window duration in seconds.
272            max_denials: Threshold count of denials.
273
274        Returns:
275            AbuseThrottleTracker: Configured tracker instance.
276        """
277        return cls(window_seconds=window_seconds, max_denials=max_denials)

Factory method to construct AbuseThrottleTracker.

Args: window_seconds: Window duration in seconds. max_denials: Threshold count of denials.

Returns: AbuseThrottleTracker: Configured tracker instance.

window_seconds: float View Source

279    @property
280    def window_seconds(self) -> float:
281        """Get sliding window duration in seconds."""
282        return self._window_seconds

Get sliding window duration in seconds.

max_denials: int View Source

284    @property
285    def max_denials(self) -> int:
286        """Get maximum allowed denials threshold."""
287        return self._max_denials

Get maximum allowed denials threshold.

def record_denial(self, subject_id: str, timestamp: float) -> bool: View Source

295    def record_denial(self, subject_id: str, timestamp: float) -> bool:
296        """Record a denial occurrence for a subject and check if abuse threshold is breached.
297
298        Args:
299            subject_id: Subject identifier.
300            timestamp: Epoch timestamp in seconds.
301
302        Returns:
303            bool: True if subject exceeds abuse threshold (is abusing), False otherwise.
304        """
305        timestamps = self._prune(subject_id, timestamp)
306        timestamps.append(timestamp)
307        self._denial_history[subject_id] = timestamps
308        return len(timestamps) >= self._max_denials

Record a denial occurrence for a subject and check if abuse threshold is breached.

Args: subject_id: Subject identifier. timestamp: Epoch timestamp in seconds.

Returns: bool: True if subject exceeds abuse threshold (is abusing), False otherwise.

def is_throttled(self, subject_id: str, timestamp: float) -> bool: View Source

310    def is_throttled(self, subject_id: str, timestamp: float) -> bool:
311        """Check if subject is currently throttled due to recent denial density.
312
313        Args:
314            subject_id: Subject identifier.
315            timestamp: Epoch timestamp in seconds.
316
317        Returns:
318            bool: True if subject is throttled, False otherwise.
319        """
320        timestamps = self._prune(subject_id, timestamp)
321        return len(timestamps) >= self._max_denials

Check if subject is currently throttled due to recent denial density.

Args: subject_id: Subject identifier. timestamp: Epoch timestamp in seconds.

Returns: bool: True if subject is throttled, False otherwise.

def reset(self, *args: Any, **kwargs: Any) -> None: View Source

323    def reset(self, *args: Any, **kwargs: Any) -> None:
324        """Reset denial history for a specific subject or all subjects.
325
326        Args:
327            *args: Optional positional subject_id.
328            **kwargs: Optional keyword subject_id.
329        """
330        subject_id = args[0] if len(args) > 0 else kwargs.get("subject_id")
331        if subject_id is not None:
332            self._denial_history.pop(str(subject_id), None)
333        else:
334            self._denial_history.clear()

Reset denial history for a specific subject or all subjects.

Args: *args: Optional positional subject_id. **kwargs: Optional keyword subject_id.