Security¶

async def evaluate_pre_tool(
    self,
    context: SecurityContext,
) -> SecurityVerdict:
    """Evaluate a tool invocation before execution.

    Args:
        context: The tool call's security context.

    Returns:
        A verdict: allow, deny, or escalate.
    """
    ...

async def scan_output(
    self,
    context: SecurityContext,
    output: str,
) -> OutputScanResult:
    """Scan tool output for sensitive data after execution.

    Args:
        context: The tool call's security context.
        output: The tool's output string.

    Returns:
        An ``OutputScanResult`` with findings and optional redaction.
    """
    ...

async def evaluate_pre_tool(
    self,
    context: SecurityContext,
) -> SecurityVerdict:
    """Evaluate a tool invocation before execution.

    Steps:
        1. Check disabled/DISABLED enforcement mode.
        2. Run rule engine.
        3. LLM fallback for uncertain evaluations.
        4. Autonomy augmentation (tighten only).
        5. If ESCALATE, create approval item or convert to DENY.
        6. Record audit entry.
        7. Apply shadow mode conversion if applicable.
        8. Return verdict.
    """
    if (
        not self._config.enabled
        or self._config.enforcement_mode == SecurityEnforcementMode.DISABLED
    ):
        logger.warning(SECURITY_DISABLED, tool_name=context.tool_name)
        verdict = SecurityVerdict(
            verdict=SecurityVerdictType.ALLOW,
            reason="Security subsystem disabled",
            risk_level=ApprovalRiskLevel.LOW,
            evaluated_at=datetime.now(UTC),
            evaluation_duration_ms=0.0,
        )
        if self._config.audit_enabled:
            self._record_audit(context, verdict)
        return verdict

    logger.info(
        SECURITY_EVALUATE_START,
        tool_name=context.tool_name,
        action_type=context.action_type,
        agent_id=context.agent_id,
    )

    # Always run the rule engine first -- security detectors must
    # never be bypassed, regardless of autonomy configuration.
    try:
        verdict = self._rule_engine.evaluate(context)
    except MemoryError, RecursionError:
        raise
    except Exception:
        logger.exception(
            SECURITY_INTERCEPTOR_ERROR,
            tool_name=context.tool_name,
            note="Rule engine evaluation failed (fail-closed)",
        )
        verdict = SecurityVerdict(
            verdict=SecurityVerdictType.DENY,
            reason="Rule engine evaluation failed (fail-closed)",
            risk_level=ApprovalRiskLevel.CRITICAL,
            evaluated_at=datetime.now(UTC),
            evaluation_duration_ms=0.0,
        )

    # LLM fallback for uncertain evaluations (~5% of cases).
    verdict = await self._maybe_llm_fallback(context, verdict)

    # Apply autonomy augmentation *after* the rule engine (and
    # optional LLM fallback).  Autonomy can only add stricter
    # requirements (ALLOW -> ESCALATE), never weaken a DENY or
    # ESCALATE from security detectors.
    verdict = self._apply_autonomy_augmentation(context, verdict)

    # Handle escalation.
    if verdict.verdict == SecurityVerdictType.ESCALATE:
        verdict = await self._handle_escalation(context, verdict)

    # Record audit.
    if self._config.audit_enabled:
        self._record_audit(context, verdict)

    # Shadow mode: log blocking verdicts but return ALLOW.
    # Only replace non-ALLOW verdicts to preserve legitimate
    # ALLOW reasons in the audit trail.
    if (
        self._config.enforcement_mode == SecurityEnforcementMode.SHADOW
        and verdict.verdict != SecurityVerdictType.ALLOW
    ):
        logger.warning(
            SECURITY_SHADOW_WOULD_BLOCK,
            tool_name=context.tool_name,
            original_verdict=verdict.verdict.value,
            risk_level=verdict.risk_level.value,
            reason=verdict.reason,
        )
        verdict = SecurityVerdict(
            verdict=SecurityVerdictType.ALLOW,
            reason=f"Shadow mode (original: {verdict.verdict.value})",
            risk_level=verdict.risk_level,
            confidence=verdict.confidence,
            matched_rules=verdict.matched_rules,
            evaluated_at=verdict.evaluated_at,
            evaluation_duration_ms=verdict.evaluation_duration_ms,
        )

    # Log verdict.
    event = {
        SecurityVerdictType.ALLOW: SECURITY_VERDICT_ALLOW,
        SecurityVerdictType.DENY: SECURITY_VERDICT_DENY,
        SecurityVerdictType.ESCALATE: SECURITY_VERDICT_ESCALATE,
    }.get(verdict.verdict, SECURITY_EVALUATE_COMPLETE)
    logger.info(
        event,
        tool_name=context.tool_name,
        verdict=verdict.verdict.value,
        risk_level=verdict.risk_level.value,
    )

    return verdict

