Kinesis Security for MCP Servers

IAM least-privilege for producers and consumers

Kinesis IAM permissions follow the standard AWS pattern of resource ARN scoping plus condition keys. The most useful condition key is kinesis:StreamARN — it restricts a permission to a specific stream ARN, preventing a compromised producer role from publishing to a different stream in the same account.

// CDK — separate producer and consumer IAM roles with least-privilege
import { PolicyStatement, Effect, Role, ServicePrincipal } from "aws-cdk-lib/aws-iam";
import { Stream } from "aws-cdk-lib/aws-kinesis";

const stream = new Stream(this, "McpEventStream", {
  streamName: "mcp-tool-events",
  shardCount: 2,
  encryption: StreamEncryption.KMS,
  encryptionKey: cmk,  // customer-managed CMK (see below)
  retentionPeriod: Duration.days(7),
});

// Producer role — MCP server Lambda/ECS task
const producerRole = new Role(this, "McpProducerRole", {
  assumedBy: new ServicePrincipal("lambda.amazonaws.com"),
});
producerRole.addToPolicy(new PolicyStatement({
  effect: Effect.ALLOW,
  actions: ["kinesis:PutRecord", "kinesis:PutRecords"],
  resources: [stream.streamArn],
}));
// KMS permission to encrypt records before PutRecord
cmk.grantEncrypt(producerRole);

// Consumer role — analytics Lambda / KCL worker
const consumerRole = new Role(this, "McpConsumerRole", {
  assumedBy: new ServicePrincipal("lambda.amazonaws.com"),
});
consumerRole.addToPolicy(new PolicyStatement({
  effect: Effect.ALLOW,
  actions: [
    "kinesis:GetRecords",
    "kinesis:GetShardIterator",
    "kinesis:DescribeStream",
    "kinesis:DescribeStreamSummary",
    "kinesis:ListShards",
    "kinesis:ListStreams",
  ],
  resources: [stream.streamArn],
}));
// Enhanced fan-out: add SubscribeToShard + RegisterStreamConsumer
consumerRole.addToPolicy(new PolicyStatement({
  effect: Effect.ALLOW,
  actions: [
    "kinesis:SubscribeToShard",
    "kinesis:RegisterStreamConsumer",
    "kinesis:DescribeStreamConsumer",
  ],
  resources: [
    stream.streamArn,
    // Enhanced fan-out consumer ARN is a child resource of the stream
    `${stream.streamArn}/consumer/*`,
  ],
}));
// KMS permission to decrypt records on GetRecords
cmk.grantDecrypt(consumerRole);

KMS server-side encryption — CMK vs aws/kinesis

Kinesis Data Streams supports server-side encryption using either the AWS managed key (aws/kinesis) or a customer-managed CMK. The managed key is the easier option but has a critical limitation: it cannot be used for cross-account access because other accounts cannot reference your account's managed key. If any consumer of your stream runs in a different AWS account — a separate analytics account, a partner integration, a security tooling account — you must use a CMK.

A second difference: with a CMK you get KMS CloudTrail events for every Encrypt and Decrypt call, giving you record-level access audit. With aws/kinesis you get no KMS-level audit trail — only Kinesis data events if you enable CloudTrail data events for Kinesis separately.

// CDK — customer-managed CMK for Kinesis encryption with cross-account access
import { Key, KeySpec, KeyUsage } from "aws-cdk-lib/aws-kms";
import { AccountPrincipal, ArnPrincipal, PolicyStatement, Effect } from "aws-cdk-lib/aws-iam";

const cmk = new Key(this, "McpKinesisKey", {
  description: "CMK for mcp-tool-events Kinesis stream encryption",
  keySpec: KeySpec.SYMMETRIC_DEFAULT,
  keyUsage: KeyUsage.ENCRYPT_DECRYPT,
  enableKeyRotation: true,
  policy: new PolicyDocument({
    statements: [
      // Admin — key management (not data access)
      new PolicyStatement({
        effect: Effect.ALLOW,
        principals: [new AccountPrincipal(this.account)],
        actions: ["kms:*"],
        resources: ["*"],
      }),
      // Cross-account consumer in analytics account — decrypt only
      new PolicyStatement({
        effect: Effect.ALLOW,
        principals: [new ArnPrincipal("arn:aws:iam::444455556666:role/AnalyticsConsumerRole")],
        actions: ["kms:Decrypt", "kms:GenerateDataKey"],
        resources: ["*"],
        conditions: {
          StringEquals: {
            "kms:ViaService": `kinesis.${this.region}.amazonaws.com`,
            "kms:CallerAccount": "444455556666",
          },
        },
      }),
    ],
  }),
});

