🤖 Agentic AI Security

Multi-Agent Security Architecture Guide

When you chain AI agents together, new security problems emerge. Inter-agent authentication, trust boundaries, blast radius limitation, and audit logging for multi-agent SOC systems.

🎯 Multi-Agent Threat Model

The core problem: In a single-agent system, you have one trust boundary between human and AI. In a multi-agent system, you have N trust boundaries between every agent pair. Each boundary is an attack surface. A compromised downstream agent can poison the entire pipeline.

Agent Impersonation

A malicious actor or compromised service impersonates a legitimate agent, injecting false data into the pipeline. Agent B has no way to verify that the message claiming to be from Agent A actually came from Agent A.

Prompt Injection via Agent Output

Agent A fetches external content (web page, email, document) containing a prompt injection payload. Agent A's output — containing the injected instructions — is passed to Agent B, which executes the attacker's commands instead of the legitimate task.

Trust Chain Escalation

Agent A is given limited permissions. It delegates to Agent B with instructions to "use all available tools." Agent B, trusted by the orchestrator, executes high-privilege actions that Agent A was never authorised to take directly.

State Manipulation

Shared memory or context passed between agents can be modified in transit. An attacker who controls the message bus or shared state store can alter Agent A's conclusions before Agent B reads them, changing the pipeline's decisions.

Cascading False Positives

Agent A makes an incorrect determination. Agent B takes that as ground truth and escalates. Agent C takes Agent B's escalation as confirmation. A single error becomes an automated false-positive incident chain that triggers real containment actions.

Audit Gap — Who Decided?

In a multi-agent pipeline, it becomes impossible to determine which agent made which decision that led to an action. Forensic investigation of agent-caused incidents becomes difficult without structured audit trails per agent.

📊 Attack Surface by Agent Topology

Topology	Description	Primary Attack Surfaces	Risk Level
Sequential pipeline	A → B → C — each agent processes and passes to next	Poisoned output from any agent propagates forward. No feedback loop to catch errors.	Medium
Orchestrator + sub-agents	Central orchestrator delegates to specialist agents	Orchestrator compromise controls all sub-agents. Sub-agents trusted by orchestrator may be over-privileged.	High
Peer-to-peer mesh	Agents communicate directly with each other	Every agent is an attack surface. No central control point. Circular trust loops possible.	Very High
Human + agent collaboration	Human in the loop with AI agents handling subtasks	Prompt injection in agent output presented to human. Human may trust agent output uncritically.	Medium
Hierarchical with sub-orchestrators	Orchestrator → sub-orchestrators → workers	Trust hierarchy complexity. Privilege escalation through sub-orchestrator manipulation.	High

🔐 Inter-Agent Authentication

Each agent-to-agent message must be authenticated. Agent B must be able to verify that a message claiming to come from Agent A actually did, and that the message was not modified in transit.

Option 1 — JWT-based agent identity (recommended for most SOC systems)

import jwt, time, hashlib, json from cryptography.hazmat.primitives import serialization from cryptography.hazmat.backends import default_backend # Each agent has a unique key pair generated at deployment # Private key stored in secrets manager (AWS SSM, Vault, etc.) # Public key registered in agent registry AGENT_REGISTRY = { "triage-agent-v1": "-----BEGIN PUBLIC KEY-----\n...\n-----END PUBLIC KEY-----", "enrichment-agent-v1":"-----BEGIN PUBLIC KEY-----\n...\n-----END PUBLIC KEY-----", "response-agent-v1": "-----BEGIN PUBLIC KEY-----\n...\n-----END PUBLIC KEY-----", } def sign_agent_message(payload: dict, agent_id: str, private_key_pem: str) -> str: """Sign an inter-agent message with the sending agent's private key.""" private_key = serialization.load_pem_private_key( private_key_pem.encode(), password=None, backend=default_backend() ) token = jwt.encode( { **payload, "iss": agent_id, # Issuer — sending agent ID "iat": int(time.time()), # Issued at "exp": int(time.time()) + 60, # Short TTL — 60 seconds "jti": hashlib.sha256( # Unique token ID — prevents replay f"{agent_id}{time.time()}".encode() ).hexdigest()[:16], }, private_key, algorithm="RS256" ) return token def verify_agent_message(token: str, expected_sender: str) -> dict: """Verify a message from another agent. Raises on failure.""" public_key_pem = AGENT_REGISTRY.get(expected_sender) if not public_key_pem: raise ValueError(f"Unknown agent: {expected_sender}") payload = jwt.decode( token, public_key_pem, algorithms=["RS256"], options={"require": ["iss", "iat", "exp", "jti"]} ) # Verify the token was issued by who we expect if payload["iss"] != expected_sender: raise ValueError(f"Token issuer mismatch: expected {expected_sender}, got {payload['iss']}") return payload # Usage in receiving agent def receive_from_triage_agent(token: str): try: data = verify_agent_message(token, expected_sender="triage-agent-v1") # Now safe to use data — authenticity verified return process_triage_result(data) except jwt.ExpiredSignatureError: raise ValueError("Message from triage agent expired — possible replay attack") except jwt.InvalidSignatureError: raise ValueError("Invalid signature from triage agent — possible impersonation")

