nat.middleware.defense.defense_middleware_pii#

PII Defense Middleware using Microsoft Presidio.

This middleware detects and anonymizes Personally Identifiable Information (PII) in function outputs using Microsoft Presidio.

Attributes#

logger

Classes#

`PIIDefenseMiddlewareConfig`	Configuration for PII Defense Middleware using Microsoft Presidio.
`PIIDefenseMiddleware`	PII Defense Middleware using Microsoft Presidio.

Module Contents#

logger#

class PIIDefenseMiddlewareConfig(/, **data: Any)#

Bases: nat.middleware.defense.defense_middleware.DefenseMiddlewareConfig

Configuration for PII Defense Middleware using Microsoft Presidio.

Detects PII in function outputs using Presidio’s rule-based entity recognition (no LLM required).

See <microsoft/presidio> for more information about Presidio.

Actions: - ‘partial_compliance’: Detect and log PII, but allow content to pass through - ‘refusal’: Block content if PII detected (hard stop) - ‘redirection’: Replace PII with anonymized placeholders (e.g., <EMAIL_ADDRESS>)

Note: Only output analysis is currently supported (target_location=’output’).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

llm_name: str | None = None#

entities: list[str] = None#

score_threshold: float = None#

class PIIDefenseMiddleware(config: PIIDefenseMiddlewareConfig, builder)#

Bases: nat.middleware.defense.defense_middleware.DefenseMiddleware

PII Defense Middleware using Microsoft Presidio.

Detects PII in function outputs using Presidio’s rule-based entity recognition. Only output analysis is currently supported (target_location='output').

See microsoft/presidio for more information about Presidio.

Streaming Behavior:: For ‘refusal’ and ‘redirection’ actions, chunks are buffered and checked before yielding to prevent PII from being streamed to clients. For ‘partial_compliance’ action, chunks are yielded immediately; violations are logged but content passes through.

Initialize defense middleware.

Args:: config: Configuration for the defense middleware builder: Builder instance for loading LLMs and other resources

config: PIIDefenseMiddlewareConfig#

_analyzer = None#

_anonymizer = None#

_lazy_load_presidio()#: Lazy load Presidio components when first needed.

_analyze_content( text: str, ) → nat.middleware.defense.defense_middleware_data_models.PIIAnalysisResult#

Analyze content for PII entities using Presidio.

Args:: text: The text to analyze
Returns:: PIIAnalysisResult with detection results and anonymized text.

_process_pii_detection( value: Any, location: str, context: nat.middleware.middleware.FunctionMiddlewareContext, ) → Any#

Process PII detection and sanitization for a given value.

This is a common helper method that handles: - Field extraction (if target_field is specified) - PII analysis - Action handling (refusal, redirection, partial_compliance) - Applying sanitized value back to original structure

Args:: value: The value to analyze (input or output) location: Either “input” or “output” (for logging) context: Function context metadata
Returns:: The value after PII handling (may be unchanged, sanitized, or raise exception)

_handle_threat( content: Any, analysis_result: nat.middleware.defense.defense_middleware_data_models.PIIAnalysisResult, context: nat.middleware.middleware.FunctionMiddlewareContext, location: str, entities_str: str, ) → Any#

Handle detected PII threat based on configured action.

Args:: content: The content with PII analysis_result: Detection result from Presidio context: Function context location: Either “input” or “output” (for logging) entities_str: String representation of detected entities
Returns:: Handled content (anonymized, original, or raises exception for refusal)

async function_middleware_invoke( *args: Any, call_next: nat.middleware.function_middleware.CallNext, context: nat.middleware.middleware.FunctionMiddlewareContext, \*\*kwargs: Any, ) → Any#

Intercept function calls to detect and anonymize PII in inputs or outputs.

Args:: args: Positional arguments passed to the function (first arg is typically the input value). call_next: Function to call the next middleware or the actual function. context: Context containing function metadata. kwargs: Keyword arguments passed to the function.
Returns:: The function result, with PII anonymized if action=’redirection’.

async function_middleware_stream( *args: Any, call_next: nat.middleware.function_middleware.CallNextStream, context: nat.middleware.middleware.FunctionMiddlewareContext, \*\*kwargs: Any, ) → collections.abc.AsyncIterator[Any]#

Intercept streaming calls to detect and anonymize PII in inputs or outputs.

For ‘refusal’ and ‘redirection’ actions: Chunks are buffered and checked before yielding. For ‘partial_compliance’ action: Chunks are yielded immediately; violations are logged.

Args:: args: Positional arguments passed to the function (first arg is typically the input value). call_next: Function to call the next middleware or the actual function. context: Context containing function metadata. kwargs: Keyword arguments passed to the function.
Yields:: The function result chunks, with PII anonymized if action=’redirection’.