ai-agent-security

Protect agentic AI systems from adversarial input, unsafe tool execution, data leakage, and privilege abuse with layered security controls.

When to Use This Skill

Use this skill when:

• Building AI agents that invoke tools, APIs, or shell commands
• Deploying agents with access to production databases, cloud accounts, or internal services
• Hardening multi-tenant agent platforms against cross-tenant data leakage
• Adding guardrails to autonomous coding agents or SRE bots
• Designing approval workflows for high-risk agent actions
• Conducting red-team exercises against agentic systems
• Responding to incidents involving compromised or misbehaving agents

Prerequisites

• Python 3.10+ for guardrail code examples
• Docker or Podman for sandbox execution
• OpenTelemetry collector for audit logging
• Familiarity with your agent framework (LangChain, CrewAI, Autogen, custom)
• Access to policy engine (OPA/Cedar) for permission boundaries

Threat Model — STRIDE for AI Agents

AI agents introduce a unique threat surface. Apply STRIDE specifically to agentic components:

| Threat | Agent-Specific Example | Control | |--------|----------------------|---------| | Spoofing | Attacker crafts input that mimics a trusted internal tool response | Signed tool responses, HMAC verification | | Tampering | Prompt injection modifies agent reasoning mid-chain | Input validation, prompt armoring | | Repudiation | Agent takes destructive action with no audit trail | Immutable structured logging | | Information Disclosure | Agent leaks PII, secrets, or internal architecture in responses | Output filtering, content classifiers | | Denial of Service | Adversarial prompt causes infinite tool loops or token exhaustion | Rate limits, token budgets, circuit breakers | | Elevation of Privilege | Agent escalates from read-only to write via chained tool calls | RBAC per tool, least-privilege scoping |

Key Threat Categories

Prompt Injection — Untrusted content (user input, web scrapes, document contents) manipulates the agent's system prompt or reasoning chain to execute unintended actions.

Tool Abuse — The agent calls tools in sequences or with parameters the designer did not anticipate, achieving effects beyond its intended scope.

Data Exfiltration — The agent encodes sensitive data (credentials, PII, internal IPs) into its responses, tool calls, or outbound HTTP requests.

Cross-Tenant Leakage — In multi-tenant deployments, context from one tenant's session bleeds into another through shared memory, vector stores, or cache.

Privilege Escalation — The agent chains low-privilege tool calls to achieve high-privilege outcomes (e.g., read config -> extract credentials -> call admin API).

Input Validation

Every input to an agent must be sanitized before it reaches the model or any tool. This includes user messages, tool outputs being fed back, and retrieved documents.

Prompt Injection Detection

import re
from dataclasses import dataclass
from enum import Enum

class RiskLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class ValidationResult:
    is_safe: bool
    risk_level: RiskLevel
    matched_rules: list[str]
    sanitized_input: str

INJECTION_PATTERNS = [
    (r"ignore\s+(all\s+)?(previous|prior|above)\s+(instructions|prompts|rules)", "instruction_override"),
    (r"you\s+are\s+now\s+(a|an|the)\s+", "role_hijack"),
    (r"system\s*:\s*", "system_prompt_inject"),
    (r"<\|?(system|im_start|endoftext)\|?>", "control_token_inject"),
    (r"\[INST\]|\[\/INST\]|<<SYS>>", "template_inject"),
    (r"(?:execute|run|eval)\s*\(", "code_execution_attempt"),
    (r"(?:curl|wget|nc|ncat)\s+", "network_command_inject"),
    (r"(?:rm\s+-rf|mkfs|dd\s+if=|chmod\s+777)", "destructive_command"),
    (r"(?:\/etc\/passwd|\/etc\/shadow|\.env\b|\.ssh\/)", "path_traversal"),
    (r"(?:BEGIN\s+(?:RSA|DSA|EC)\s+PRIVATE\s+KEY)", "secret_exfil_attempt"),
]

def validate_agent_input(user_input: str, max_length: int = 4096) -> ValidationResult:
    """Validate and sanitize input before passing to agent."""
    matched = []
    risk = RiskLevel.LOW

    # Length check
    if len(user_input) > max_length:
        matched.append("input_too_long")
        risk = RiskLevel.MEDIUM

    # Null byte and control character removal
    sanitized = user_input.replace("\x00", "")
    sanitized = re.sub(r"[\x01-\x08\x0b\x0c\x0e-\x1f]", "", sanitized)

    # Pattern matching
    for pattern, rule_name in INJECTION_PATTERNS:
        if re.search(pattern, sanitized, re.IGNORECASE):
            matched.append(rule_name)
            risk = RiskLevel.HIGH

    # Stacked injection detection (multiple suspicious patterns)
    if len(matched) >= 3:
        risk = RiskLevel.CRITICAL

    is_safe = risk in (RiskLevel.LOW, RiskLevel.MEDIUM)

    return ValidationResult(
        is_safe=is_safe,
        risk_level=risk,
        matched_rules=matched,
        sanitized_input=sanitized[:max_length] if is_safe else "",
    )