// Stream with CMK encryption
const stream = new Stream(this, "McpEventStream", {
  streamName: "mcp-tool-events",
  shardCount: 2,
  encryption: StreamEncryption.KMS,
  encryptionKey: cmk,
});

VPC interface endpoints — keeping Kinesis traffic off the public internet

By default, SDK calls to Kinesis resolve to public endpoints and route through the public internet even when made from a Lambda inside a VPC. To keep Kinesis API traffic within AWS's network fabric, create a VPC interface endpoint for Kinesis. Unlike S3 and DynamoDB gateway endpoints (free), Kinesis uses an interface endpoint (PrivateLink) that costs $0.01/endpoint-hour per AZ.

# Create VPC interface endpoint for Kinesis Data Streams
aws ec2 create-vpc-endpoint \
  --vpc-id vpc-0abc12345 \
  --service-name com.amazonaws.us-east-1.kinesis-streams \
  --vpc-endpoint-type Interface \
  --subnet-ids subnet-0111 subnet-0222 subnet-0333 \
  --security-group-ids sg-0endpoint \
  --private-dns-enabled  # resolves kinesis.us-east-1.amazonaws.com to private IP

# Security group for the endpoint — allow inbound 443 from Lambda/ECS task SG
aws ec2 authorize-security-group-ingress \
  --group-id sg-0endpoint \
  --protocol tcp \
  --port 443 \
  --source-group sg-0lambda-tasks \
  --description "Kinesis endpoint access from Lambda/ECS tasks"

# Verify private DNS resolution (should return 10.x.x.x, not 52.x.x.x)
nslookup kinesis.us-east-1.amazonaws.com

A common misconfiguration: creating the endpoint without enabling private DNS. Without private DNS, Lambda functions still resolve kinesis.amazonaws.com to a public IP and the endpoint goes unused. Always set --private-dns-enabled when creating the endpoint.

Endpoint policy: you can attach a resource policy to the VPC endpoint to restrict which streams can be accessed through it. This adds a defense-in-depth layer — even if a Lambda function has IAM permissions to access all streams, the endpoint policy limits it to only streams in the approved list.

# VPC endpoint policy — restrict to specific stream ARNs only
cat > endpoint-policy.json <<'EOF'
{
  "Statement": [{
    "Effect": "Allow",
    "Principal": "*",
    "Action": [
      "kinesis:PutRecord",
      "kinesis:PutRecords",
      "kinesis:GetRecords",
      "kinesis:GetShardIterator",
      "kinesis:DescribeStream",
      "kinesis:ListShards"
    ],
    "Resource": [
      "arn:aws:kinesis:us-east-1:123456789012:stream/mcp-tool-events",
      "arn:aws:kinesis:us-east-1:123456789012:stream/mcp-anomalies"
    ]
  }]
}
EOF

aws ec2 modify-vpc-endpoint \
  --vpc-endpoint-id vpce-0abc12345 \
  --policy-document file://endpoint-policy.json

CloudTrail data events for Kinesis API audit

CloudTrail management events (CreateStream, DeleteStream, UpdateShardCount) are logged automatically with no configuration. Data events (PutRecord, PutRecords, GetRecords, GetShardIterator) must be explicitly enabled and cost $0.10 per 100K API calls — significant for high-throughput MCP streams. Enable data events only on streams that carry sensitive data or where you have a compliance requirement to track who read what.

# Enable CloudTrail data events for specific Kinesis streams
aws cloudtrail put-event-selectors \
  --trail-name mcp-audit-trail \
  --advanced-event-selectors '[
    {
      "Name": "KinesisDataEventsForMcpStreams",
      "FieldSelectors": [
        { "Field": "eventCategory", "Equals": ["Data"] },
        { "Field": "resources.type", "Equals": ["AWS::Kinesis::Stream"] },
        { "Field": "resources.ARN", "StartsWith": [
          "arn:aws:kinesis:us-east-1:123456789012:stream/mcp-"
        ]}
      ]
    }
  ]'

