Realtime Compute for Apache Flink：Enkripsi dan Dekripsi Kata Sandi Database di Flink dengan Key Management Service

Metode 1: Panggil operasi Enkripsi KMS di portal OpenAPI

1. Pergi ke Portal OpenAPI Alibaba Cloud. Pilih wilayah target.

2. Tetapkan KeyId dan Plaintext.

3. Klik Initiate Call.

4. Lihat ciphertext.

Untuk informasi lebih lanjut, lihat Enkripsi.

Metode 2: Panggil operasi Enkripsi KMS dari IntelliJ IDEA

1. Aktifkan akses Internet ke kunci KMS. Untuk informasi lebih lanjut, lihat Akses kunci instance KMS melalui Internet.

   Kunci KMS hanya dapat diakses melalui jaringan VPC secara default.

2. Konfigurasikan variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_ID dan ALIBABA_CLOUD_ACCESS_KEY_SECRET. Untuk instruksi tentang cara mendapatkan pasangan AccessKey, lihat Bagaimana Cara Melihat Pasangan AccessKey dari Sebuah Akun?.

3. Di proyek IntelliJ IDEA target, buat file kelas bernama EncryptFlink.

4. Salin dan tempel potongan kode berikut ke file kelas EncryptFlink. Ingatlah untuk memodifikasi nilai opsi konfigurasi agar sesuai dengan pengaturan spesifik Anda.

   package org.example;
   
   import com.aliyun.kms20160120.models.EncryptResponse;
   import com.aliyun.kms20160120.models.EncryptResponseBody;
   import com.aliyun.tea.*;
   
   public class EncryptFlink {
   
       /**
        *  :
        * <p>Gunakan pasangan AccessKey Anda untuk menginisialisasi klien.</p>
        * @return Client
        *
        * @throws Exception
        */
       public static com.aliyun.kms20160120.Client createClient() throws Exception {
           // Jika kode proyek bocor, pasangan AccessKey mungkin bocor, membahayakan keamanan semua sumber daya dalam akun Anda. Contoh kode berikut hanya disediakan sebagai referensi. 
           com.aliyun.teaopenapi.models.Config config = new com.aliyun.teaopenapi.models.Config()
                   // Diperlukan. Pastikan bahwa variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_ID dikonfigurasi. 
                   .setAccessKeyId(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"))
                   // Diperlukan. Pastikan bahwa variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_SECRET dikonfigurasi. 
                   .setAccessKeySecret(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
           // Tentukan titik akhir. Untuk informasi lebih lanjut, kunjungi 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 {
               // Tulis kode Anda sendiri untuk menampilkan respons operasi API jika diperlukan.
               EncryptResponse encryptResponse = client.encryptWithOptions(encryptRequest, runtime);
               EncryptResponseBody body = encryptResponse.getBody();
               System.out.println(body.getCiphertextBlob());
           } catch (TeaException error) {
               // Tangani pengecualian dengan hati-hati dalam skenario bisnis aktual dan jangan abaikan pengecualian dalam proyek Anda. Dalam contoh ini, pesan kesalahan hanya dicetak untuk tujuan demonstrasi. 
               // Pesan kesalahan.
               System.out.println(error.getMessage());
               // URL untuk pemecahan masalah.
               System.out.println(error.getData().get("Recommend"));
               com.aliyun.teautil.Common.assertAsString(error.message);
           } catch (Exception _error) {
               TeaException error = new TeaException(_error.getMessage(), _error);
               // Tangani pengecualian dengan hati-hati dalam skenario bisnis aktual dan jangan abaikan pengecualian dalam proyek Anda. Dalam contoh ini, pesan kesalahan hanya dicetak untuk tujuan demonstrasi. 
               // Pesan kesalahan.
               System.out.println(error.getMessage());
               // URL untuk pemecahan masalah.
               System.out.println(error.getData().get("Recommend"));
               com.aliyun.teautil.Common.assertAsString(error.message);
           }
       }
   }

   Opsi	Deskripsi	Contoh
   config.endpoint	Titik akhir dari instance KMS Anda.	kms.cn-hangzhou.aliyuncs.com
   Teks Biasa	Kata sandi teks biasa untuk dienkripsi.	flink_rds_password@123
   KeyId	ID kunci KMS.	key-hzz67ab1ff4e750h****

5. Tambahkan dependensi berikut ke file 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. Jalankan file kelas EncryptFlink untuk mendapatkan ciphertext.

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();
       
        // Parsing parameter untuk mendapatkan pasangan AccessKey.
        final ParameterTool params = ParameterTool.fromArgs(args);
        String ak = params.get("akid");
        String sk = params.get("aksecret");
        
        // Mendekripsi kata sandi terenkripsi.
        String encryptedPassword = "a2V5LWh6ejY3YTJmZTY5ejR2NTlpOHE1MC03d3ozYWU1anlzC6NLoC0JHHEXTJ4P/4iVOe/B+eniv8EcaviQDzZWWNPedOYkoFFYWA==";
        String decryptedPassword = KmsUtil.decrypt(ak, sk, encryptedPassword);

        // Membangun deserializer.
        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();

        // Konfigurasikan sumber data (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") // Tentukan database.
                        .tableList("school.student") // Tentukan tabel.
                        .username("flink_rds_user")
                        .password(decryptedPassword)
                        // Inisialisasi data dalam struktur RowData.
                        .deserializer(deserializer)
                        .build();

        // Integrasikan sumber data eksternal ke dalam program Flink DataStream
        // Jangan gunakan strategi watermark.
        DataStreamSource<RowData> mySQLSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "Sumber MySQL");

        // Tulis ke stdout.
        mySQLSource.print();

        // Jalankan program.
        env.execute("Uji CDC MySQL");
    }
}

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>Demo Flink MySQL CDC</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>
        <! -- Dependensi inti Flink (Tetapkan ruang lingkup ke provided saat mengepak program) -->
        <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>

        <!-- Realtime Compute for Apache Flink's MySQL CDC connector -->
        <dependency>
            <groupId>com.alibaba.ververica</groupId>
            <artifactId>ververica-connector-mysql</artifactId>
            <version>1.17-vvr-8.0.4-1</version>
            <! -- Komentari baris berikut untuk eksekusi lokal -->
            <!-- <scope>provided</scope> -->
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-runtime</artifactId>
            <version>1.17.1</version> <! -- Versi harus konsisten dengan versi Flink yang Anda tentukan sebelumnya -->
            <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> <! -- Versi harus konsisten dengan versi Flink yang Anda tentukan sebelumnya -->
            <scope>provided</scope>
        </dependency>


        <! -- Dependensi log -->
        <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 kompiler -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.13.0</version> <! -- Versi tambalan -->
                <configuration>
                    <source>${maven.compiler.source}</source>
                    <target>${maven.compiler.target}</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>

            <! -- Plugin untuk membuat 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>
                                    <! -- Pertahankan dependensi logging -->
                                    <!-- <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> <! -- Tambahkan filter kunci -->
                                    </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"/> <! -- Tambahkan transformer yang diperlukan -->
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>