Option 2 — HMAC message signing (simpler, for internal pipelines)

import hmac, hashlib, json, time, os # Shared secret per agent pair — stored in secrets manager # Different secret for each agent-pair relationship AGENT_SECRETS = { ("triage-agent", "enrichment-agent"): os.environ["TRIAGE_TO_ENRICH_SECRET"], ("enrichment-agent", "response-agent"): os.environ["ENRICH_TO_RESPONSE_SECRET"], } def sign_message(payload: dict, sender: str, receiver: str) -> dict: secret = AGENT_SECRETS.get((sender, receiver)) if not secret: raise ValueError(f"No shared secret for {sender} → {receiver}") message = json.dumps(payload, sort_keys=True) timestamp = str(int(time.time())) sig_input = f"{timestamp}.{message}" signature = hmac.new(secret.encode(), sig_input.encode(), hashlib.sha256).hexdigest() return { "payload": payload, "sender": sender, "timestamp": timestamp, "signature": signature } def verify_message(signed_msg: dict, expected_sender: str, receiver: str) -> dict: secret = AGENT_SECRETS.get((expected_sender, receiver)) if not secret: raise ValueError(f"No shared secret for {expected_sender} → {receiver}") # Check timestamp — reject messages older than 30 seconds msg_time = int(signed_msg["timestamp"]) if abs(time.time() - msg_time) > 30: raise ValueError("Message timestamp out of range — possible replay attack") message = json.dumps(signed_msg["payload"], sort_keys=True) sig_input = f"{signed_msg['timestamp']}.{message}" expected_sig = hmac.new(secret.encode(), sig_input.encode(), hashlib.sha256).hexdigest() if not hmac.compare_digest(expected_sig, signed_msg["signature"]): raise ValueError("HMAC signature verification failed") return signed_msg["payload"]

🏰 Trust Boundaries & Least Privilege

Every agent should have only the permissions it needs for its specific role. A triage agent that can also execute containment actions has unnecessary blast radius. Define a capability set per agent and enforce it at the tool layer.

Agent capability matrix — SOC pipeline example

Agent Role	Allowed Tools	Forbidden Tools	Can Delegate To
Triage Agent	query_siem, get_alert_details, get_asset_info	ALL write/action tools	Enrichment Agent only
Enrichment Agent	virustotal_lookup, whois, shodan, get_threat_intel	ALL write/action tools	None — outputs only
Correlation Agent	query_siem, get_historical_alerts, get_user_activity	ALL write/action tools	None — outputs only
Response Agent	block_ip, quarantine_host (HITL gated), create_ticket	account_disable, data_delete, cert_in_notify	None
Reporting Agent	create_ticket, send_email (draft only), update_wiki	ALL security action tools	None
Orchestrator	delegate_to_agent, read_agent_output, manage_workflow	Direct tool execution (delegates only)	All sub-agents

Enforcing capability boundaries in code

