ai-agent-security – skill Agent & Tự động hóa cho agent AI

AI Agent Security

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

These are partial design sketches, not a working security boundary. Regexes/classifiers have false positives and false negatives; prompt text cannot grant permissions. Enforce tenant identity, resource scope and approved actions in the tool service using authenticated server-side context. Input signing authenticates origin, not the safety of signed content. Bind approvals to the exact operation/arguments and recheck immediately before execution. Test failure paths and concurrency before production use.

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 = not matched  # heuristic result only; never authorization

    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 not result.is_safe:
                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}"
                )

            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

Isolate untrusted code in a sandbox appropriate to the threat model. Containers share the host kernel and do not replace tool-service authorization. Do not mount the Docker socket or host credentials; run as a non-root user and restrict mounts and egress. NET_BIND_SERVICE permits low-port binding, not network access.

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
    tmpfs:
      - /tmp:size=64M,noexec,nosuid
    mem_limit: 512m
    cpus: "0.5"
    pids_limit: 64
    networks:
      - sandbox-net
    environment:
      - TOOL_TIMEOUT=30 # application must enforce this; env vars alone do nothing
      - 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
# Illustrative fragment only: merge into a backed-up daemon.json, validate,
# and schedule the disruptive Docker restart. Do not overwrite existing settings.
cat <<'EOF' > runsc-daemon-fragment.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 --network=none --cap-drop=ALL --security-opt=no-new-privileges --user=65534:65534 \
  --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 Sketch (Incomplete)

This sketch fails closed for approval-gated operations but is not a production gatekeeper.

… (nội dung đầy đủ có trong gói tải về)