Amazon MSKでSASL/SCRAM認証を試してみた

			
			.
├── cfn
│   ├── msk_infra.yml
│   └── secrets_manager.yml
└── sam
    ├── function
    │   └── msk_processor.py
    ├── samconfig.toml
    └── template.yaml

		

			
			AWSTemplateFormatVersion: 2010-09-09
Description: MSK User Credentials for SASL/SCRAM

Parameters:
  Username:
    Description: Username for MSK Cluster
    Default: test-user
    Type: String
  Password:
    Description: Password for MSK Cluster
    Type: String
    NoEcho: true

Resources:
  # KMSキーの作成
  MSKSecretsKMSKey:
    Type: AWS::KMS::Key
    Properties:
      Description: "KMS Key for MSK Secrets Manager"

  # KMSキーエイリアス
  MSKSecretsKMSKeyAlias:
    Type: AWS::KMS::Alias
    Properties:
      AliasName: !Sub "alias/uehara-msk-secrets"
      TargetKeyId: !Ref MSKSecretsKMSKey

  # Secrets Managerシークレット
  MSKUserSecret:
    Type: AWS::SecretsManager::Secret
    Properties:
      Name: "AmazonMSK_test_user"
      Description: "MSK SASL/SCRAM user credentials"
      KmsKeyId: !Ref MSKSecretsKMSKey
      SecretString: !Sub |
        {
          "username": "${Username}",
          "password": "${Password}"
        }

		

			
			# SASL/SCRAM認証
ClientAuthentication:
  Sasl:
    Scram:
      Enabled: true

		

			
			version = 0.1

[default]
region = "ap-northeast-1"

[default.build.parameters]
debug = true

[default.deploy.parameters]
stack_name = "uehara-msk-test-function"
s3_bucket = "<デプロイ用のS3バケット名>"
s3_prefix = "sam-deploy"
capabilities = "CAPABILITY_NAMED_IAM"
confirm_changeset = true
resolve_image_repos = true

		

			
			AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: "SAM Application"

Parameters:
  MSKClusterArn:
    Type: String
    Description: ARN of the MSK Cluster
  MSKUserSecretArn:
    Type: String
    Description: ARN of the MSK Secrets

Resources:
  # Lambda実行ロール
  MSKProcessorExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: msk-processor-execution-role
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
        - arn:aws:iam::aws:policy/service-role/AWSLambdaMSKExecutionRole
        - arn:aws:iam::aws:policy/SecretsManagerReadWrite
      Policies:
        - PolicyName: MSKEventSourceMappingPolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              # SCRAMシークレット関連の権限
              - Effect: Allow
                Action:
                  - kafka:ListScramSecrets
                  - kafka:BatchAssociateScramSecret
                  - kafka:BatchDisassociateScramSecret
                Resource: "*"

  # Lambda関数
  MSKProcessorFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: msk-processor-function
      CodeUri: function/
      Handler: msk_processor.lambda_handler
      Runtime: python3.11
      Architectures:
        - x86_64
      Timeout: 300
      MemorySize: 512
      Role: !GetAtt MSKProcessorExecutionRole.Arn
      Layers:
        - "arn:aws:lambda:ap-northeast-1:336392948345:layer:AWSSDKPandas-Python311:22"
      Environment:
        Variables:
          LOG_LEVEL: INFO
      Events:
        MSKEvent:
          Type: MSK
          Properties:
            Stream: !Ref MSKClusterArn
            StartingPosition: LATEST
            Topics:
              - test-topic
            BatchSize: 5
            # SASL/SCRAM認証設定
            SourceAccessConfigurations:
              - Type: SASL_SCRAM_512_AUTH
                URI: !Ref MSKUserSecretArn