from functools import wraps # Define allowed tools per agent role AGENT_CAPABILITIES = { "triage": {"query_siem", "get_alert_details", "get_asset_info"}, "enrichment": {"virustotal_lookup", "whois_lookup", "shodan_query", "get_threat_intel"}, "response": {"block_ip", "quarantine_host", "create_ticket"}, "orchestrator": {"delegate_to_agent", "read_agent_output"}, } def enforce_capabilities(agent_role: str): """Decorator — raises if agent tries to call a tool outside its capability set.""" def decorator(tool_func): @wraps(tool_func) def wrapper(*args, **kwargs): tool_name = tool_func.__name__ allowed = AGENT_CAPABILITIES.get(agent_role, set()) if tool_name not in allowed: # Log the attempt — potential capability escalation security_log.warning( f"CAPABILITY VIOLATION: {agent_role} attempted {tool_name}", extra={"agent": agent_role, "tool": tool_name, "args": str(args)[:200]} ) raise PermissionError( f"Agent '{agent_role}' is not permitted to call '{tool_name}'. " f"Allowed tools: {allowed}" ) return tool_func(*args, **kwargs) return wrapper return decorator # Apply to each tool @enforce_capabilities("triage") def query_siem(query: str): ... @enforce_capabilities("triage") def block_ip(ip: str): ... # This will raise — triage agent cannot block IPs # Enforce delegation limits — orchestrator cannot delegate to arbitrary agents DELEGATION_MAP = { "orchestrator": {"triage", "enrichment", "correlation"}, "triage": set(), # Cannot delegate "enrichment": set(), # Cannot delegate "response": set(), # Cannot delegate } def validate_delegation(from_agent: str, to_agent: str): allowed = DELEGATION_MAP.get(from_agent, set()) if to_agent not in allowed: raise PermissionError( f"Agent '{from_agent}' cannot delegate to '{to_agent}'" )

📝 Audit Trail for Multi-Agent Pipelines

Every agent action must be logged with enough context to reconstruct exactly what happened and why. Without structured audit trails, investigating an agent-caused incident is nearly impossible.

Required fields per audit log entry

Field	Type	Purpose	Example
`trace_id`	UUID	Links all log entries from a single pipeline run	a3f7-4b2c-9e1d
`span_id`	UUID	Unique ID for this specific agent action	7e21-8f3a-1c9b
`parent_span_id`	UUID	ID of the action that triggered this one	3d41-2a5c-8f7e
`agent_id`	String	Which agent took the action	triage-agent-v1
`tool_called`	String	What tool was invoked	query_siem
`tool_args`	JSON	Parameters passed to the tool	{"query": "EventID=4625..."}
`tool_result_hash`	SHA256	Hash of result (not raw result — may be sensitive)	3f2a9b...
`model_reasoning`	String	Why the agent decided to call this tool	"Checking failed logons for..."
`confidence`	Float	Model's self-reported confidence	0.87
`hitl_required`	Bool	Was human approval required for this action	true
`approved_by`	String	Analyst who approved (if HITL)	analyst@org.com
`timestamp_utc`	ISO8601	When the action occurred	2025-07-15T09:32:11Z

import uuid, hashlib, json, logging from datetime import datetime, timezone from contextvars import ContextVar # Thread-local trace ID — propagated across entire pipeline run current_trace_id: ContextVar[str] = ContextVar('trace_id', default=None) current_span_id: ContextVar[str] = ContextVar('span_id', default=None) class AgentAuditLogger: def __init__(self, agent_id: str): self.agent_id = agent_id self.logger = logging.getLogger(f"agent.audit.{agent_id}") def log_tool_call(self, tool: str, args: dict, result, reasoning: str = "", confidence: float = 0.0, approved_by: str = None): span_id = str(uuid.uuid4())[:8] parent_span = current_span_id.get() current_span_id.set(span_id) # Hash result to avoid logging sensitive data result_str = json.dumps(result, default=str) if not isinstance(result, str) else result result_hash = hashlib.sha256(result_str.encode()).hexdigest()[:16] entry = { "event_type": "agent_tool_call", "trace_id": current_trace_id.get() or str(uuid.uuid4())[:8], "span_id": span_id, "parent_span_id": parent_span, "agent_id": self.agent_id, "tool_called": tool, "tool_args": {k: str(v)[:200] for k, v in args.items()}, # Truncate long args "tool_result_hash": result_hash, "model_reasoning": reasoning[:500], "confidence": round(confidence, 3), "hitl_required": approved_by is not None or self._is_high_risk(tool), "approved_by": approved_by, "timestamp_utc": datetime.now(timezone.utc).isoformat(), } self.logger.info(json.dumps(entry)) return span_id def _is_high_risk(self, tool: str) -> bool: HIGH_RISK_TOOLS = {"block_ip", "quarantine_host", "disable_account", "delete_file", "send_cert_in_notification"} return tool in HIGH_RISK_TOOLS # Usage — inject logger into each agent audit = AgentAuditLogger("triage-agent-v1") audit.log_tool_call("query_siem", {"query": "EventID=4625"}, result, reasoning="Checking for brute force pattern", confidence=0.92)