Content Classification Middleware

Use a lightweight classifier as middleware before the agent processes any input:

from functools import wraps
from typing import Callable

def input_guard(validator: Callable = validate_agent_input):
    """Decorator that guards agent entry points against unsafe input."""
    def decorator(func):
        @wraps(func)
        async def wrapper(user_input: str, *args, **kwargs):
            result = validator(user_input)

            if result.risk_level == RiskLevel.CRITICAL:
                await log_security_event(
                    event="input_blocked",
                    risk=result.risk_level.value,
                    rules=result.matched_rules,
                    input_hash=hashlib.sha256(user_input.encode()).hexdigest(),
                )
                raise InputRejectedError(
                    f"Input blocked: matched {result.matched_rules}"
                )

            if result.risk_level == RiskLevel.HIGH:
                await log_security_event(
                    event="input_flagged",
                    risk=result.risk_level.value,
                    rules=result.matched_rules,
                )
                # Allow through but flag for review
                kwargs["_security_flags"] = result.matched_rules

            return await func(result.sanitized_input, *args, **kwargs)
        return wrapper
    return decorator

# Usage
@input_guard()
async def handle_user_message(message: str, session_id: str, **kwargs):
    """Process a validated user message through the agent."""
    flags = kwargs.get("_security_flags", [])
    if flags:
        # Route to sandboxed execution path
        return await agent.run_sandboxed(message, session_id)
    return await agent.run(message, session_id)

Tool Execution Sandboxing

Never let an agent execute tools directly on the host. Isolate every tool invocation inside a sandbox.

Docker Sandbox Configuration

# docker-compose.agent-sandbox.yml
version: "3.8"

services:
  agent-sandbox:
    image: agent-tools:latest
    read_only: true
    security_opt:
      - no-new-privileges:true
      - seccomp:seccomp-profile.json
    cap_drop:
      - ALL
    cap_add:
      - NET_BIND_SERVICE   # Only if tool needs network
    tmpfs:
      - /tmp:size=64M,noexec,nosuid
    mem_limit: 512m
    cpus: "0.5"
    pids_limit: 64
    networks:
      - sandbox-net
    environment:
      - TOOL_TIMEOUT=30
      - MAX_OUTPUT_BYTES=65536
    volumes:
      - type: bind
        source: ./tool-workspace
        target: /workspace
        read_only: false
    dns:
      - 127.0.0.1           # Block external DNS by default

networks:
  sandbox-net:
    driver: bridge
    internal: true           # No external network access

gVisor Runtime for Stronger Isolation

# Install gVisor runsc runtime
curl -fsSL https://gvisor.dev/archive.key | sudo gpg --dearmor -o /usr/share/keyrings/gvisor-archive-keyring.gpg
echo "deb [signed-by=/usr/share/keyrings/gvisor-archive-keyring.gpg] https://storage.googleapis.com/gvisor/releases release main" | \
  sudo tee /etc/apt/sources.list.d/gvisor.list
sudo apt-get update && sudo apt-get install -y runsc

# Configure Docker to use gVisor
cat <<'EOF' | sudo tee /etc/docker/daemon.json
{
  "runtimes": {
    "runsc": {
      "path": "/usr/bin/runsc",
      "runtimeArgs": [
        "--network=none",
        "--directfs=false"
      ]
    }
  }
}
EOF
sudo systemctl restart docker

# Run agent sandbox with gVisor
docker run --runtime=runsc --rm \
  --read-only \
  --memory=512m \
  --cpus=0.5 \
  --pids-limit=64 \
  agent-tools:latest \
  python /tools/execute.py --tool="$TOOL_NAME" --args="$TOOL_ARGS"

Tool Allowlist Enforcement

from dataclasses import dataclass, field

@dataclass
class ToolPolicy:
    name: str
    allowed_args: dict[str, type]     # parameter name -> expected type
    max_calls_per_session: int = 10
    requires_approval: bool = False
    allowed_patterns: list[str] = field(default_factory=list)
    blocked_patterns: list[str] = field(default_factory=list)

