Index

security

Security guardrails — scanners, engine wrapper, audit, SSRF.

Classes

BaseScanner

Bases: ABC

Base class for all security scanners.

Subclasses implement pattern-based scanning for secrets, PII, or other sensitive content.

Methods:

scan abstractmethod

scan(text: str) -> ScanResult

Scan text and return findings.

Source code in src/diapason/security/_stubs.py

@abstractmethod
def scan(self, text: str) -> ScanResult:
    """Scan *text* and return findings."""

redact abstractmethod

redact(text: str) -> str

Return text with sensitive matches replaced by redaction markers.

Source code in src/diapason/security/_stubs.py

@abstractmethod
def redact(self, text: str) -> str:
    """Return *text* with sensitive matches replaced by redaction markers."""

AuditLogger

AuditLogger(
    db_path: Union[str, Path] = DEFAULT_CONFIG_DIR
    / "audit.db",
    bus: Optional[EventBus] = None,
)

Append-only SQLite audit log for security events.

PARAMETER	DESCRIPTION
db_path	Path to the SQLite database file. TYPE: Union[str, Path] DEFAULT: DEFAULT_CONFIG_DIR / 'audit.db'
bus	Optional event bus — if provided, subscribes to security events (SECURITY_SCAN, SECURITY_ALERT, SECURITY_BLOCK). TYPE: Optional[EventBus] DEFAULT: None

Source code in src/diapason/security/audit.py

def __init__(
    self,
    db_path: Union[str, Path] = DEFAULT_CONFIG_DIR / "audit.db",
    bus: Optional[EventBus] = None,
) -> None:
    # `Path()` accepte tout objet exposant `__fspath__` — un `MagicMock`
    # en fait partie, et le sien rend « MagicMock/<nom>/<id> ». Un test
    # qui patchait `load_config` sans configurer `security.audit_log_path`
    # faisait donc créer, en silence, un vrai répertoire et une vraie base
    # SQLite à la racine du dépôt : 42 fichiers y ont dormi jusqu'au
    # 25 août 2026. La signature promettait `str | Path` ; elle le vérifie
    # désormais, et l'échec est bruyant plutôt qu'écrit sur le disque.
    if not isinstance(db_path, (str, Path)):
        raise TypeError(
            f"db_path doit être un str ou un Path, pas un {type(db_path).__name__}."
        )
    self._db_path = Path(db_path)
    from diapason.security.file_utils import secure_create

    secure_create(self._db_path)
    self._conn = sqlite3.connect(str(self._db_path), check_same_thread=False)
    self._lock = threading.RLock()
    self._conn.execute(
        """
        CREATE TABLE IF NOT EXISTS security_events (
            id          INTEGER PRIMARY KEY,
            timestamp   REAL,
            event_type  TEXT,
            findings_json TEXT,
            content_preview TEXT,
            action_taken TEXT,
            row_hash    TEXT DEFAULT '',
            prev_hash   TEXT DEFAULT ''
        )
        """
    )
    self._conn.commit()
    self._migrate_schema()

    if bus is not None:
        bus.subscribe(EventType.SECURITY_SCAN, self._on_event)
        bus.subscribe(EventType.SECURITY_ALERT, self._on_event)
        bus.subscribe(EventType.SECURITY_BLOCK, self._on_event)

Methods:

log

log(event: SecurityEvent) -> None

Insert a security event into the audit log with Merkle hash chain.

Source code in src/diapason/security/audit.py

def log(self, event: SecurityEvent) -> None:
    """Insert a security event into the audit log with Merkle hash chain."""
    findings_json = json.dumps(
        [
            {
                "pattern_name": f.pattern_name,
                # Audit metadata must never become a second secret store.
                "matched_text": "",
                "matched_sha256": hashlib.sha256(
                    f.matched_text.encode()
                ).hexdigest()
                if f.matched_text
                else "",
                "threat_level": f.threat_level.value,
                "start": f.start,
                "end": f.end,
                "description": f.description,
            }
            for f in event.findings
        ]
    )
    content_marker = ""
    if event.content_preview:
        content_marker = (
            "sha256:"
            f"{hashlib.sha256(event.content_preview.encode()).hexdigest()}"
            f";len:{len(event.content_preview)}"
        )

    with self._lock:
        # Compute and insert the next link atomically across worker threads.
        prev_hash = self.tail_hash()
        hash_input = (
            f"{prev_hash}|{event.timestamp}|{event.event_type.value}"
            f"|{findings_json}|{content_marker}|{event.action_taken}"
        )
        row_hash = hashlib.sha256(hash_input.encode()).hexdigest()

        self._conn.execute(
            """
            INSERT INTO security_events
                (timestamp, event_type, findings_json, content_preview,
                 action_taken, row_hash, prev_hash)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            """,
            (
                event.timestamp,
                event.event_type.value,
                findings_json,
                content_marker,
                event.action_taken,
                row_hash,
                prev_hash,
            ),
        )
        self._conn.commit()