async def scan_output(
    self,
    context: SecurityContext,
    output: str,
) -> OutputScanResult:
    """Scan tool output for sensitive data.

    Steps:
        1. Delegate to the output scanner.
        2. Record an audit entry if sensitive data is found.
        3. Apply the output scan response policy to transform
           the result before returning.
    """
    if not self._config.post_tool_scanning_enabled:
        logger.debug(
            SECURITY_EVALUATE_COMPLETE,
            note="output scanning disabled",
            tool_name=context.tool_name,
        )
        return OutputScanResult()

    result = self._output_scanner.scan(output)

    if result.has_sensitive_data and self._config.audit_enabled:
        entry = AuditEntry(
            id=str(uuid.uuid4()),
            timestamp=datetime.now(UTC),
            agent_id=context.agent_id,
            task_id=context.task_id,
            tool_name=context.tool_name,
            tool_category=context.tool_category,
            action_type=context.action_type,
            arguments_hash=_hash_arguments(context.arguments),
            verdict=OUTPUT_SCAN_VERDICT,
            risk_level=ApprovalRiskLevel.HIGH,
            reason=("Sensitive data in output: " + ", ".join(result.findings)),
            evaluation_duration_ms=0.0,
        )
        try:
            self._audit_log.record(entry)
        except MemoryError, RecursionError:
            raise
        except Exception:
            logger.exception(
                SECURITY_AUDIT_RECORD_ERROR,
                tool_name=context.tool_name,
                note="Output scan audit recording failed",
            )

    # Apply the output scan response policy.  On failure, fall back
    # to the raw scan result which already has scanner-level redaction
    # applied (pattern matches replaced with [REDACTED]), so the
    # fallback is still reasonably safe even if the intended policy
    # (e.g. WithholdPolicy) would have been stricter.
    policy_name = getattr(self._output_scan_policy, "name", "<unknown>")
    try:
        result = self._output_scan_policy.apply(result, context)
    except MemoryError, RecursionError:
        raise
    except Exception:
        logger.exception(
            SECURITY_INTERCEPTOR_ERROR,
            tool_name=context.tool_name,
            policy=policy_name,
            fallback_outcome=result.outcome.value,
            note="Output scan policy application failed "
            "-- returning raw scan result "
            "(may be less strict than intended policy)",
        )

    return result

def is_registered(self, action_type: str) -> bool:
    """Check if an action type is known (built-in or custom)."""
    return action_type in self._all_types

def validate(self, action_type: str) -> None:
    """Validate that an action type is registered.

    Args:
        action_type: The action type string to check.

    Raises:
        ValueError: If the action type is not registered.
    """
    if not self.is_registered(action_type):
        msg = f"Unknown action type: {action_type!r}"
        logger.warning(SECURITY_ACTION_TYPE_INVALID, error=msg)
        raise ValueError(msg)

def expand_category(self, category: str) -> frozenset[str]:
    """Expand a category prefix into all matching action types.

    Args:
        category: A category prefix (e.g. ``"code"``).

    Returns:
        All action types under that category. Returns an empty
        frozenset if the category has no built-in types (custom
        types under unknown categories are included).
    """
    builtin = _CATEGORY_MAP.get(category, frozenset())
    custom = frozenset(
        ct for ct in self._custom_types if ct.split(":")[0] == category
    )
    return builtin | custom

@staticmethod
def get_category(action_type: str) -> str:
    """Extract the category prefix from an action type string.

    Args:
        action_type: A ``category:action`` string.

    Returns:
        The category prefix before the first ``:``.

    Raises:
        ValueError: If the string does not contain ``:``.
    """
    if action_type.count(":") != 1:
        msg = (
            f"Action type {action_type!r} must use 'category:action' "
            "format (exactly one ':')"
        )
        raise ValueError(msg)
    category, action = action_type.split(":")
    if not category or not action:
        msg = (
            f"Action type {action_type!r} must have non-empty "
            "category and action parts"
        )
        raise ValueError(msg)
    return category

def all_types(self) -> frozenset[str]:
    """Return all registered action types (built-in + custom)."""
    return self._all_types

def record(self, entry: AuditEntry) -> None:
    """Append an audit entry.

    Args:
        entry: The audit entry to record.
    """
    if len(self._entries) >= self._max_entries:
        logger.warning(
            SECURITY_AUDIT_EVICTION,
            max_entries=self._max_entries,
            total_recorded=self._total_recorded,
            note="oldest entry evicted to make room",
        )
    self._entries.append(entry)
    self._total_recorded += 1
    logger.debug(
        SECURITY_AUDIT_RECORDED,
        audit_id=entry.id,
        tool_name=entry.tool_name,
        verdict=entry.verdict,
    )

