MCP Server Message Queue Security: Replay Attacks, Dead-Letter Queue Poisoning, Consumer Authentication, and Idempotent Tool Execution

Why MCP servers use message queues

The synchronous MCP transport model — client sends a tool call, server executes it, server returns the result — breaks at scale in three ways. First, long-running tools (file ingestion, LLM sub-calls, external API calls with multi-second latencies) block the HTTP connection and the agent waiting on the other end. Second, a synchronous server that crashes mid-execution loses the in-flight tool call entirely — the client sees a 502 and retries, potentially executing the tool twice. Third, fan-out patterns — where one agent action triggers tool calls across multiple downstream services — require the caller to wait for all of them serially unless you add parallel coordination.

Message queues solve all three: the initial tool call acknowledgment is immediate, retries happen at the broker layer rather than requiring client-side retry logic, and fan-out becomes a simple multi-topic publish. But they move the execution window out of the synchronous request-response cycle, which breaks the security model that assumes tool execution is bounded by the authenticated HTTP request.

1. Broker authentication: preventing unauthorized tool injection

The first and most fundamental question: who is allowed to publish messages to the MCP tool execution queue? In many internal deployments, the answer is "any service on the internal network" — which is the same as "any attacker who compromises a single internal service can inject arbitrary tool calls as any tenant."

// DANGEROUS: no auth, open broker connection — any internal service can publish tool calls
const kafka = new Kafka({ brokers: ['kafka:9092'] });

// CORRECT: SASL/SCRAM with TLS
const kafka = new Kafka({
  brokers: ['kafka:9092'],
  ssl: true,
  sasl: {
    mechanism: 'scram-sha-256',
    username: process.env.KAFKA_USERNAME,
    password: process.env.KAFKA_PASSWORD,
  },
});

// ACL: only the MCP gateway service principal can produce to mcp-tool-calls topic
// kafka-acls.sh --add --allow-principal User:mcp-gateway \
//   --operation Write --topic mcp-tool-calls

2. Replay attacks: re-executing tool calls after initial success

Message brokers guarantee at-least-once delivery by default — meaning a message may be delivered more than once, especially after consumer crashes, network partitions, or explicit redelivery requests. For tool calls that read data (readFile, listRecords), duplicate delivery is harmless — the second execution returns the same result. For tool calls with side effects, a replay is a second execution of the side effect:

• A sendEmail tool call replayed 3× sends 3 emails
• A chargeCard tool call replayed 2× charges the card twice
• A provisionVirtualMachine tool call replayed creates two VMs
• A postWebhook tool call replayed delivers the webhook twice — if the receiver isn't idempotent, this causes double-processing on their end

The replay may happen legitimately (broker redelivery after consumer crash) or maliciously (attacker captures a message off the wire or from a leaked DLQ and re-publishes it with the same payload).

Idempotency key pattern for exactly-once tool execution

The solution is an idempotency key stored in a shared deduplication table. Every tool call message carries a globally unique idempotency key. Before executing the side effect, the consumer attempts to reserve the key with an INSERT ON CONFLICT DO NOTHING. If the insert succeeds (rowCount = 1), this is the first execution — proceed with the side effect. If it fails (rowCount = 0), this message has already been processed — skip the side effect, optionally return the cached result.

-- Deduplication table
CREATE TABLE IF NOT EXISTS tool_idempotency_keys (
  key         TEXT PRIMARY KEY,
  tool_name   TEXT NOT NULL,
  tenant_id   TEXT NOT NULL,
  status      TEXT NOT NULL DEFAULT 'processing',  -- processing | done | failed
  result      JSONB,
  created_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
  expires_at  TIMESTAMPTZ NOT NULL DEFAULT now() + INTERVAL '7 days'
);

CREATE INDEX ON tool_idempotency_keys (expires_at);  -- for TTL cleanup