query

query(
    *,
    event_type: Optional[str] = None,
    since: Optional[float] = None,
    limit: int = 100,
) -> List[SecurityEvent]

Query logged security events with optional filters.

Source code in src/diapason/security/audit.py

def query(
    self,
    *,
    event_type: Optional[str] = None,
    since: Optional[float] = None,
    limit: int = 100,
) -> List[SecurityEvent]:
    """Query logged security events with optional filters."""
    sql = (
        "SELECT timestamp, event_type, findings_json,"
        " content_preview, action_taken"
        " FROM security_events WHERE 1=1"
    )
    params: list = []

    if event_type is not None:
        sql += " AND event_type = ?"
        params.append(event_type)
    if since is not None:
        sql += " AND timestamp >= ?"
        params.append(since)

    sql += " ORDER BY timestamp DESC LIMIT ?"
    params.append(limit)

    with self._lock:
        rows = self._conn.execute(sql, params).fetchall()
    events: List[SecurityEvent] = []
    for row in rows:
        ts, etype, findings_json, preview, action = row
        findings_raw = json.loads(findings_json) if findings_json else []
        findings = [
            ScanFinding(
                pattern_name=f["pattern_name"],
                matched_text=f["matched_text"],
                threat_level=ThreatLevel(f["threat_level"]),
                start=f["start"],
                end=f["end"],
                description=f.get("description", ""),
            )
            for f in findings_raw
        ]
        events.append(
            SecurityEvent(
                event_type=SecurityEventType(etype),
                timestamp=ts,
                findings=findings,
                content_preview=preview or "",
                action_taken=action or "",
            )
        )
    return events

tail_hash

tail_hash() -> str

Return the hash of the last row in the chain, or empty string.

Source code in src/diapason/security/audit.py

def tail_hash(self) -> str:
    """Return the hash of the last row in the chain, or empty string."""
    with self._lock:
        row = self._conn.execute(
            "SELECT row_hash FROM security_events ORDER BY id DESC LIMIT 1"
        ).fetchone()
    return row[0] if row and row[0] else ""

verify_chain

verify_chain() -> Tuple[bool, Optional[int]]

Verify the Merkle hash chain integrity.

RETURNS	DESCRIPTION
tuple	(True, None) if the chain is valid, or (False, row_id) where row_id is the first broken link.

Source code in src/diapason/security/audit.py

def verify_chain(self) -> Tuple[bool, Optional[int]]:
    """Verify the Merkle hash chain integrity.

    Returns
    -------
    tuple
        ``(True, None)`` if the chain is valid, or
        ``(False, row_id)`` where *row_id* is the first broken link.
    """
    with self._lock:
        rows = self._conn.execute(
            "SELECT id, timestamp, event_type, findings_json,"
            " content_preview, action_taken, row_hash, prev_hash"
            " FROM security_events ORDER BY id"
        ).fetchall()

    expected_prev = ""
    for row in rows:
        rid, ts, etype, fj, preview, action, stored_hash, stored_prev = row
        # Skip rows that predate the Merkle upgrade
        if not stored_hash:
            continue
        # Verify prev_hash link
        if stored_prev != expected_prev:
            return False, rid
        # Verify row_hash
        hash_input = f"{stored_prev}|{ts}|{etype}|{fj}|{preview}|{action}"
        computed = hashlib.sha256(hash_input.encode()).hexdigest()
        if computed != stored_hash:
            return False, rid
        expected_prev = stored_hash

    return True, None

count

count() -> int

Return the total number of logged security events.

Source code in src/diapason/security/audit.py

def count(self) -> int:
    """Return the total number of logged security events."""
    with self._lock:
        row = self._conn.execute("SELECT COUNT(*) FROM security_events").fetchone()
    return row[0] if row else 0

