Realtime Compute for Apache Flink:Key Management Service を使用して Flink のデータベースパスワードを暗号化および復号化する

方法 1: OpenAPI ポータルで KMS Encrypt 操作を呼び出す

1. Alibaba Cloud OpenAPI ポータル にアクセスします。ターゲットリージョンを選択します。

2. KeyId と Plaintext を設定します。

3. [呼び出しを開始] をクリックします。

4. 暗号文を表示します。

詳細については、「暗号化する」をご参照ください。

方法 2: IntelliJ IDEA から KMS Encrypt 操作を呼び出す

1. KMS キーへのインターネットアクセスを有効にします。詳細については、「インターネット経由で KMS インスタンスキーにアクセスする。」をご参照ください。

   KMS キーは、デフォルトでは VPC ネットワークを介してのみアクセスできます。

2. ALIBABA_CLOUD_ACCESS_KEY_ID および ALIBABA_CLOUD_ACCESS_KEY_SECRET 環境変数を構成します。AccessKey ペアの取得方法については、「アカウントの AccessKey ペアを表示するにはどうすればよいですか？」をご参照ください。

3. ターゲットの IntelliJ IDEA プロジェクトで、EncryptFlink という名前のクラスファイルを作成します。

4. 次のコードスニペットを EncryptFlink クラスファイルにコピーして貼り付けます。構成オプションの値を特定のセットアップに合わせて変更することを忘れないでください。

   package org.example;
   
   import com.aliyun.kms20160120.models.EncryptResponse;
   import com.aliyun.kms20160120.models.EncryptResponseBody;
   import com.aliyun.tea.*;
   
   public class EncryptFlink {
   
       /**
        * <b>説明</b> :
        * <p>AccessKey ペアを使用してクライアントを初期化します。</p>
        * @return Client
        *
        * @throws Exception
        */
       public static com.aliyun.kms20160120.Client createClient() throws Exception {
           // プロジェクトコードが漏洩した場合、AccessKey ペアが漏洩し、アカウント内のすべてのリソースのセキュリティが損なわれる可能性があります。次のサンプルコードは参考用としてのみ提供されています。
           com.aliyun.teaopenapi.models.Config config = new com.aliyun.teaopenapi.models.Config()
                   // 必須。 ALIBABA_CLOUD_ACCESS_KEY_ID 環境変数が構成されていることを確認してください。
                   .setAccessKeyId(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"))
                   // 必須。 ALIBABA_CLOUD_ACCESS_KEY_SECRET 環境変数が構成されていることを確認してください。
                   .setAccessKeySecret(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
           // エンドポイントを指定します。詳細については、https://api.aliyun.com/product/Kms をご覧ください。
           config.endpoint = "kms.cn-hangzhou.aliyuncs.com";
           return new com.aliyun.kms20160120.Client(config);
       }
   
       public static void main(String[] args_) throws Exception {
           java.util.List<String> args = java.util.Arrays.asList(args_);
           com.aliyun.kms20160120.Client client = EncryptFlink.createClient();
           com.aliyun.kms20160120.models.EncryptRequest encryptRequest = new com.aliyun.kms20160120.models.EncryptRequest()
                   .setPlaintext("flink_rds_password@123")
                   .setKeyId("key-hzz67ab1ff4e750h****");
           com.aliyun.teautil.models.RuntimeOptions runtime = new com.aliyun.teautil.models.RuntimeOptions();
           try {
               // 必要に応じて、API 操作のレスポンスを表示するための独自のコードを記述してください。
               EncryptResponse encryptResponse = client.encryptWithOptions(encryptRequest, runtime);
               EncryptResponseBody body = encryptResponse.getBody();
               System.out.println(body.getCiphertextBlob());
           } catch (TeaException error) {
               // 実際のビジネスシナリオでは、例外を慎重に処理し、プロジェクトで例外を無視しないでください。この例では、デモンストレーションのためにエラーメッセージが単に印刷されます。
               // エラーメッセージ。
               System.out.println(error.getMessage());
               // トラブルシューティング用の URL。
               System.out.println(error.getData().get("Recommend"));
               com.aliyun.teautil.Common.assertAsString(error.message);
           } catch (Exception _error) {
               TeaException error = new TeaException(_error.getMessage(), _error);
               // 実際のビジネスシナリオでは、例外を慎重に処理し、プロジェクトで例外を無視しないでください。この例では、デモンストレーションのためにエラーメッセージが単に印刷されます。
               // エラーメッセージ。
               System.out.println(error.getMessage());
               // トラブルシューティング用の URL。
               System.out.println(error.getData().get("Recommend"));
               com.aliyun.teautil.Common.assertAsString(error.message);
           }
       }
   }

   オプション	説明	例
   config.endpoint	KMS インスタンスのエンドポイント。	kms.cn-hangzhou.aliyuncs.com
   Plaintext	暗号化するプレーンテキストパスワード。	flink_rds_password@123
   KeyId	KMS キー ID。	key-hzz67ab1ff4e750h****

5. POM.xml ファイルに次の依存関係を追加します。

       <dependency>
         <groupId>com.aliyun</groupId>
         <artifactId>kms20160120</artifactId>
         <version>1.2.3</version>
       </dependency>
   
       <dependency>
         <groupId>com.aliyun</groupId>
         <artifactId>tea</artifactId>
         <version>1.3.2</version>
       </dependency>

6. EncryptFlink クラスファイルを実行して、暗号文を取得します。

KmsJavaDemo

package org.example;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.utils.TypeConversions;
import org.apache.flink.api.java.utils.ParameterTool;


public class JavaDemo {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // パラメーターを解析して AccessKey ペアを取得します。
        final ParameterTool params = ParameterTool.fromArgs(args);
        String ak = params.get("akid");
        String sk = params.get("aksecret");

        // 暗号化されたパスワードを復号します。
        String encryptedPassword = "a2V5LWh6ejY3YTJmZTY5ejR2NTlpOHE1MC03d3ozYWU1anlzC6NLoC0JHHEXTJ4P/4iVOe/B+eniv8EcaviQDzZWWNPedOYkoFFYWA==";
        String decryptedPassword = KmsUtil.decrypt(ak, sk, encryptedPassword);

        // デシリアライザーを構築します。
        DataType dataType =
                DataTypes.ROW(
                        DataTypes.FIELD("id", DataTypes.INT()),
                        DataTypes.FIELD("username", DataTypes.STRING()),
                        DataTypes.FIELD("age", DataTypes.INT()));

        LogicalType logicalType = TypeConversions.fromDataToLogicalType(dataType);
        InternalTypeInfo<RowData> typeInfo = InternalTypeInfo.of(logicalType);
        RowDataDebeziumDeserializeSchema deserializer =
                RowDataDebeziumDeserializeSchema.newBuilder()
                        .setPhysicalRowType((RowType) dataType.getLogicalType())
                        .setResultTypeInfo(typeInfo)
                        .build();

        // データソース (com.ververica.cdc.connectors.mysql.source.MySqlSource) を構成します。
        MySqlSource<RowData> mySqlSource =
                MySqlSource.<RowData>builder()
                        .hostname("rm-bp****2ye09w72zjq.mysql.rds.aliyuncs.com")
                        .port(3306)
                        .databaseList("school") // データベースを指定します。
                        .tableList("school.student") // テーブルを指定します。
                        .username("flink_rds_user")
                        .password(decryptedPassword)
                        // RowData 構造のデータを初期化します。
                        .deserializer(deserializer)
                        .build();

        // 外部データソースを Flink DataStream プログラムに統合します
        // ウォーターマーク戦略を使用しないでください。
        DataStreamSource<RowData> mySQLSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source");

        // 標準出力に書き込みます。
        mySQLSource.print();

        // プログラムを実行します。
        env.execute("MySQL CDC Test");
    }
}

POM.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.aliyun</groupId>
    <artifactId>KmsJavaDemo</artifactId>
    <version>1.0-SNAPSHOT</version>
    <name>Flink MySQL CDC Demo</name>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <flink.version>1.17.1</flink.version>
        <flink-cdc.version>2.4.2</flink-cdc.version>
        <log4j.version>2.17.1</log4j.version>
    </properties>
    <dependencies>
        <! -- Flink コア依存関係 (プログラムをパッケージングする際、スコープを provided に設定します) -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- Apache Flink 向けリアルタイムコンピューティングの MySQL CDC コネクタ -->
        <dependency>
            <groupId>com.alibaba.ververica</groupId>
            <artifactId>ververica-connector-mysql</artifactId>
            <version>1.17-vvr-8.0.4-1</version>
            <! -- ローカルで実行する場合は、次の行をコメントアウトします -->
            <!-- <scope>provided</scope> -->
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-runtime</artifactId>
            <version>1.17.1</version> <! -- バージョンは、先に指定した Flink のバージョンと一致させる必要があります -->
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-core</artifactId>
            <version>1.17.1</version>
            <scope>provided</scope>
        </dependency>


        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-common</artifactId>
            <version>1.17.1</version> <! -- バージョンは、先に指定した Flink のバージョンと一致させる必要があります -->
            <scope>provided</scope>
        </dependency>


        <! -- ログの依存関係 -->
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>2.17.1</version>
            <scope>runtime</scope>
        </dependency>
        <dependency>
            <groupId>com.aliyun</groupId>
            <artifactId>kms20160120</artifactId>
            <version>1.2.3</version>
        </dependency>

        <dependency>
            <groupId>com.aliyun</groupId>
            <artifactId>tea</artifactId>
            <version>1.3.2</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <! -- コンパイラプラグイン -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.13.0</version> <! -- パッチバージョン -->
                <configuration>
                    <source>${maven.compiler.source}</source>
                    <target>${maven.compiler.target}</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>

            <! -- fat JAR をビルドするためのプラグイン -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.5.1</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <artifactSet>
                                <excludes>
                                    <exclude>org.apache.flink:force-shading</exclude>
                                    <exclude>com.google.code.findbugs:jsr305</exclude>
                                    <! -- ログの依存関係を保持 -->
                                    <!-- <exclude>org.slf4j:*</exclude> -->
                                    <!-- <exclude>org.apache.logging.log4j:*</exclude> -->
                                </excludes>
                            </artifactSet>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                        <exclude>META-INF/MANIFEST.MF</exclude> <! -- キーフィルターを追加 -->
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>org.example.JavaDemo</mainClass>
                                </transformer>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/> <! -- 必要なトランスフォーマーを追加 -->
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>