✅ Secure Multi-Agent Architecture Patterns

🔒 Sanitise All Inter-Agent Messages

Before any agent processes output from another agent or external source, strip content that could be interpreted as instructions. Use a dedicated sanitisation layer between agents. Never pass raw external content directly as agent input.

🏝️ Agent Isolation — Separate Processes

Run each agent in a separate process or container. A compromised agent cannot directly read another agent's memory or environment variables. Use message queues (Redis, SQS, RabbitMQ) for inter-agent communication — not shared memory.

⏱️ Per-Agent Rate Limiting

Limit how many tool calls each agent can make per minute and per pipeline run. A runaway or compromised agent performing thousands of SIEM queries or containment actions is a denial-of-service risk. Hard limits per agent per action type.

🎯 Immutable Pipeline Definitions

Agent orchestration logic — which agents can talk to which, with what permissions — should be defined at deploy time and immutable at runtime. Agents should not be able to modify their own capability set or create new delegation relationships.

🔄 Independent Verification for High-Risk

For high-consequence decisions, have two independent agents analyse the same evidence separately, then compare conclusions before acting. Disagreement → escalate to human. This is the multi-agent equivalent of a two-person integrity check.

💀 Dead Man's Switch

If an agent fails to heartbeat within its expected interval, automatically suspend all downstream agents in the same pipeline. A silently-failed agent that stops sending outputs is preferable to a compromised agent sending poisoned outputs.

Minimum viable secure multi-agent setup

""" Minimum viable secure multi-agent SOC pipeline. Demonstrates: auth, capability enforcement, audit logging, HITL gates. """ import uuid from contextvars import ContextVar trace_id: ContextVar = ContextVar('trace_id') class SecureAgentPipeline: def __init__(self): self.agents = {} self.message_queue = [] def register_agent(self, agent_id: str, role: str, allowed_tools: set): self.agents[agent_id] = { "role": role, "allowed_tools": allowed_tools, "audit_log": AgentAuditLogger(agent_id) } def send_message(self, from_agent: str, to_agent: str, payload: dict): """Send authenticated message between agents.""" if to_agent not in self.agents: raise ValueError(f"Unknown agent: {to_agent}") # Sign the message signed = sign_message(payload, sender=from_agent, receiver=to_agent) # Sanitise before delivery — strip instruction-like content sanitised_payload = self._sanitise(signed["payload"]) # Log the inter-agent communication self.agents[from_agent]["audit_log"].log_tool_call( tool=f"send_to_{to_agent}", args={"to": to_agent, "payload_keys": list(payload.keys())}, result={"status": "sent"}, reasoning="Inter-agent delegation" ) # Deliver to receiving agent return self.agents[to_agent]["handler"](from_agent, sanitised_payload) def _sanitise(self, payload: dict) -> dict: """Strip potential prompt injection from inter-agent messages.""" INJECTION_PATTERNS = [ "ignore previous instructions", "you are now", "system: ", "assistant: ", "<|im_start|>", ] sanitised = {} for k, v in payload.items(): if isinstance(v, str): v_lower = v.lower() for pattern in INJECTION_PATTERNS: if pattern in v_lower: v = f"[SANITISED: potential injection detected in field '{k}']" break sanitised[k] = v return sanitised # Initialise pipeline pipeline = SecureAgentPipeline() pipeline.register_agent( "triage-agent-v1", role="triage", allowed_tools={"query_siem", "get_alert_details", "get_asset_info"} ) pipeline.register_agent( "enrichment-agent-v1", role="enrichment", allowed_tools={"virustotal_lookup", "whois_lookup", "shodan_query"} ) pipeline.register_agent( "response-agent-v1", role="response", allowed_tools={"block_ip", "quarantine_host", "create_ticket"} )