Realtime Compute for Apache Flink:Encrypt and decrypt database passwords in Flink with Key Management Service

Method 1: Call the KMS Encrypt operation on OpenAPI portal

1. Go to Alibaba Cloud OpenAPI Portal. Choose the target region.

2. Set KeyId and Plaintext.

3. Click Initiate Call.

4. View the ciphertext.

For more information, see Encrypt.

Method 2: Call the KMS Encrypt operation from IntelliJ IDEA

1. Enable Internet access to KMS keys. For more information, see Access KMS instance keys over the Internet.

   KMS keys are only accessible through the VPC network by default.

2. Configure the ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variables. For instructions on how to obtain the AccessKey pair, see How do I view the AccessKey pair of an account?.

3. In the target IntelliJ IDEA project, create a class file named EncryptFlink.

4. Copy and paste the following code snippet to the EncryptFlink class file. Remember to modify the values of the configuration options to match your specific setup.

   package org.example;
   
   import com.aliyun.kms20160120.models.EncryptResponse;
   import com.aliyun.kms20160120.models.EncryptResponseBody;
   import com.aliyun.tea.*;
   
   public class EncryptFlink {
   
       /**
        * <b>description</b> :
        * <p>Use your AccessKey pair to initialize the client.</p>
        * @return Client
        *
        * @throws Exception
        */
       public static com.aliyun.kms20160120.Client createClient() throws Exception {
           // If the project code is leaked, the AccessKey pair may be leaked, compromising the security of all resources within your account. The following sample code is provided only for reference. 
           com.aliyun.teaopenapi.models.Config config = new com.aliyun.teaopenapi.models.Config()
                   // Required. Make sure that the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is configured. 
                   .setAccessKeyId(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"))
                   // Required. Make sure that the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is configured. 
                   .setAccessKeySecret(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
           // Specify an endpoint. For more information, visit 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 {
               // Write your own code to display the response of the API operation if necessary.
               EncryptResponse encryptResponse = client.encryptWithOptions(encryptRequest, runtime);
               EncryptResponseBody body = encryptResponse.getBody();
               System.out.println(body.getCiphertextBlob());
           } catch (TeaException error) {
               // Handle exceptions with caution in actual business scenarios and do not ignore the exceptions in your project. In this example, error messages are simply printed for demonstration purposes. 
               // The error message.
               System.out.println(error.getMessage());
               // The URL for troubleshooting.
               System.out.println(error.getData().get("Recommend"));
               com.aliyun.teautil.Common.assertAsString(error.message);
           } catch (Exception _error) {
               TeaException error = new TeaException(_error.getMessage(), _error);
               // Handle exceptions with caution in actual business scenarios and do not ignore the exceptions in your project. In this example, error messages are simply printed for demonstration purposes. 
               // The error message.
               System.out.println(error.getMessage());
               // The URL for troubleshooting.
               System.out.println(error.getData().get("Recommend"));
               com.aliyun.teautil.Common.assertAsString(error.message);
           }
       }
   }

   Option	Description	Example
   config.endpoint	The endpoint of your KMS instance.	kms.cn-hangzhou.aliyuncs.com
   Plaintext	The plaintext password to encrypt.	flink_rds_password@123
   KeyId	The KMS key ID.	key-hzz67ab1ff4e750h****

5. Add the following dependencies to the POM.xml file.

       <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. Run the EncryptFlink class file to obtain the 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();
       
        // Parse parameters to get the AccessKey pair.
        final ParameterTool params = ParameterTool.fromArgs(args);
        String ak = params.get("akid");
        String sk = params.get("aksecret");
        
        // Decrypt the encrypted password.
        String encryptedPassword = "a2V5LWh6ejY3YTJmZTY5ejR2NTlpOHE1MC03d3ozYWU1anlzC6NLoC0JHHEXTJ4P/4iVOe/B+eniv8EcaviQDzZWWNPedOYkoFFYWA==";
        String decryptedPassword = KmsUtil.decrypt(ak, sk, encryptedPassword);

        // Build a 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();

        // Configure the data source (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") // Specify the database.
                        .tableList("school.student") // Specify the table.
                        .username("flink_rds_user")
                        .password(decryptedPassword)
                        // Initialize data in the RowData structure.
                        .deserializer(deserializer)
                        .build();

        // Integrate the external data source into the Flink DataStream program
        // Do not use a watermark strategy.
        DataStreamSource<RowData> mySQLSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source");

        // Write to stdout.
        mySQLSource.print();

        // Execute the program.
        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 core dependencies (Set the scope to provided when packaging the 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>
            <! -- Comment out the following line for local execution -->
            <!-- <scope>provided</scope> -->
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-runtime</artifactId>
            <version>1.17.1</version> <! -- The version should be consistent with the Flink version you specified earlier -->
            <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> <! -- The version should be consistent with the Flink version you specified earlier -->
            <scope>provided</scope>
        </dependency>


        <! -- Log dependencies -->
        <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>
            <! -- Compiler plug-in -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.13.0</version> <! -- Patch version -->
                <configuration>
                    <source>${maven.compiler.source}</source>
                    <target>${maven.compiler.target}</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>

            <! -- Plug-in for building a 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>
                                    <! -- Retain logging dependencies -->
                                    <!-- <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> <! -- Add a key filter -->
                                    </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"/> <! -- Add a necessary transformer -->
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>