close

close() -> None

Close the SQLite connection.

Source code in src/diapason/security/audit.py

def close(self) -> None:
    """Close the SQLite connection."""
    with self._lock:
        self._conn.close()

BoundaryGuard

BoundaryGuard(
    mode: str = "redact",
    *,
    enabled: bool = True,
    bus: Optional["EventBus"] = None,
    scanners: Optional[List["BaseScanner"]] = None,
)

Scans outbound content for secrets and PII at device boundaries.

PARAMETER	DESCRIPTION
mode	Action on findings: "redact" replaces matches, "warn" logs but passes through, "block" raises. TYPE: str DEFAULT: 'redact'
enabled	Master switch. When False, all content passes through. TYPE: bool DEFAULT: True
bus	Optional event bus for publishing SECURITY_ALERT events. TYPE: Optional['EventBus'] DEFAULT: None
scanners	Custom scanners. Defaults to SecretScanner + PIIScanner. TYPE: Optional[List['BaseScanner']] DEFAULT: None

Source code in src/diapason/security/boundary.py

def __init__(
    self,
    mode: str = "redact",
    *,
    enabled: bool = True,
    bus: Optional["EventBus"] = None,
    scanners: Optional[List["BaseScanner"]] = None,
) -> None:
    self._mode = mode
    self._enabled = enabled
    self._bus = bus
    self._scanners = scanners if scanners is not None else self._default_scanners()
    if self._enabled and not self._scanners:
        raise RuntimeError("BoundaryGuard requires at least one working scanner")

Methods:

scan_outbound

scan_outbound(content: str, destination: str) -> str

Scan text before it leaves the device.

Returns redacted text in "redact" mode, original text in "warn" mode, or raises SecurityBlockError in "block" mode when findings are detected.

Source code in src/diapason/security/boundary.py

def scan_outbound(self, content: str, destination: str) -> str:
    """Scan text before it leaves the device.

    Returns redacted text in ``"redact"`` mode, original text in
    ``"warn"`` mode, or raises ``SecurityBlockError`` in ``"block"``
    mode when findings are detected.
    """
    if not self._enabled or not content:
        return content

    has_findings = False
    redacted = content
    for scanner in self._scanners:
        result = scanner.scan(content)
        if result.findings:
            has_findings = True
            if self._mode == "redact":
                redacted = scanner.redact(redacted)

    if has_findings:
        self._emit_alert(destination, content)
        if self._mode == "block":
            raise SecurityBlockError(
                f"Secrets/PII detected in outbound content to {destination}"
            )
        if self._mode == "warn":
            logger.warning(
                "Secrets/PII detected in outbound content to %s", destination
            )
            return content
        return redacted

    return content

check_outbound

check_outbound(tool_call: ToolCall) -> ToolCall

Scan tool call arguments before execution.

Returns a new ToolCall with redacted arguments if needed.

Source code in src/diapason/security/boundary.py

def check_outbound(self, tool_call: ToolCall) -> ToolCall:
    """Scan tool call arguments before execution.

    Returns a new ToolCall with redacted arguments if needed.
    """
    if not self._enabled or not tool_call.arguments:
        return tool_call

    redacted_args = self.scan_outbound(
        tool_call.arguments, destination=f"tool:{tool_call.name}"
    )
    if redacted_args != tool_call.arguments:
        return replace(tool_call, arguments=redacted_args)
    return tool_call

redact_for_storage

redact_for_storage(content: str) -> str

Always redact secrets and PII before telemetry or trace storage.

Source code in src/diapason/security/boundary.py

def redact_for_storage(self, content: str) -> str:
    """Always redact secrets and PII before telemetry or trace storage."""
    if not self._enabled or not content:
        return content
    redacted = content
    for scanner in self._scanners:
        redacted = scanner.redact(redacted)
    return redacted

GuardrailsEngine

GuardrailsEngine(
    engine: InferenceEngine,
    *,
    scanners: Optional[List[BaseScanner]] = None,
    mode: RedactionMode = REDACT,
    scan_input: bool = True,
    scan_output: bool = True,
    bus: Optional[EventBus] = None,
)

Bases: InferenceEngine

Wraps an existing InferenceEngine with security scanning.

Not registered in EngineRegistry — instantiated dynamically to wrap any engine at runtime.

