nat.middleware.defense.defense_middleware#

Base Defense Middleware.

This module provides a utility base class for defense middleware with common configuration and helper methods. Each defense middleware implements its own core logic based on its specific defense strategy (LLM-based, rule-based, etc.).

Classes#

DefenseMiddlewareConfig

Base configuration for defense middleware.

DefenseMiddleware

Utility base class for defense middleware.

Module Contents#

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

Bases: nat.data_models.middleware.FunctionMiddlewareBaseConfig

Base configuration for defense middleware.

Actions use safety domain terminology: - ‘partial_compliance’: Comply with user request with warning (monitoring mode) - ‘refusal’: Refuse user request (hard refusal) - ‘redirection’: Redirect user request to a safe place; provide a safer response

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.

action: Literal['partial_compliance', 'refusal', 'redirection'] = None#
llm_wrapper_type: nat.builder.framework_enum.LLMFrameworkEnum | str = None#
target_function_or_group: str | None = None#
target_location: Literal['output'] = None#
target_field: str | None = None#
target_field_resolution_strategy: Literal['error', 'first', 'last', 'random', 'all'] = None#
class DefenseMiddleware(config: DefenseMiddlewareConfig, builder)#

Bases: nat.middleware.function_middleware.FunctionMiddleware

Utility base class for defense middleware.

This base class provides:

  • Common configuration fields (action, check_input, check_output, llm_wrapper_type)

  • Helper methods for LLM loading (for LLM-based defenses)

  • Access to builder for any resources needed

Unlike an abstract base class, this does NOT enforce a specific pattern. Each defense middleware implements its own invoke/stream logic based on its specific defense strategy:

  • LLM-based analysis (guard models, verifiers)

  • Rule-based detection (regex, signatures)

  • Heuristic-based checks

  • Statistical anomaly detection

  • etc.

Each defense owns its core logic, just like red_teaming_middleware does.

LLM Wrapper Types:

The llm_wrapper_type config field supports different framework wrappers: langchain (default) for LangChain/LangGraph-based workflows, llama_index for LlamaIndex-based workflows, crewai for CrewAI-based workflows, semantic_kernel for Semantic Kernel-based workflows, and agno, adk, strands for other supported frameworks.

Initialize defense middleware.

Args:

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

config#
builder#
_should_apply_defense(context_name: str) → bool#

Check if defense should be applied to this function based on targeting configuration.

This method mirrors the targeting logic from RedTeamingMiddleware to provide consistent behavior between attack and defense middleware.

Args:
context_name: The name of the function from context (e.g., “calculator__add”).

For workflow-level middleware, this will be “<workflow>”

Returns:

True if defense should be applied, False otherwise

Examples:

  • target=None → defends all functions and workflow

  • target=”my_calculator” → defends all functions in my_calculator group

  • target=”my_calculator__divide” → defends only the divide function

  • target=”<workflow>” or “workflow” → defends only at workflow level

async _get_llm_for_defense(
llm_name: str,
wrapper_type: nat.builder.framework_enum.LLMFrameworkEnum | str | None = None,
)#

Helper to lazy load an LLM for defense purposes.

This is a utility method for LLM-based defenses. Not all defenses will use this - some may use rule-based or other detection methods.

Args:

llm_name: Name of the LLM to load wrapper_type: Framework wrapper type (defaults to config.llm_wrapper_type if not specified)

Returns:

The loaded LLM instance with the specified framework wrapper

_resolve_multiple_field_matches(matches)#

Resolve multiple JSONPath matches based on resolution strategy.

Args:

matches: List of JSONPath match objects

Returns:

List of matches based on resolution strategy

_extract_field_from_value(value: Any) → tuple[Any, dict | None]#

Extract field(s) from value using JSONPath if target_field is specified.

Args:

value: The value to extract fields from (can be simple or complex type).

Returns:

A tuple of (content_to_analyze, field_info_dict) where content_to_analyze is the extracted field value(s) or original value if no targeting, and field_info_dict contains target_field, matches, and original_value if field was extracted, or None otherwise.

_apply_field_result_to_value(
original_value: Any,
field_info: dict,
analysis_result: Any,
) → Any#

Apply analysis result back to original value if field was extracted.

This is used when defense needs to modify the value based on field analysis. For example, if analyzing $.result and need to replace it with sanitized value.

Args:

original_value: The original complex value field_info: Field info dict from _extract_field_from_value (None if no field extraction) analysis_result: The result from defense analysis (could be sanitized value)

Returns:

Modified value with field updated, or original value if no field extraction