TOOL_ALLOWLIST: dict[str, ToolPolicy] = {
    "read_file": ToolPolicy(
        name="read_file",
        allowed_args={"path": str},
        max_calls_per_session=20,
        allowed_patterns=[r"^/workspace/", r"^/data/public/"],
        blocked_patterns=[r"\.env$", r"\.key$", r"\.pem$", r"/etc/", r"/proc/"],
    ),
    "run_query": ToolPolicy(
        name="run_query",
        allowed_args={"sql": str, "database": str},
        max_calls_per_session=5,
        allowed_patterns=[r"^SELECT\s", r"^EXPLAIN\s"],
        blocked_patterns=[r"\bDROP\b", r"\bDELETE\b", r"\bUPDATE\b", r"\bINSERT\b", r"\bALTER\b"],
    ),
    "http_request": ToolPolicy(
        name="http_request",
        allowed_args={"url": str, "method": str},
        max_calls_per_session=10,
        requires_approval=True,
        allowed_patterns=[r"^https://api\.internal\."],
        blocked_patterns=[r"^https?://169\.254\.", r"^https?://metadata\.google\."],
    ),
    "execute_code": ToolPolicy(
        name="execute_code",
        allowed_args={"code": str, "language": str},
        max_calls_per_session=3,
        requires_approval=True,
        blocked_patterns=[r"import\s+subprocess", r"import\s+os", r"__import__", r"eval\(", r"exec\("],
    ),
}

class ToolGatekeeper:
    def __init__(self, allowlist: dict[str, ToolPolicy]):
        self.allowlist = allowlist
        self.call_counts: dict[str, int] = {}

    async def authorize(self, tool_name: str, args: dict) -> bool:
        if tool_name not in self.allowlist:
            await log_security_event(
                event="tool_denied_not_in_allowlist",
                tool=tool_name,
            )
            return False

        policy = self.allowlist[tool_name]

        # Check call count
        count = self.call_counts.get(tool_name, 0)
        if count >= policy.max_calls_per_session:
            await log_security_event(
                event="tool_denied_rate_limit",
                tool=tool_name,
                count=count,
            )
            return False

        # Validate argument types
        for arg_name, expected_type in policy.allowed_args.items():
            if arg_name in args and not isinstance(args[arg_name], expected_type):
                return False

        # Check patterns against all string arguments
        for arg_value in args.values():
            if not isinstance(arg_value, str):
                continue
            # Must match at least one allowed pattern (if any defined)
            if policy.allowed_patterns:
                if not any(re.search(p, arg_value, re.IGNORECASE) for p in policy.allowed_patterns):
                    return False
            # Must not match any blocked pattern
            if any(re.search(p, arg_value, re.IGNORECASE) for p in policy.blocked_patterns):
                await log_security_event(
                    event="tool_denied_blocked_pattern",
                    tool=tool_name,
                    arg_value_hash=hashlib.sha256(arg_value.encode()).hexdigest(),
                )
                return False

        self.call_counts[tool_name] = count + 1
        return True

Permission Boundaries

Enforce least-privilege at every layer: model context, tool access, infrastructure credentials.

RBAC Policy for Agent Tools (OPA Rego)

# policy/agent_tool_access.rego
package agent.tool_access

default allow = false

# Role definitions
roles := {
    "reader": {"read_file", "run_query", "search"},
    "writer": {"read_file", "run_query", "search", "write_file", "create_ticket"},
    "operator": {"read_file", "run_query", "search", "write_file", "create_ticket",
                  "restart_service", "scale_deployment"},
    "admin": {"read_file", "run_query", "search", "write_file", "create_ticket",
              "restart_service", "scale_deployment", "execute_code", "manage_secrets"},
}

# Allow if the agent's role includes the requested tool
allow {
    role := input.agent_role
    tool := input.tool_name
    roles[role][tool]
}

# Deny any tool call outside business hours for operator/admin roles
deny_outside_hours {
    input.agent_role == "operator"
    hour := time.clock(time.now_ns())[0]
    hour < 6
}

deny_outside_hours {
    input.agent_role == "operator"
    hour := time.clock(time.now_ns())[0]
    hour > 22
}

allow {
    not deny_outside_hours
    role := input.agent_role
    tool := input.tool_name
    roles[role][tool]
}

# High-risk tools always require human approval
requires_approval {
    high_risk := {"execute_code", "manage_secrets", "restart_service", "scale_deployment"}
    high_risk[input.tool_name]
}