PARAMETER	DESCRIPTION
engine	The wrapped inference engine. TYPE: InferenceEngine
scanners	List of scanners to run. Defaults to SecretScanner + PIIScanner. TYPE: Optional[List[BaseScanner]] DEFAULT: None
mode	Action taken on findings: WARN, REDACT, or BLOCK. TYPE: RedactionMode DEFAULT: REDACT
scan_input	Whether to scan input messages. TYPE: bool DEFAULT: True
scan_output	Whether to scan output content. TYPE: bool DEFAULT: True
bus	Optional event bus for publishing security events. TYPE: Optional[EventBus] DEFAULT: None

Source code in src/diapason/security/guardrails.py

def __init__(
    self,
    engine: InferenceEngine,
    *,
    scanners: Optional[List[BaseScanner]] = None,
    mode: RedactionMode = RedactionMode.REDACT,
    scan_input: bool = True,
    scan_output: bool = True,
    bus: Optional[EventBus] = None,
) -> None:
    self._engine = engine
    self._scanners: List[BaseScanner] = (
        scanners
        if scanners is not None
        else [
            SecretScanner(),
            PIIScanner(),
        ]
    )
    self._mode = mode
    self._scan_input = scan_input
    self._scan_output = scan_output
    self._bus = bus

Attributes

engine_id property

engine_id: str

Delegate to the wrapped engine.

Methods:

generate