def query(  # noqa: PLR0913
    self,
    *,
    agent_id: str | None = None,
    tool_name: str | None = None,
    verdict: str | None = None,
    risk_level: ApprovalRiskLevel | None = None,
    since: AwareDatetime | None = None,
    limit: int = 100,
) -> tuple[AuditEntry, ...]:
    """Query audit entries with optional filters.

    Filters are AND-combined.  Results are returned newest-first,
    up to *limit* entries.

    Args:
        agent_id: Filter by agent ID.
        tool_name: Filter by tool name.
        verdict: Filter by verdict string.
        risk_level: Filter by risk level.
        since: Entries before this datetime are excluded.
        limit: Maximum results to return (must be >= 1).

    Returns:
        Tuple of matching entries, newest first.

    Raises:
        ValueError: If *limit* < 1.
    """
    if limit < 1:
        msg = f"limit must be >= 1, got {limit}"
        logger.warning(
            SECURITY_AUDIT_CONFIG_ERROR,
            error=msg,
        )
        raise ValueError(msg)
    results: list[AuditEntry] = []
    for entry in reversed(self._entries):
        if agent_id is not None and entry.agent_id != agent_id:
            continue
        if tool_name is not None and entry.tool_name != tool_name:
            continue
        if verdict is not None and entry.verdict != verdict:
            continue
        if risk_level is not None and entry.risk_level != risk_level:
            continue
        if since is not None and entry.timestamp < since:
            continue
        results.append(entry)
        if len(results) >= limit:
            break
    return tuple(results)

def count(self) -> int:
    """Return the number of entries in the log."""
    return len(self._entries)

def scan(self, output: str) -> OutputScanResult:
    """Scan output text for sensitive patterns.

    Detection runs on the original output.  Redaction builds
    a separate redacted copy by applying substitutions in order.

    Args:
        output: The tool's output string.

    Returns:
        An ``OutputScanResult`` with findings and optional
        redacted content.
    """
    logger.debug(
        SECURITY_OUTPUT_SCAN_START,
        output_length=len(output),
    )
    findings: list[str] = []
    redacted = output

    for pattern_name, pattern in _OUTPUT_PATTERNS:
        if pattern.search(output):
            findings.append(pattern_name)
            logger.warning(
                SECURITY_OUTPUT_SCAN_FINDING,
                finding=pattern_name,
            )
            redacted = pattern.sub(_REDACTED, redacted)

    if not findings:
        return OutputScanResult()

    return OutputScanResult(
        has_sensitive_data=True,
        findings=tuple(sorted(set(findings))),
        redacted_content=redacted,
        outcome=ScanOutcome.REDACTED,
    )

def apply(
    self,
    scan_result: OutputScanResult,
    context: SecurityContext,
) -> OutputScanResult:
    """Apply the policy to a scan result.

    Args:
        scan_result: Result from the output scanner.
        context: Security context of the tool invocation.

    Returns:
        Transformed scan result.
    """
    ...

def apply(
    self,
    scan_result: OutputScanResult,
    context: SecurityContext,  # noqa: ARG002
) -> OutputScanResult:
    """Pass through the scan result unchanged.

    Args:
        scan_result: Result from the output scanner.
        context: Security context (unused).

    Returns:
        The original scan result.
    """
    logger.debug(
        SECURITY_OUTPUT_SCAN_POLICY_APPLIED,
        policy="redact",
        has_sensitive_data=scan_result.has_sensitive_data,
    )
    return scan_result

def apply(
    self,
    scan_result: OutputScanResult,
    context: SecurityContext,  # noqa: ARG002
) -> OutputScanResult:
    """Clear redacted content on sensitive results.

    Args:
        scan_result: Result from the output scanner.
        context: Security context (unused).

    Returns:
        Scan result with ``redacted_content`` cleared if sensitive.
    """
    logger.debug(
        SECURITY_OUTPUT_SCAN_POLICY_APPLIED,
        policy="withhold",
        has_sensitive_data=scan_result.has_sensitive_data,
    )
    if not scan_result.has_sensitive_data:
        return scan_result
    return scan_result.model_copy(
        update={"redacted_content": None, "outcome": ScanOutcome.WITHHELD},
    )

def apply(
    self,
    scan_result: OutputScanResult,
    context: SecurityContext,
) -> OutputScanResult:
    """Return a clean ``OutputScanResult`` regardless of findings.

    Suppresses enforcement while preserving the audit log entry
    written by ``SecOpsService.scan_output``.

    Args:
        scan_result: Result from the output scanner.
        context: Security context of the tool invocation.

    Returns:
        Clean ``OutputScanResult`` with ``has_sensitive_data=False``.
    """
    if scan_result.has_sensitive_data:
        logger.warning(
            SECURITY_OUTPUT_SCAN_POLICY_APPLIED,
            policy="log_only",
            has_sensitive_data=True,
            findings=scan_result.findings,
            tool_name=context.tool_name,
            agent_id=context.agent_id,
            note="Sensitive data detected but passed through by log_only policy",
        )
        return OutputScanResult(outcome=ScanOutcome.LOG_ONLY)
    logger.debug(
        SECURITY_OUTPUT_SCAN_POLICY_APPLIED,
        policy="log_only",
        has_sensitive_data=False,
    )
    return OutputScanResult()