Querying the Policy at Runtime

import httpx

OPA_URL = "http://localhost:8181/v1/data/agent/tool_access"

async def check_tool_permission(agent_role: str, tool_name: str, context: dict) -> dict:
    """Query OPA for tool access decision."""
    payload = {
        "input": {
            "agent_role": agent_role,
            "tool_name": tool_name,
            "session_id": context.get("session_id"),
            "tenant_id": context.get("tenant_id"),
        }
    }
    async with httpx.AsyncClient(timeout=2.0) as client:
        resp = await client.post(OPA_URL, json=payload)
        resp.raise_for_status()
        result = resp.json().get("result", {})
    return {
        "allowed": result.get("allow", False),
        "requires_approval": result.get("requires_approval", False),
    }

Scoped Credentials with Short TTLs

# vault-agent-policy.hcl — Vault policy for AI agent credentials
path "secret/data/agent/{{identity.entity.aliases.auth_approle.metadata.tenant_id}}/*" {
  capabilities = ["read"]
}

# Agent tokens expire in 15 minutes, cannot be renewed beyond 1 hour
path "auth/token/create" {
  capabilities = ["update"]
  allowed_parameters = {
    "ttl"       = ["15m"]
    "max_ttl"   = ["1h"]
    "policies"  = ["agent-readonly"]
    "no_parent" = ["true"]
  }
}

# Issue a short-lived agent credential
vault token create \
  -policy=agent-readonly \
  -ttl=15m \
  -explicit-max-ttl=1h \
  -metadata="agent_session=$SESSION_ID" \
  -metadata="tenant=$TENANT_ID" \
  -no-parent

Output Filtering

Every agent response must be scanned before delivery to the user or downstream system.

PII Detection and Redaction

import re
from typing import NamedTuple

class PIIMatch(NamedTuple):
    pii_type: str
    start: int
    end: int

PII_PATTERNS = {
    "ssn": r"\b\d{3}-\d{2}-\d{4}\b",
    "credit_card": r"\b(?:\d{4}[\s-]?){3}\d{4}\b",
    "email": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",
    "phone_us": r"\b(?:\+1[\s.-]?)?\(?\d{3}\)?[\s.-]?\d{3}[\s.-]?\d{4}\b",
    "aws_key": r"\bAKIA[0-9A-Z]{16}\b",
    "private_key": r"-----BEGIN (?:RSA |EC |DSA )?PRIVATE KEY-----",
    "jwt": r"\beyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\b",
    "ipv4_internal": r"\b(?:10\.\d{1,3}\.\d{1,3}\.\d{1,3}|172\.(?:1[6-9]|2\d|3[01])\.\d{1,3}\.\d{1,3}|192\.168\.\d{1,3}\.\d{1,3})\b",
    "connection_string": r"(?:mongodb|postgres|mysql|redis):\/\/[^\s\"']+",
}

def scan_for_pii(text: str) -> list[PIIMatch]:
    """Scan text for PII and secrets."""
    matches = []
    for pii_type, pattern in PII_PATTERNS.items():
        for m in re.finditer(pattern, text, re.IGNORECASE):
            matches.append(PIIMatch(pii_type, m.start(), m.end()))
    return matches

def redact_output(text: str) -> tuple[str, list[PIIMatch]]:
    """Redact PII from agent output. Returns redacted text and match list."""
    matches = scan_for_pii(text)
    if not matches:
        return text, []

    # Sort by position descending so replacements don't shift indices
    sorted_matches = sorted(matches, key=lambda m: m.start, reverse=True)
    redacted = text
    for match in sorted_matches:
        placeholder = f"[REDACTED_{match.pii_type.upper()}]"
        redacted = redacted[:match.start] + placeholder + redacted[match.end:]

    return redacted, matches

Response Validation Middleware

@dataclass
class OutputPolicy:
    max_length: int = 16384
    block_on_pii: bool = True
    block_on_secrets: bool = True
    allowed_domains: list[str] = field(default_factory=lambda: [
        "docs.example.com", "api.example.com"
    ])