generate(
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> Dict[str, Any]

Scan input, call wrapped engine, scan output.

Source code in src/diapason/security/guardrails.py

def generate(
    self,
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> Dict[str, Any]:
    """Scan input, call wrapped engine, scan output."""
    messages = self._process_input_messages(messages)

    # Call wrapped engine
    response = self._engine.generate(
        messages,
        model=model,
        temperature=temperature,
        max_tokens=max_tokens,
        **kwargs,
    )

    # Scan output
    if self._scan_output:
        content = response.get("content", "")
        if content:
            result = self._scan_text(content)
            if not result.clean:
                response["content"] = self._handle_findings(
                    content, result, "output"
                )

    return response

stream async

stream(
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> AsyncIterator[str]

Use the same safe boundaries as rich output, including long secrets.

Source code in src/diapason/security/guardrails.py

async def stream(
    self,
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> AsyncIterator[str]:
    """Use the same safe boundaries as rich output, including long secrets."""
    messages = self._process_input_messages(messages)

    async def chunks() -> AsyncIterator[StreamChunk]:
        async with aclosing(
            self._engine.stream(
                messages,
                model=model,
                temperature=temperature,
                max_tokens=max_tokens,
                **kwargs,
            )
        ) as source:
            async for token in source:
                yield StreamChunk(content=token)

    async with aclosing(self._checked_stream(chunks())) as checked:
        async for chunk in checked:
            if chunk.content is not None:
                yield chunk.content

stream_full async

stream_full(
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> AsyncIterator[StreamChunk]

Stream verified text without releasing tools or terminal data early.

Source code in src/diapason/security/guardrails.py

async def stream_full(
    self,
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> AsyncIterator[StreamChunk]:
    """Stream verified text without releasing tools or terminal data early."""
    messages = self._process_input_messages(messages)
    source = self._engine.stream_full(
        messages,
        model=model,
        temperature=temperature,
        max_tokens=max_tokens,
        **kwargs,
    )
    async with aclosing(self._checked_stream(source)) as checked:
        async for chunk in checked:
            yield chunk

list_models

list_models() -> List[str]

Delegate to wrapped engine.

Source code in src/diapason/security/guardrails.py

def list_models(self) -> List[str]:
    """Delegate to wrapped engine."""
    return self._engine.list_models()

health

health() -> bool

Delegate to wrapped engine.

Source code in src/diapason/security/guardrails.py

def health(self) -> bool:
    """Delegate to wrapped engine."""
    return self._engine.health()

SecurityBlockError

Bases: Exception

Raised when mode is BLOCK and security findings are detected.

PIIScanner

PIIScanner()

Bases: BaseScanner

Detect personally identifiable information in text.

Source code in src/diapason/security/scanner.py

def __init__(self) -> None:
    try:
        _rust = get_rust_module()
        self._rust_impl = _rust.PIIScanner()
    except (ImportError, AttributeError, RuntimeError):
        self._rust_impl = None

Methods:

scan

scan(text: str) -> ScanResult

Scan text for PII patterns using Rust or the safe fallback.

Source code in src/diapason/security/scanner.py

def scan(self, text: str) -> ScanResult:
    """Scan *text* for PII patterns using Rust or the safe fallback."""
    if self._rust_impl is not None:
        return scan_result_from_json(self._rust_impl.scan(text))
    return _scan_python(text, self.PATTERNS)

redact

redact(text: str) -> str

Replace PII matches with [REDACTED:{pattern_name}].

Source code in src/diapason/security/scanner.py

def redact(self, text: str) -> str:
    """Replace PII matches with ``[REDACTED:{pattern_name}]``."""
    if self._rust_impl is not None:
        return self._rust_impl.redact(text)
    return _redact_python(text, self.PATTERNS)

SecretScanner

SecretScanner()

Bases: BaseScanner

Detect API keys, tokens, passwords, and other secrets in text.

Source code in src/diapason/security/scanner.py

def __init__(self) -> None:
    try:
        _rust = get_rust_module()
        self._rust_impl = _rust.SecretScanner()
    except (ImportError, AttributeError, RuntimeError):
        # Security must not disappear merely because the optional native
        # accelerator is unavailable.  The Python implementation below is
        # intentionally feature-equivalent, just slower.
        self._rust_impl = None

Methods:

scan

scan(text: str) -> ScanResult

Scan text for secret patterns using Rust or the safe fallback.

Source code in src/diapason/security/scanner.py

def scan(self, text: str) -> ScanResult:
    """Scan *text* for secret patterns using Rust or the safe fallback."""
    if self._rust_impl is not None:
        return scan_result_from_json(self._rust_impl.scan(text))
    return _scan_python(text, self.PATTERNS)

redact

redact(text: str) -> str

Replace secret matches with [REDACTED:{pattern_name}].

Source code in src/diapason/security/scanner.py

def redact(self, text: str) -> str:
    """Replace secret matches with ``[REDACTED:{pattern_name}]``."""
    if self._rust_impl is not None:
        return self._rust_impl.redact(text)
    return _redact_python(text, self.PATTERNS)

RedactionMode

Bases: str, Enum

Action mode when findings are detected.

ScanFinding dataclass

ScanFinding(
    pattern_name: str,
    matched_text: str,
    threat_level: ThreatLevel,
    start: int,
    end: int,
    description: str = "",
)

A single finding from a security scanner.

ScanResult dataclass

ScanResult(findings: List[ScanFinding] = list())

Aggregated result from one or more scanners.

Attributes

clean property

clean: bool

Return True if no findings were detected.

highest_threat property

highest_threat: Optional[ThreatLevel]

Return the highest threat level among findings, or None.

SecurityEvent dataclass

SecurityEvent(
    event_type: SecurityEventType,
    timestamp: float,
    findings: List[ScanFinding] = list(),
    content_preview: str = "",
    action_taken: str = "",
)

A recorded security event for audit logging.

SecurityEventType

Bases: str, Enum

Categories of security events.

ThreatLevel

Bases: str, Enum

Severity classification for security findings.

SecurityContext dataclass

SecurityContext(
    engine: Any,
    capability_policy: Any = None,
    audit_logger: Any = None,
    boundary_guard: Any = None,
    rate_limiter: Any = None,
)

Result of setup_security() — wrapped engine, policy, audit.

Functions:

filter_sensitive_paths

filter_sensitive_paths(
    paths: Iterable[Union[str, Path]],
) -> List[Path]

Return only non-sensitive paths from paths.

Source code in src/diapason/security/file_policy.py

def filter_sensitive_paths(paths: Iterable[Union[str, Path]]) -> List[Path]:
    """Return only non-sensitive paths from *paths*."""
    return [Path(p) for p in paths if not is_sensitive_file(p)]

is_sensitive_file

is_sensitive_file(path: Union[str, Path]) -> bool

Return True if path matches a sensitive file pattern.

Checks both the filename and the full name against DEFAULT_SENSITIVE_PATTERNS using :func:fnmatch.fnmatch. Uses the Rust implementation when available, falls back to Python.

Source code in src/diapason/security/file_policy.py

def is_sensitive_file(path: Union[str, Path]) -> bool:
    """Return ``True`` if *path* matches a sensitive file pattern.

    Checks both the filename and the full name against
    ``DEFAULT_SENSITIVE_PATTERNS`` using :func:`fnmatch.fnmatch`.
    Uses the Rust implementation when available, falls back to Python.
    """
    try:
        from diapason._rust_bridge import get_rust_module

        _rust = get_rust_module()
        return _rust.is_sensitive_file(str(path))
    except ImportError:
        return _is_sensitive_file_py(str(path))

check_ssrf

check_ssrf(url: str) -> Optional[str]

Check a URL for SSRF vulnerabilities.

Prefers the Rust backend, but falls back to the pure-Python implementation when the compiled extension is unavailable. The SSRF guard is security-critical, so it must never be silently skipped — or crash with ImportError — merely because Rust was not built.

Source code in src/diapason/security/ssrf.py

def check_ssrf(url: str) -> Optional[str]:
    """Check a URL for SSRF vulnerabilities.

    Prefers the Rust backend, but falls back to the pure-Python
    implementation when the compiled extension is unavailable. The SSRF
    guard is security-critical, so it must never be silently skipped — or
    crash with ``ImportError`` — merely because Rust was not built.
    """
    from diapason._rust_bridge import RUST_AVAILABLE, get_rust_module

    if RUST_AVAILABLE:
        return get_rust_module().check_ssrf(url)
    return _check_ssrf_python(url)

is_private_ip

is_private_ip(ip_str: str) -> bool

Check if an IP address is private/reserved.

Source code in src/diapason/security/ssrf.py

def is_private_ip(ip_str: str) -> bool:
    """Check if an IP address is private/reserved."""
    try:
        addr = ipaddress.ip_address(ip_str)
    except ValueError:
        return False
    # Normalize IPv4-mapped / IPv4-compatible IPv6 to the embedded IPv4 so
    # the IPv4 private-range CIDRs apply. Without this, e.g. ::ffff:127.0.0.1
    # bypasses the loopback / RFC1918 checks.
    if isinstance(addr, ipaddress.IPv6Address):
        embedded = _embedded_ipv4(addr)
        if embedded is not None:
            addr = embedded
    return any(addr in net for net in _BLOCKED_CIDR)

setup_security

setup_security(
    config: Any, engine: Any, bus: Optional[EventBus] = None
) -> SecurityContext

Apply security guardrails to an engine based on config.

Returns a SecurityContext. No-ops if config.security.enabled is False.

Source code in src/diapason/security/__init__.py

def setup_security(
    config: Any,
    engine: Any,
    bus: Optional[EventBus] = None,
) -> SecurityContext:
    """Apply security guardrails to an engine based on config.

    Returns a SecurityContext. No-ops if config.security.enabled is False.
    """
    if not config.security.enabled:
        return SecurityContext(engine=engine)

    scanners: list[BaseScanner] = []
    if config.security.secret_scanner:
        scanners.append(SecretScanner())
    if config.security.pii_scanner:
        scanners.append(PIIScanner())
    if (config.security.scan_input or config.security.scan_output) and not scanners:
        raise RuntimeError("Security scanning is enabled but no scanner is configured")

    mode = RedactionMode(config.security.mode)
    if scanners:
        engine = GuardrailsEngine(
            engine,
            scanners=scanners,
            mode=mode,
            scan_input=config.security.scan_input,
            scan_output=config.security.scan_output,
            bus=bus,
        )

    boundary_guard = BoundaryGuard(
        mode=config.security.mode,
        enabled=True,
        bus=bus,
        scanners=scanners,
    )

    from diapason.security.rate_limiter import RateLimitConfig, RateLimiter

    rate_limiter = RateLimiter(
        RateLimitConfig(
            requests_per_minute=config.security.rate_limit_rpm,
            burst_size=config.security.rate_limit_burst,
            enabled=config.security.rate_limit_enabled,
        )
    )

    # Capability policy
    cap_policy = None
    if config.security.capabilities.enabled:
        from diapason.security.capabilities import CapabilityPolicy

        cap_policy = CapabilityPolicy(
            policy_path=config.security.capabilities.policy_path or None,
            default_deny=config.security.capabilities.default_deny,
        )

    # Audit logger
    audit = AuditLogger(
        db_path=config.security.audit_log_path,
        bus=bus,
    )

    return SecurityContext(
        engine=engine,
        capability_policy=cap_policy,
        audit_logger=audit,
        boundary_guard=boundary_guard,
        rate_limiter=rate_limiter,
    )