async function processToolCallMessage(message) {
  const { idempotencyKey, toolName, tenantId, args } = message;

  // Step 1: reserve the key before executing any side effect
  const { rowCount } = await db.query(
    `INSERT INTO tool_idempotency_keys (key, tool_name, tenant_id)
     VALUES ($1, $2, $3)
     ON CONFLICT (key) DO NOTHING`,
    [idempotencyKey, toolName, tenantId]
  );

  if (rowCount === 0) {
    // Already processed (or currently processing by another consumer)
    const existing = await db.query(
      'SELECT status, result FROM tool_idempotency_keys WHERE key = $1',
      [idempotencyKey]
    );
    if (existing.rows[0].status === 'done') {
      // Safe to ack — we already have the result
      return existing.rows[0].result;
    }
    // Still 'processing' from another consumer — nack and let broker redeliver
    // after the processing lock expires or the other consumer commits
    throw new Error('Concurrent processing — nack for redeliver');
  }

  // Step 2: execute the side effect (only runs if we reserved the key)
  let result;
  try {
    result = await executeTool(toolName, args, tenantId);
  } catch (err) {
    // Mark failed so retries are allowed
    await db.query(
      `UPDATE tool_idempotency_keys SET status = 'failed' WHERE key = $1`,
      [idempotencyKey]
    );
    throw err;
  }

  // Step 3: mark done and cache the result
  await db.query(
    `UPDATE tool_idempotency_keys SET status = 'done', result = $1 WHERE key = $2`,
    [JSON.stringify(result), idempotencyKey]
  );

  return result;
}

3. Dead-letter queue poisoning

A dead-letter queue (DLQ) is where messages go when the main consumer fails to process them after the configured retry count. DLQs serve a legitimate purpose: they prevent a poison-pill message (one that always throws an exception) from blocking the entire queue indefinitely. But they introduce a security vulnerability when:

1. The DLQ consumer runs with different (often higher) permissions than the main consumer
2. The DLQ consumer is a human-reviewed replay interface operated by an engineer with production access
3. The DLQ message payload is rendered in a UI for inspection without sanitization

An attacker who can publish to the tool-calls queue can craft a message that reliably fails processing — for example, by including a tool argument that triggers a validation exception in the consumer — and then waits for that message to reach the DLQ. If the DLQ consumer re-processes messages with elevated permissions ("replay this failed job as admin"), the attacker has successfully escalated through the error path.

// DANGEROUS: DLQ consumer re-executes with elevated system permissions
const dlqConsumer = kafka.consumer({ groupId: 'dlq-replay-admin' });
dlqConsumer.subscribe({ topic: 'mcp-tool-calls-dlq' });
dlqConsumer.run({
  eachMessage: async ({ message }) => {
    const payload = JSON.parse(message.value);
    // Re-runs with system-level service account (more privileged than original caller)
    await executeTool(payload.toolName, payload.args, 'system');  // WRONG
  }
});

// CORRECT: DLQ consumer enforces the original tenant/permission context
dlqConsumer.run({
  eachMessage: async ({ message }) => {
    const payload = JSON.parse(message.value);
    // Validate the idempotency key and original tenant before re-executing
    if (!payload.idempotencyKey || !payload.tenantId || !payload.originalPermissions) {
      // Malformed or tampered DLQ message — quarantine, do not execute
      await quarantineMessage(message);
      return;
    }
    // Re-execute with ORIGINAL permissions only — no elevation
    await executeTool(payload.toolName, payload.args, payload.tenantId, {
      permissions: payload.originalPermissions,
    });
  }
});

The full DLQ hardening strategy:

• Preserve original context: embed the original tenantId, permissions, and sessionId in the message at publish time. DLQ re-processing must use these, never elevated credentials.
• Message signing: HMAC-sign the message payload at publish time with a producer-side secret. DLQ consumers verify the signature before re-execution — tampered payloads (with elevated permission fields injected) fail verification.
• Schema validation on DLQ entry: validate the message against the expected tool call schema before adding to DLQ. A message that fails schema validation is not a failed tool call — it is malformed input. Route to a quarantine store, not the DLQ.
• DLQ access controls: the DLQ topic should have stricter ACLs than the main topic — only the DLQ consumer service principal and designated operations accounts should read it.