# Query recent GetRecords calls from CloudTrail Lake (faster than S3 Athena for recent events)
aws cloudtrail start-query \
  --query-statement "
    SELECT userIdentity.arn, eventTime, requestParameters
    FROM events
    WHERE eventName = 'GetRecords'
      AND eventSource = 'kinesis.amazonaws.com'
      AND eventTime > '2026-09-20 00:00:00'
    ORDER BY eventTime DESC
    LIMIT 50
  " \
  --event-data-store arn:aws:cloudtrail:us-east-1:123456789012:eventdatastore/EXAMPLE

Cross-account stream access

Kinesis Data Streams does not support resource-based policies on the stream itself (unlike S3 or SQS). Cross-account access requires the IAM principal in the other account to assume a role in the stream's account. The typical pattern for cross-account consumers: the consumer account's role assumes a cross-account role in the stream's account that has Kinesis read permissions, and that cross-account role trusts the consumer role via its trust policy.

// In the stream's account — cross-account consumer role
const crossAccountConsumerRole = new Role(this, "CrossAccountKinesisConsumerRole", {
  roleName: "mcp-kinesis-cross-account-consumer",
  assumedBy: new ArnPrincipal("arn:aws:iam::444455556666:role/AnalyticsWorkerRole"),
  description: "Allows analytics account to consume mcp-tool-events stream",
});
crossAccountConsumerRole.addToPolicy(new PolicyStatement({
  effect: Effect.ALLOW,
  actions: [
    "kinesis:GetRecords",
    "kinesis:GetShardIterator",
    "kinesis:DescribeStream",
    "kinesis:DescribeStreamSummary",
    "kinesis:ListShards",
    "kinesis:SubscribeToShard",
    "kinesis:RegisterStreamConsumer",
    "kinesis:DescribeStreamConsumer",
  ],
  resources: [
    stream.streamArn,
    `${stream.streamArn}/consumer/*`,
  ],
}));
// Allow decrypt of CMK-encrypted records
cmk.grantDecrypt(crossAccountConsumerRole);

// In the consumer account — assume the cross-account role
// (set AWS_ROLE_ARN env var to the cross-account role ARN)
const stsClient = new STSClient({ region: "us-east-1" });
const assumed = await stsClient.send(new AssumeRoleCommand({
  RoleArn: process.env.KINESIS_CROSS_ACCOUNT_ROLE_ARN!,
  RoleSessionName: "analytics-consumer",
  DurationSeconds: 3600,
}));
const credentials = {
  accessKeyId: assumed.Credentials!.AccessKeyId!,
  secretAccessKey: assumed.Credentials!.SecretAccessKey!,
  sessionToken: assumed.Credentials!.SessionToken!,
};
const kinesisClient = new KinesisClient({ region: "us-east-1", credentials });

Security observability — monitoring for stream access anomalies

Beyond access control, use CloudWatch and CloudTrail to detect anomalous stream access patterns: unexpected consumers, unusually high GetRecords rates from a single role, or PutRecords from an IP outside your VPC.

# CloudWatch metric filter for Kinesis GetRecords from unexpected IAM roles
# (Requires CloudTrail data events → CloudWatch Logs delivery enabled)
aws logs put-metric-filter \
  --log-group-name CloudTrail/KinesisDataEvents \
  --filter-name unexpected-kinesis-consumer \
  --filter-pattern '{ ($.eventName = "GetRecords") && ($.userIdentity.arn != "*mcp-analytics*") }' \
  --metric-transformations \
    metricName=UnexpectedKinesisGetRecords,metricNamespace=MCP/Security,metricValue=1

aws cloudwatch put-metric-alarm \
  --alarm-name mcp-unexpected-kinesis-consumer \
  --namespace MCP/Security \
  --metric-name UnexpectedKinesisGetRecords \
  --statistic Sum \
  --period 300 \
  --evaluation-periods 1 \
  --threshold 1 \
  --comparison-operator GreaterThanOrEqualToThreshold \
  --alarm-description "Kinesis stream accessed by unexpected IAM role"

Security configuration reference

Common failure modes reference