async def validate_agent_output(
    response: str,
    policy: OutputPolicy,
    session_id: str,
) -> str:
    """Validate and filter agent output before returning to user."""
    # Length check
    if len(response) > policy.max_length:
        response = response[:policy.max_length] + "\n\n[Output truncated]"

    # PII/secret scan
    redacted, matches = redact_output(response)
    if matches:
        secret_types = {m.pii_type for m in matches}
        await log_security_event(
            event="output_pii_detected",
            session_id=session_id,
            pii_types=list(secret_types),
            count=len(matches),
        )
        if policy.block_on_secrets and secret_types & {"aws_key", "private_key", "jwt", "connection_string"}:
            return "[Response blocked: contained credentials. This incident has been logged.]"
        if policy.block_on_pii:
            return redacted

    # URL allowlist check — block responses that contain links to unapproved domains
    urls = re.findall(r"https?://([^/\s\"']+)", response)
    for domain in urls:
        if not any(domain.endswith(allowed) for allowed in policy.allowed_domains):
            response = re.sub(
                rf"https?://{re.escape(domain)}[^\s\"']*",
                "[URL_REMOVED]",
                response,
            )

    return response

Audit Logging

Every agent action must produce a structured, immutable log entry. Use OpenTelemetry for distributed tracing across agent chains.

Structured Event Logger

import json
import time
import hashlib
from datetime import datetime, timezone

class AgentAuditLogger:
    def __init__(self, service_name: str = "agent-platform"):
        self.service_name = service_name

    def log_event(self, event: dict) -> str:
        """Emit a structured audit log entry. Returns the event ID."""
        event_id = hashlib.sha256(
            f"{time.time_ns()}-{json.dumps(event, sort_keys=True)}".encode()
        ).hexdigest()[:16]

        record = {
            "event_id": event_id,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "service": self.service_name,
            **event,
        }

        # Emit as structured JSON line (ship to SIEM via Fluent Bit / Vector)
        print(json.dumps(record, default=str), flush=True)
        return event_id

    def log_tool_call(self, session_id: str, tool: str, args: dict,
                      result_status: str, duration_ms: float, agent_role: str):
        return self.log_event({
            "event_type": "tool_call",
            "session_id": session_id,
            "tool": tool,
            "args_hash": hashlib.sha256(json.dumps(args, sort_keys=True).encode()).hexdigest(),
            "result_status": result_status,
            "duration_ms": round(duration_ms, 2),
            "agent_role": agent_role,
        })

    def log_input_validation(self, session_id: str, risk_level: str,
                             matched_rules: list[str]):
        return self.log_event({
            "event_type": "input_validation",
            "session_id": session_id,
            "risk_level": risk_level,
            "matched_rules": matched_rules,
        })

    def log_output_filter(self, session_id: str, pii_types: list[str],
                          action_taken: str):
        return self.log_event({
            "event_type": "output_filter",
            "session_id": session_id,
            "pii_types_detected": pii_types,
            "action": action_taken,
        })

OpenTelemetry Spans for Agent Traces

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource

# Initialize tracer
resource = Resource.create({"service.name": "agent-platform"})
provider = TracerProvider(resource=resource)
exporter = OTLPSpanExporter(endpoint="http://otel-collector:4317", insecure=True)
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("agent.security")

async def traced_tool_call(tool_name: str, args: dict, session_id: str):
    """Execute a tool call with full OpenTelemetry tracing."""
    with tracer.start_as_current_span(
        f"tool.{tool_name}",
        attributes={
            "agent.session_id": session_id,
            "agent.tool.name": tool_name,
            "agent.tool.args_keys": ",".join(args.keys()),
        },
    ) as span:
        try:
            result = await execute_tool(tool_name, args)
            span.set_attribute("agent.tool.status", "success")
            span.set_attribute("agent.tool.output_length", len(str(result)))
            return result
        except Exception as e:
            span.set_attribute("agent.tool.status", "error")
            span.set_attribute("agent.tool.error", str(e)[:256])
            span.record_exception(e)
            raise

OpenTelemetry Collector Config

# otel-collector-config.yaml
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  batch:
    timeout: 5s
    send_batch_size: 256
  attributes:
    actions:
      - key: agent.session_id
        action: upsert
      - key: agent.tool.args_raw   # Never log raw tool args
        action: delete

exporters:
  otlp/jaeger:
    endpoint: jaeger:4317
    tls:
      insecure: true
  loki:
    endpoint: http://loki:3100/loki/api/v1/push
    labels:
      resource:
        service.name: "service_name"
      attributes:
        agent.tool.name: "tool_name"
        agent.tool.status: "tool_status"

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [batch, attributes]
      exporters: [otlp/jaeger]
    logs:
      receivers: [otlp]
      processors: [batch, attributes]
      exporters: [loki]

Rate Limiting and Abuse Prevention

Prevent runaway agents and adversarial users from exhausting resources.

Toke

<!-- Content truncated for initial SEO render. Open the source file tab for the full file. -->