Insecure Inter-Agent Communication

Description

Multi-agent systems exchange messages, delegate tasks, and share context without authentication, encryption, or integrity verification, enabling spoofing, eavesdropping, and unauthorized delegation.

Mitigations

Mutual TLS (mTLS)

Authenticate and encrypt all inter-agent communication:

• Issue unique X.509 certificates to each agent instance from an organization-controlled CA
• Require mutual TLS verification: both the sending and receiving agent must present valid certificates
• Enforce certificate pinning to prevent man-in-the-middle attacks even if a CA is compromised
• Rotate certificates on a regular schedule and revoke certificates for decommissioned agents

Message Signing

Ensure message integrity and non-repudiation:

• Sign every inter-agent message using the sending agent's private key
• Include a timestamp, nonce, and message sequence number in the signed payload to prevent replay attacks
• Receiving agents must verify the signature and reject messages with expired timestamps or reused nonces
• Maintain a message audit log with signatures for post-incident forensics

Schema Validation

Enforce strict message structure:

• Define JSON Schema or Protocol Buffer schemas for every message type exchanged between agents
• Reject messages that do not conform to the expected schema — no tolerance for unexpected fields
• Version message schemas and require both sender and receiver to agree on the version
• Treat schema violations as security events and alert on repeated violations from a specific agent

Delegation Controls

Govern task delegation between agents:

• Implement an explicit delegation policy that defines which agents can delegate to which other agents
• Delegation requests must include the delegator's identity, the task scope, and an authorization token
• Receiving agents must verify they are authorized to accept delegations from the requesting agent
• Track delegation chains to prevent circular delegation and detect anomalous delegation patterns

Channel Isolation

Separate communication channels by sensitivity:

• Use dedicated, encrypted channels for different message categories (control plane vs. data plane)
• Apply network segmentation so agents can only reach the specific agents they need to communicate with
• Implement message-level access controls: agents can only read messages addressed to them

Code Examples

Signed Message Envelope

import json
import time
import uuid
import hmac
import hashlib
from dataclasses import dataclass, asdict
from typing import Any, Optional


@dataclass
class AgentMessage:
    sender_id: str
    receiver_id: str
    message_type: str
    payload: dict[str, Any]
    timestamp: float
    nonce: str
    sequence: int
    signature: str = ""


class SecureMessageChannel:
    """Authenticated, integrity-verified inter-agent messaging."""

    REPLAY_WINDOW_SECONDS = 300  # 5-minute replay window

    def __init__(self, agent_id: str, signing_key: bytes):
        self._agent_id = agent_id
        self._signing_key = signing_key
        self._sequence = 0
        self._seen_nonces: dict[str, float] = {}  # nonce -> timestamp
        self._allowed_senders: set[str] = set()
        self._message_schemas: dict[str, dict] = {}

    def allow_sender(self, sender_id: str):
        self._allowed_senders.add(sender_id)

    def register_schema(self, message_type: str, schema: dict):
        self._message_schemas[message_type] = schema

    def create_message(
        self, receiver_id: str, message_type: str, payload: dict
    ) -> AgentMessage:
        """Create a signed message for transmission."""
        self._sequence += 1
        msg = AgentMessage(
            sender_id=self._agent_id,
            receiver_id=receiver_id,
            message_type=message_type,
            payload=payload,
            timestamp=time.time(),
            nonce=uuid.uuid4().hex,
            sequence=self._sequence,
        )
        msg.signature = self._sign(msg)
        return msg

    def verify_message(self, msg: AgentMessage, sender_key: bytes) -> dict:
        """Verify authenticity, integrity, freshness, and schema compliance."""
        checks = {
            "sender_allowed": False,
            "signature_valid": False,
            "not_replayed": False,
            "not_expired": False,
            "schema_valid": False,
            "addressed_to_us": False,
        }

        # Verify receiver
        checks["addressed_to_us"] = msg.receiver_id == self._agent_id
        if not checks["addressed_to_us"]:
            return {"valid": False, "checks": checks, "reason": "Message not addressed to this agent"}

        # Verify sender is allowed
        checks["sender_allowed"] = msg.sender_id in self._allowed_senders
        if not checks["sender_allowed"]:
            return {"valid": False, "checks": checks, "reason": f"Sender '{msg.sender_id}' not in allowed list"}

        # Verify signature
        expected_sig = self._compute_signature(msg, sender_key)
        checks["signature_valid"] = hmac.compare_digest(msg.signature, expected_sig)
        if not checks["signature_valid"]:
            return {"valid": False, "checks": checks, "reason": "Signature verification failed"}

        # Check replay (nonce uniqueness)
        if msg.nonce in self._seen_nonces:
            return {"valid": False, "checks": checks, "reason": "Replayed nonce detected"}
        checks["not_replayed"] = True

        # Check freshness
        age = time.time() - msg.timestamp
        checks["not_expired"] = age <= self.REPLAY_WINDOW_SECONDS
        if not checks["not_expired"]:
            return {"valid": False, "checks": checks, "reason": f"Message expired ({age:.0f}s old)"}

        # Schema validation
        schema = self._message_schemas.get(msg.message_type)
        if schema:
            checks["schema_valid"] = self._validate_schema(msg.payload, schema)
        else:
            checks["schema_valid"] = False
            return {"valid": False, "checks": checks, "reason": f"No schema registered for '{msg.message_type}'"}

        # Record nonce to prevent replay
        self._seen_nonces[msg.nonce] = msg.timestamp
        self._cleanup_old_nonces()

        return {"valid": True, "checks": checks}

    def _sign(self, msg: AgentMessage) -> str:
        return self._compute_signature(msg, self._signing_key)

    def _compute_signature(self, msg: AgentMessage, key: bytes) -> str:
        signable = f"{msg.sender_id}|{msg.receiver_id}|{msg.message_type}|{json.dumps(msg.payload, sort_keys=True)}|{msg.timestamp}|{msg.nonce}|{msg.sequence}"
        return hmac.new(key, signable.encode(), hashlib.sha256).hexdigest()

    def _validate_schema(self, payload: dict, schema: dict) -> bool:
        # Simplified: check required fields exist and no unexpected fields
        required = set(schema.get("required", []))
        allowed = set(schema.get("properties", {}).keys())
        present = set(payload.keys())
        return required.issubset(present) and present.issubset(allowed)

    def _cleanup_old_nonces(self):
        cutoff = time.time() - self.REPLAY_WINDOW_SECONDS * 2
        self._seen_nonces = {
            n: t for n, t in self._seen_nonces.items() if t > cutoff
        }

Evidence Requirements

• mTLS certificate inventory with issuance, rotation, and revocation records
• Message signature verification logs showing accepted and rejected messages
• Replay detection logs with captured nonce collisions
• Schema violation alerts with sender attribution
• Delegation chain audit trails mapping task flow between agents