4. Consumer group isolation for multi-tenant MCP deployments

In Kafka, multiple consumers in the same consumer group share partition assignments — each partition is consumed by exactly one group member. Consumer groups with different group IDs each get their own independent copy of messages (all groups see all messages). This fan-out model creates a cross-tenant leakage risk if all tenants' tool calls are on the same topic:

// DANGEROUS: all tenants' tool calls on one topic, consumed by a shared group
// Tenant A's tool call for their private data reaches Tenant B's consumer
// if partition assignment puts them on the same consumer instance

// CORRECT Option 1: per-tenant topics with ACLs
// Topic naming: mcp-tool-calls-tenant-{tenantId}
// Each tenant's consumer group has ACLs only on their own topic

// CORRECT Option 2: single topic with per-message tenant validation
// Consumer validates tenantId in message against the authenticated session
// before executing — rejects messages where tenantId doesn't match the
// consumer's assigned tenant scope
async function validateTenantScope(message, consumerTenantId) {
  const payload = JSON.parse(message.value);
  if (payload.tenantId !== consumerTenantId) {
    // Cross-tenant message — could be routing error or injection attempt
    await quarantineMessage(message, 'cross-tenant-routing');
    throw new Error(`Message tenant ${payload.tenantId} !== consumer scope ${consumerTenantId}`);
  }
  return payload;
}

For SQS, tenant isolation is straightforward: each tenant gets their own SQS queue, and the IAM policy on each queue grants access only to that tenant's service account. The queue URL itself is a capability — knowing the queue URL isn't sufficient without the IAM credentials that authorize sqs:ReceiveMessage.

5. Message schema validation

Consumers that trust the structure of inbound messages without validation are vulnerable to prototype pollution, type confusion, and injection attacks via unexpected field values. Message schema validation at the consumer boundary — before any processing — provides the same protection that HTTP request validation provides at the API boundary.

import { z } from 'zod';

const ToolCallMessageSchema = z.object({
  idempotencyKey: z.string().uuid(),
  toolName: z.string().min(1).max(100).regex(/^[a-zA-Z][a-zA-Z0-9_]*$/),
  tenantId: z.string().uuid(),
  sessionId: z.string().uuid(),
  args: z.record(z.unknown()),  // further validated per-tool
  originalPermissions: z.array(z.string()).max(50),
  publishedAt: z.string().datetime(),
  signature: z.string().min(64),  // HMAC-SHA256 hex
}).strict();  // .strict() rejects extra fields — prevents __proto__ injection

async function consumeMessage(rawMessage) {
  let payload;
  try {
    payload = ToolCallMessageSchema.parse(JSON.parse(rawMessage.value));
  } catch (err) {
    // Invalid schema — quarantine immediately, do not process
    await quarantineMessage(rawMessage, 'schema-validation-failed', err.errors);
    return;
  }

  // Verify HMAC signature before processing
  const expectedSig = computeHmac(payload, process.env.MESSAGE_SIGNING_SECRET);
  if (!timingSafeEqual(Buffer.from(payload.signature, 'hex'), expectedSig)) {
    await quarantineMessage(rawMessage, 'signature-mismatch');
    return;
  }

  await processToolCallMessage(payload);
}

6. Transport encryption and at-rest protection for tool arguments

Tool call messages often contain arguments that are sensitive: file paths, search queries that reveal user intent, API parameters that encode PII. Two protection layers are needed:

In-flight encryption (TLS): All broker connections should use TLS — this is non-negotiable for any production deployment. For Kafka, set ssl: true in client config. For RabbitMQ, use the amqps:// scheme. For SQS, the HTTPS endpoint is the only endpoint; there is no unencrypted alternative. For self-hosted brokers, provision certificates from an internal CA and rotate them on the same schedule as your TLS certificates elsewhere.

