MSK Connect コネクタにおける転送時の暗号化設定について

2025.08.15

このブログシリーズ「クラウドセキュリティ実践集」では、一般的なセキュリティ課題を取り上げ、「なぜ危険なのか？」というリスクの解説から、「どうやって直すのか？」という具体的な修復手順（コンソール、AWS CLI、Terraformなど）まで、分かりやすく解説します。

この記事では、Amazon MSK Connectコネクタにおける転送中のデータ暗号化の必要性について、リスクと対策を解説します。

修復方法

AWSマネジメントコンソールでの修復手順

ステップ1: MSK Connectコンソールへのアクセス

AWS管理コンソールにログイン
- https://console.aws.amazon.com にアクセス
- 適切なIAM権限（kafkaconnect:*）を持つユーザーでログイン
Amazon MSKサービスに移動
- サービス検索バーで「MSK」と入力
- 「Amazon MSK」を選択
MSK Connectセクションに移動
- 左側のナビゲーションペインから「MSK Connect」を選択
- 「Connectors」タブが表示されることを確認

ステップ2: 新しいコネクタの作成

重要: 既存のコネクタでは暗号化設定を変更できないため、新しいコネクタを作成する必要があります。

コネクタ作成の開始
- 「Create connector」ボタンをクリック
- 作成ウィザードが開始されることを確認
プラグインの選択
- Choose connectorセクションで：
  - 「Use existing custom plugin」または「Create custom plugin」を選択
  - 使用するコネクタプラグイン（例：Confluent S3 Sink Connector）を選択
- 「Next」をクリック

ステップ3: 基本設定

Connector configurationセクションで：
- Connector name: 命名規則に従った名前を入力（例：prod-s3-sink-encrypted）
- Description: 暗号化設定を含む説明を追加（例：「S3への暗号化転送を有効化したシンクコネクタ」）
Kafka clusterセクションで：
- MSK cluster: 対象のMSKクラスターを選択
- Authentication type: 「IAM」を選択（最もセキュア）
- IAM認証により、MSKクラスターとの通信も暗号化されます

ステップ4: 接続先の暗号化設定

Connector configurationセクションで、接続先に応じた暗号化設定を行います：

S3シンクコネクタの暗号化設定例:

# 基本設定
connector.class=io.confluent.connect.s3.S3SinkConnector
tasks.max=2
topics=your-kafka-topic

# S3接続設定（HTTPS使用）
s3.region=ap-northeast-1
s3.bucket.name=my-secure-bucket
# S3へのHTTPS接続はデフォルトで有効

# S3サーバーサイド暗号化（SSE-KMS）
s3.sse.enabled=true
s3.sse.kms.key.id=arn:aws:kms:ap-northeast-1:123456789012:key/12345678-1234-1234-1234-123456789012

# データフォーマット
format.class=io.confluent.connect.s3.format.json.JsonFormat
flush.size=1000
rotate.interval.ms=60000

RDS/Aurora（JDBC）コネクタの暗号化設定例:

# 基本設定
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
tasks.max=1
topics=your-kafka-topic

# データベース接続（SSL/TLS必須）
connection.url=jdbc:mysql://my-rds-instance.region.rds.amazonaws.com:3306/mydb?useSSL=true&requireSSL=true&verifyServerCertificate=true
# Secrets Managerから認証情報を取得（平文パスワードを避ける）
connection.user=${secretsmanager:my-db-secret:username}
connection.password=${secretsmanager:my-db-secret:password}

# SSL/TLS設定（重要）
ssl.mode=REQUIRED
ssl.truststore.location=/opt/kafka/connect-certs/rds-ca-2019-root.jks
ssl.truststore.password=${secretsmanager:my-tls-secret:truststore-password}

# テーブル設定
table.name.format=kafka_${topic}
auto.create=true
auto.evolve=true

Amazon OpenSearchコネクタの暗号化設定例:

# 基本設定
connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
tasks.max=2
topics=your-kafka-topic

# OpenSearch接続（HTTPS必須）
connection.url=https://my-opensearch-domain.region.es.amazonaws.com:443
# HTTPSプロトコルの使用を確認（httpは使用不可）

# 認証設定（Secrets Manager使用）
connection.username=${secretsmanager:opensearch-secret:username}
connection.password=${secretsmanager:opensearch-secret:password}

# SSL/TLS検証
ssl.ignore.validity=false

# データ処理設定
behavior.on.null.values=delete
key.ignore=false
schema.ignore=true
type.name=_doc

ステップ5: セキュリティとアクセス設定

Worker configurationセクションで：
- Worker configuration: デフォルトまたはカスタム設定を選択
- カスタム設定の場合、以下を追加： # ワーカーレベルでのSSL設定 producer.ssl.endpoint.identification.algorithm=https consumer.ssl.endpoint.identification.algorithm=https
Service execution roleセクションで：
- IAM role: MSK Connect用のサービスロールを選択または作成
- 必要な権限：
  - MSKクラスターへのアクセス権限
  - Secrets Managerへの読み取り権限
  - 接続先リソース（S3、RDS等）へのアクセス権限
Securityセクションで：
- VPC: MSKクラスターと同じVPCを選択
- Subnets: プライベートサブネットを選択（必須）
- Security groups: 以下のルールを含むセキュリティグループを選択：
  - MSKクラスターへの9094番ポート（TLS）許可
  - 接続先サービスへの適切なポート許可（HTTPS:443、MySQL:3306等）

ステップ6: ログ設定と作成

Logsセクションで：
- CloudWatch logs: 「Enabled」を選択
- Log group: 新規作成または既存のロググループを選択
- Log level: 「INFO」を選択（暗号化関連のエラーを確実に記録）
確認と作成
- すべての設定を確認（特に暗号化関連の設定）
- 「Create connector」をクリック
- 作成には5-10分程度かかります
作成後の確認
- コネクタのステータスが「Running」になることを確認
- CloudWatch Logsで接続が暗号化されていることを確認