(aws-lambda-event-sources): Unable to establish MSK trigger with secret
- Dominant language
- TypeScript
- Stars
- 12.9k
- Forks
- 4.6k
- Avg merge
- 2d 3h
- Merged PRs (30d)
- 83
Description
### What is the problem?
I have a stack with a MSK cluster that uses clientAuthentication via SASL/SCRAM. I have created a KMS key and a secret, that uses that key via the Web UI. The secret is setup according to the documentation (username, password, secret has the AmazonMSK_ prefix). In the stack I also set up a Lambda function which should be triggered for a specific topic. At first I had tested it without clientAuthentication and it worked nicely. Now that I have clientAuthentication enabled with the secret I get the following error as soon as it comes to creating the event source mapping:
```
Failed resources:
MskExampleStack | 20:35:08 | CREATE_FAILED | AWS::Lambda::EventSourceMapping | ListenerHandler/KafkaEventSource:MskExampleStackListenerHandler886655MyTopic (ListenerHandlerKafkaEventSourceMskExampleStackListenerHandler886655MyTopicBSSH423)
Resource handler returned message: "Invalid request provided: Cannot access secret manager value arn:aws:secretsmanager:eu-central-1:4711:secret:AmazonMSK_dev-clientsecret-0815.
Please ensure the role can perform the 'secretsmanager:GetSecretValue' action on your broker in IAM.
(Service: Lambda, Status Code: 400, Request ID: 123456789, Extended Request ID: null)"
(RequestToken: 987654321, HandlerErrorCode: InvalidRequest)
```
### Reproduction Steps
This is a minimal stack to demonstrate the issue:
```ts
import {
aws_ec2,
aws_iam,
aws_kms,
aws_lambda,
aws_msk,
aws_secretsmanager,
custom_resources as cr,
Stack,
StackProps,
} from 'aws-cdk-lib';
import {Construct} from 'constructs';
import {ManagedKafkaEventSource} from "aws-cdk-lib/aws-lambda-event-sources";
import {StartingPosition} from "aws-cdk-lib/aws-lambda";
const KAFKA_ACCESS_SECRET_ARN = '';
const CLUSTER_ENCRYPTION_KEY = '';
export class MskDemoStack extends Stack {
constructor(scope: Construct, id: string, props?: StackProps) {
super(scope, id, props);
const vpc = new aws_ec2.Vpc(this, 'VPC', {
cidr: '10.0.0.0/16',
maxAzs: 3,
subnetConfiguration: [
{
subnetType: aws_ec2.SubnetType.PRIVATE_ISOLATED,
name: "ListenerPrivate",
cidrMask: 24,
},
{
subnetType: aws_ec2.SubnetType.PRIVATE_ISOLATED,
name: "ClusterNodes",
cidrMask: 24
},
],
});
new aws_ec2.InterfaceVpcEndpoint(this, 'lambdaVPCEndpoint', {
vpc,
service: aws_ec2.InterfaceVpcEndpointAwsService.LAMBDA,
subnets: vpc.selectSubnets({subnetGroupName: 'ClusterNodes'})
});
new aws_ec2.InterfaceVpcEndpoint(this, 'secretsManagerVPCEndpoint', {
vpc,
service: aws_ec2.InterfaceVpcEndpointAwsService.SECRETS_MANAGER,
subnets: vpc.selectSubnets({subnetGroupName: 'ClusterNodes'})
});
const mskSecurityGroup = new aws_ec2.SecurityGroup(this, 'MSKSecurityGroup', {
vpc,
description: "Enable access to MSK cluster",
});
mskSecurityGroup.addIngressRule(aws_ec2.Peer.anyIpv4(), aws_ec2.Port.tcp(2181));
mskSecurityGroup.addIngressRule(aws_ec2.Peer.anyIpv4(), aws_ec2.Port.tcp(9094));
mskSecurityGroup.addIngressRule(aws_ec2.Peer.anyIpv4(), aws_ec2.Port.tcp(9092));
const cluster = new aws_msk.CfnCluster(this, 'Cluster', {
kafkaVersion: "2.8.1",
brokerNodeGroupInfo: {
clientSubnets: vpc.selectSubnets({subnetGroupName: 'ClusterNodes'}).subnetIds,
instanceType: "kafka.t3.small",
securityGroups: [mskSecurityGroup.securityGroupId],
storageInfo: {
ebsStorageInfo: {
volumeSize: 5
}
}
},
clusterName: "MSKCluster",
numberOfBrokerNodes: 3,
encryptionInfo: {
encryptionInTransit: {
clientBroker: "TLS",
inCluster: true,
}
},
enhancedMonitoring: "DEFAULT",
clientAuthentication: {
sasl: {
scram: {
enabled: true,
},
},
},
});
const secretClusterCreation = addSecretToCluster(
this,
cluster.ref,
KAFKA_ACCESS_SECRET_ARN,
CLUSTER_ENCRYPTION_KEY,
);
const bootstrapBrokers = getBootstrapBrokers(this, cluster.ref);
const listener = new aws_lambda.Function(this, 'ListenerHandler', {
vpc,
vpcSubnets: { subnetGroupName: 'ListenerPrivate' },
runtime: aws_lambda.Runtime.NODEJS_14_X,
code: aws_lambda.Code.fromAsset('lambda'),
handler: 'listener.handler',
environment: {
'BOOTSTRAP_BROKERS': bootstrapBrokers
},
});
listener.node.addDependency(secretClusterCreation);
listener.role?.addManagedPolicy(
aws_iam.ManagedPolicy
.fromAwsManagedPolicyName("service-role/AWSLambdaVPCAccessExecutionRole")
);
const kafkaClusterKey = aws_kms.Key.fromKeyArn(this, 'ClusterKey', CLUSTER_ENCRYPTION_KEY);
const kafkaAccessSecret = aws_secretsmanager.Secret
.fromSecretAttributes(this, 'kafkaAccessSecret', {
secretCompleteArn: KAFKA_ACCESS_SECRET_ARN,
encryptionKey: kafkaClusterKey
});
listener.addEventSource(new ManagedKafkaEventSource({
clusterArn: cluster.ref,
topic: "MyTopic",
startingPosition: StartingPosition.LATEST,
secret: kafkaAccessSecret,
}));
}
}
/*
Helper function to retrieve the broker node URLs for our Kafka cluster.
It is a string with multiple comma-separated URLs in it.
*/
function getBootstrapBrokers(scope: Construct, clusterArn: string): string {
const result = new cr.AwsCustomResource(scope, 'GetBootstrapBrokerStringTls', {
onUpdate: {
service: 'Kafka',
action: 'getBootstrapBrokers',
parameters: {
ClusterArn: clusterArn,
},
physicalResourceId: cr.PhysicalResourceId.of('BootstrapBrokerList'),
},
policy: cr.AwsCustomResourcePolicy.fromSdkCalls({
resources: [clusterArn],
}),
});
//return result.getResponseField('BootstrapBrokerStringTls');
return result.getResponseField('BootstrapBrokerStringSaslScram');
}
/*
Helper function to add a secret via its ARN to the cluster.
*/
function addSecretToCluster(scope: Construct, clusterArn: string, secretArn: string, keyArn: string) {
return new cr.AwsCustomResource(scope, 'BatchAssociateScramSecret', {
onUpdate: {
service: 'Kafka',
action: 'batchAssociateScramSecret',
parameters: {
ClusterArn: clusterArn,
SecretArnList: [secretArn],
},
physicalResourceId: cr.PhysicalResourceId.of('CreateMSKUser'),
},
policy: cr.AwsCustomResourcePolicy.fromStatements([
new aws_iam.PolicyStatement({
actions: ['kms:CreateGrant'],
resources: ['*'],
}),
new aws_iam.PolicyStatement({
actions: ['kafka:BatchAssociateScramSecret'],
resources: [clusterArn],
}),
]),
});
}
```
### What did you expect to happen?
I would like the event source mapping to succeed as it already does, when I am not using clientAuthentication.
### What actually happened?
Now that I have clientAuthentication enabled with the secret I get the following error as soon as it comes to creating the event source mapping:
```
Failed resources:
MskExampleStack | 20:35:08 | CREATE_FAILED | AWS::Lambda::EventSourceMapping | ListenerHandler/KafkaEventSource:MskExampleStackListenerHandler886655MyTopic (ListenerHandlerKafkaEventSourceMskExampleStackListenerHandler886655MyTopicBSSH423)
Resource handler returned message: "Invalid request provided: Cannot access secret manager value arn:aws:secretsmanager:eu-central-1:4711:secret:AmazonMSK_dev-clientsecret-0815.
Please ensure the role can perform the 'secretsmanager:GetSecretValue' action on your broker in IAM.
(Service: Lambda, Status Code: 400, Request ID: 123456789, Extended Request ID: null)"
(RequestToken: 987654321, HandlerErrorCode: InvalidRequest)
```
### CDK CLI Version
2.8.0
### Framework Version
2.1058.0
### Node.js Version
14
### OS
Windows
### Language
Typescript
### Language Version
3.9.10
### Other information
_No response_
Contributor guide
Research direction
Start with the aws-lambda-event-sources ManagedKafkaEventSource implementation and the generated AWS::Lambda::EventSourceMapping for the secret and listener role. Reproduce the stack from the issue, then trace the IAM and encrypted-secret configuration involved in creation. Done means the event source mapping is created successfully with the MSK SCRAM secret, while the existing unauthenticated case remains working.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- aws, typescript
- Domain
- cloud, infrastructure
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100