At-rest encryption for sensitive argument fields: Broker-level disk encryption protects against physical storage compromise but not against anyone with broker read access. For tool arguments that contain secrets (passwords, API keys passed as tool args — which is an anti-pattern but common), encrypt the sensitive fields before publishing using envelope encryption: generate a per-message data encryption key (DEK), encrypt the sensitive fields with the DEK using AES-256-GCM, encrypt the DEK with a key encryption key (KEK) stored in your KMS (AWS KMS, HashiCorp Vault), and store the encrypted DEK alongside the encrypted field in the message payload. The consumer decrypts the DEK from KMS, decrypts the field locally.

// Envelope encryption for sensitive tool args
import { KMSClient, GenerateDataKeyCommand, DecryptCommand } from '@aws-sdk/client-kms';
import { createCipheriv, createDecipheriv, randomBytes } from 'crypto';

const kms = new KMSClient({ region: 'us-east-1' });
const KEY_ID = process.env.KMS_KEY_ARN;

async function encryptSensitiveArgs(args) {
  const { Plaintext: dek, CiphertextBlob: encryptedDek } =
    await kms.send(new GenerateDataKeyCommand({ KeyId: KEY_ID, KeySpec: 'AES_256' }));

  const iv = randomBytes(12);
  const cipher = createCipheriv('aes-256-gcm', dek, iv);
  const encrypted = Buffer.concat([cipher.update(JSON.stringify(args)), cipher.final()]);
  const authTag = cipher.getAuthTag();

  return {
    encryptedArgs: encrypted.toString('base64'),
    encryptedDek: Buffer.from(encryptedDek).toString('base64'),
    iv: iv.toString('base64'),
    authTag: authTag.toString('base64'),
  };
}

async function decryptSensitiveArgs(encryptedPayload) {
  const { Plaintext: dek } = await kms.send(new DecryptCommand({
    KeyId: KEY_ID,
    CiphertextBlob: Buffer.from(encryptedPayload.encryptedDek, 'base64'),
  }));
  const decipher = createDecipheriv(
    'aes-256-gcm',
    dek,
    Buffer.from(encryptedPayload.iv, 'base64')
  );
  decipher.setAuthTag(Buffer.from(encryptedPayload.authTag, 'base64'));
  return JSON.parse(
    Buffer.concat([
      decipher.update(Buffer.from(encryptedPayload.encryptedArgs, 'base64')),
      decipher.final()
    ]).toString()
  );
}

Delivery semantics: what "at-least-once" means for tool security

Deployment checklist

• Broker requires authentication on all connections (SASL/SCRAM or mTLS); no unauthenticated plaintext connections permitted in production
• Topic-level ACLs restrict produce/consume permissions to named service principals; no wildcard grants
• All broker connections use TLS (ssl: true / amqps:// / HTTPS-only endpoint)
• Every side-effectful tool call message carries a caller-generated idempotency key (UUID) in the message body
• Consumers check idempotency key against deduplication table before executing any side effect
• Message payload is HMAC-signed by the producer; consumers verify signature before processing
• Zod / JSON Schema validation with .strict() on every inbound message before processing
• DLQ consumer re-executes with original tenant permissions, never elevated credentials
• DLQ messages are schema-validated before DLQ entry; tampered or malformed messages go to quarantine
• DLQ operations UI sanitizes all message fields before rendering (treat as untrusted HTML)
• Per-tenant topic ACLs or per-message tenant validation prevents cross-tenant message delivery
• Sensitive tool argument fields encrypted with envelope encryption (DEK + KMS) before publishing

SkillAudit findings for message queue vulnerabilities in MCP servers

See also: Distributed locking and race conditions · OAuth security · CSRF security · Zero-trust architecture

Run a free SkillAudit on your MCP server →