Outputs:
  MSKProcessorFunctionArn:
    Description: ARN of the MSK Processor Lambda function
    Value: !GetAtt MSKProcessorFunction.Arn
    Export:
      Name: !Sub "${AWS::StackName}-MSKProcessorFunctionArn"

  MSKProcessorExecutionRoleArn:
    Description: ARN of the MSK Processor Lambda execution role
    Value: !GetAtt MSKProcessorExecutionRole.Arn
    Export:
      Name: !Sub "${AWS::StackName}-MSKProcessorExecutionRoleArn"

		

			
			SourceAccessConfigurations:
  - Type: SASL_SCRAM_512_AUTH
    URI: !Ref MSKUserSecretArn

		

			
			import base64
import json
import logging
from typing import Any, Dict, List

# ログ設定
logger = logging.getLogger()
logger.setLevel(logging.INFO)

def process_kafka_record(record: Dict[str, Any], topic_name: str) -> Dict[str, Any]:
    try:
        # Kafkaメッセージのデコード
        if "value" in record:
            # Base64でエンコードされたメッセージをデコード
            message_bytes = base64.b64decode(record["value"])
            message_str = message_bytes.decode("utf-8")

            try:
                # JSONとしてパース試行
                message_data = json.loads(message_str)
            except json.JSONDecodeError:
                # JSONでない場合は文字列として扱う
                message_data = message_str
        else:
            message_data = None

        # メタデータの抽出
        partition = record.get("partition", "unknown")
        offset = record.get("offset", "unknown")
        timestamp = record.get("timestamp", "unknown")

        logger.info(
            f"Processing message from topic '{topic_name}', partition {partition}, offset {offset}"
        )
        logger.info(f"Message content: {message_data}")

        # ここでビジネスロジックを実装

        processed_data = {
            "topic": topic_name,
            "partition": partition,
            "offset": offset,
            "timestamp": timestamp,
            "original_message": message_data,
            "processing_status": "success",
        }

        return processed_data

    except Exception as e:
        logger.error(f"Error processing record: {str(e)}")
        return {
            "topic": topic_name,
            "partition": record.get("partition", "unknown"),
            "offset": record.get("offset", "unknown"),
            "error": str(e),
            "processing_status": "error",
        }

def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    logger.info(f"Received event: {json.dumps(event, default=str)}")

    processed_records = []

    try:
        # MSKイベントの構造: event['records']にトピック別のレコードが含まれる
        if "records" in event:
            for topic_name, topic_records in event["records"].items():
                logger.info(f"Processing topic: {topic_name}")

                for record in topic_records:
                    processed_record = process_kafka_record(record, topic_name)
                    processed_records.append(processed_record)

        logger.info(f"Successfully processed {len(processed_records)} records")

        return {
            "statusCode": 200,
            "body": json.dumps(
                {
                    "message": f"Successfully processed {len(processed_records)} records",
                    "processed_records": len(processed_records),
                }
            ),
        }

    except Exception as e:
        logger.error(f"Error processing MSK event: {str(e)}")
        raise

		

			
			$ sam deploy --parameter-overrides "MSKClusterArn=<作成したMSKクラスターのARN> MSKUserSecretArn=<作成したシークレットのARN>"

		

			
			# CloudShellでKafka CLIツールをダウンロード
$ wget https://downloads.apache.org/kafka/3.9.1/kafka_2.13-3.9.1.tgz
$ tar -xzf kafka_2.13-3.9.1.tgz

# クライアントの設定ファイルを作成
$ cat > client.properties << EOF
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
  username="<設定したユーザー名>" \
  password="<設定したパスワード>";
EOF

		

			
			# トピックの作成
$ ./kafka_2.13-3.9.1/bin/kafka-topics.sh \
  --bootstrap-server <MSKクラスターのBootstrap Serverのホスト名>:9096 \
  --command-config client.properties \
  --create \
  --topic test-topic \
  --partitions 3

# メッセージ送信  
echo '{"message": "Hello Kafka"}' | 
./kafka_2.13-3.9.1/bin/kafka-console-producer.sh \
  --bootstrap-server <MSKクラスターのBootstrap Serverのホスト名>:9096 \
  --producer.config client.properties \
  --topic test-topic