Realtime Compute for Apache Flink:安全なアクセスのベストプラクティス

ソリューションの概要

このソリューションは、最小限の権限の原則に基づいて AccessKey ペアを使用して、より安全なアクセスを実現します。このソリューションでは、Alibaba Cloud アカウントの AccessKey ペアではなく、特定のアップストリームリソースとダウンストリームリソースへのアクセス権限、または最小限の必要な権限が付与された RAM ユーザーの AccessKey ペアを使用します。さらに、AccessKey ペアは Flink の変数を使用して暗号化され、プレーンテキストでの AccessKey ペアの漏洩のリスクを軽減します。これにより、Alibaba Cloud アカウント内のすべてのクラウド リソースのセキュリティリスクが軽減されます。

このトピックでは、RAM ユーザーが作成されます。 RAM ユーザーには、Simple Log Service のプロジェクト内の特定のログストアに対する権限が付与されます。 RAM ユーザーの AccessKey ペアは、Flink の変数を使用して暗号化され、Simple Log Service からのデータの読み取りと書き込みに使用されます。

手順

1. Alibaba Cloud アカウントを使用して、または RAM 管理者として RAM ユーザーを作成します。詳細については、「RAM ユーザーの作成」をご参照ください。

   [アクセスモード] セクションで [OpenAPI アクセス] を選択する必要があります。その後、AccessKey ID と AccessKey Secret が RAM ユーザーに対して自動的に生成されます。

   重要

   RAM ユーザーの AccessKey Secret は、RAM ユーザーの作成時にのみ表示されます。 AccessKey Secret は秘密にしてください。

2. RAM ユーザーに Simple Log Service への権限を付与します。

   1. カスタムポリシーを作成します。

      1. 左側のナビゲーションペインで、権限 > ポリシー を選択します。

      2. [ポリシー] ページで、[ポリシーの作成] をクリックします。

         

      3. [ポリシーの作成] ページで、[JSON] タブをクリックし、コードエディターの既存のスクリプトを次のサンプルコードに置き換えてから、[ポリシー情報を編集するために次へ] をクリックします。

         ビジネス要件に基づいて、サンプルコードの project パラメーターと logstore パラメーターを置き換えます。カスタムポリシーの作成方法の詳細については、「カスタムポリシーを使用して RAM ユーザーに権限を付与する例」をご参照ください。

         特定のログストアに対する読み取り専用権限を付与する

         {
             "Version": "1",
             "Statement": [
                 {
                     "Action": "log:ListProject",
                     "Resource": "acs:log:*:*:project/*",
                     "Effect": "Allow"
                 },
                 {
                     "Action": "log:List*",
                     "Resource": "acs:log:*:*:project/<プロジェクト名>/logstore/*",  // プロジェクト名を指定
                     "Effect": "Allow"
                 },
                 {
                     "Action": [
                         "log:Get*",
                         "log:List*"
                     ],
                     "Resource": "acs:log:*:*:project/<プロジェクト名>/logstore/<ログストア名>", // プロジェクト名とログストア名を指定
                     "Effect": "Allow"
                 }
             ]
         }

         特定のログストアへの書き込み権限を付与する

         指定されたログストアにデータを書き込む権限のみを RAM ユーザーに付与するには、次のポリシーを使用します。

         ポリシーを作成するときに、logstore キーワードの後にログストアまたはメトリックストアを指定できます。メトリックストアを管理する場合も、次のポリシーが適用されます。

         {
           "Version":"1",
           "Statement":[
             {
               "Effect":"Allow",
               "Action":[
                 "log:PostLogStoreLogs"
               ],
               "Resource":[
                 "acs:log:*:*:project/<プロジェクト名>/logstore/<ログストア名>" // プロジェクト名とログストア名を指定
               ]
             }
           ]
         }

      4. ポリシーの名前と説明を入力し、[OK] をクリックします。

3. 前の手順で作成したカスタムポリシーを使用して、RAM ユーザーに権限を付与します。詳細については、「RAM ユーザーに権限を付与する」をご参照ください。

4. プレーンテキストでの AccessKey ペアの漏洩のリスクを軽減するために変数を構成します。

   手順 1 で RAM ユーザーを作成するときに取得した AccessKey ID と AccessKey シークレットの変数を作成します。変数が作成された後、AccessKey ID と AccessKey シークレットを指定する代わりに、変数名を呼び出すだけで済みます。変数の作成方法の詳細については、「変数とキーの管理」をご参照ください。このトピックでは、AccessKey ID と AccessKey シークレット用に slslak と slsaks という名前の変数が作成されます。

5. Simple Log Serviceからデータを読み取るFlinkデプロイメントを作成します。

   SQLドラフトを開発する際に、${secret_values.変数名} 形式で変数を呼び出すことができます。これにより、AccessKeyペアのプレーンテキストでの漏洩によるセキュリティリスクを防ぎます。サンプルコード：

   CREATE TEMPORARY TABLE sls_input(
      `__source__` STRING METADATA VIRTUAL,
      __tag__ MAP<VARCHAR, VARCHAR> METADATA VIRTUAL,
      `__topic__` STRING METADATA VIRTUAL,
      deploymentName STRING,
      `level`STRING,
      `location` STRING,
      message STRING,
      thread STRING,
      `time`STRING
   ) WITH (
     'connector' = 'sls',
     'endpoint' ='cn-beijing-intranet.log.aliyuncs.com',
     'accessId' = '${secret_values.slsak}', // AccessId を設定します
     'accessKey' = '${secret_values.slsaks}', // AccessKey を設定します
     'starttime' = '2024-08-30 15:39:00', // 開始時刻を設定します
     'project' ='test', // プロジェクト名を設定します
     'logstore' ='flinktest' // Logstore 名を設定します
   );
   
   CREATE TEMPORARY TABLE blackhole_sink(
      `__source__` STRING,
      `__topic__` STRING,
      deploymentName STRING,
      `level` STRING,
      `location` STRING,
      message STRING,
      thread STRING,
      `time` STRING,
      receive_time BIGINT
   ) WITH (
     'connector' = 'blackhole'
   );
   
   INSERT INTO blackhole_sink
   SELECT `__source__`,
      `__topic__`,
      deploymentName,
      `level`,
      `location`,
      message,
      thread,
      `time`,
      cast(__tag__['__receive_time__'] as bigint) as receive_time
   FROM sls_input; 
   

参考資料

• Flink を使用して Object Storage Service (OSS) からデータを読み書きするには、カスタムポリシーを構成して、特定のバケットに対する読み取りおよび書き込み権限を RAM ユーザーに付与します。詳細については、「RAM ポリシー」をご参照ください。

• SQL ドラフトの開発方法の詳細については、「SQL ドラフトを開発する」をご参照ください。

• Flink は、AccessKey ペアやパスワードなどの機密データのプレーンテキストでの漏洩によるセキュリティリスクを防ぐために、ジョブ変数とプロジェクト変数を提供します。これらの変数は複数のシナリオで使用できます。変数の作成方法の詳細については、「変数とキーの管理」をご参照ください。