新增kafka用户名密码加密 TSG-8835

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.zdjizhi</groupId>
     <artifactId>log-completion-schema</artifactId>
-    <version>220308-IngestionTime</version>
+    <version>220316-encryption</version>
 
     <name>log-completion-schema</name>
     <url>http://www.example.com</url>
@@ -37,6 +37,7 @@
         <hadoop.version>2.7.1</hadoop.version>
         <kafka.version>1.0.0</kafka.version>
         <hbase.version>2.2.3</hbase.version>
+        <nacos.version>1.4.1</nacos.version>
         <scope.type>provided</scope.type>
         <!--<scope.type>compile</scope.type>-->
     </properties>
@@ -287,6 +288,20 @@
             <scope>test</scope>
         </dependency>
 
+        <!-- https://mvnrepository.com/artifact/com.alibaba.nacos/nacos-client -->
+        <dependency>
+            <groupId>com.alibaba.nacos</groupId>
+            <artifactId>nacos-client</artifactId>
+            <version>${nacos.version}</version>
+        </dependency>
+
+        <!-- https://mvnrepository.com/artifact/org.jasypt/jasypt -->
+        <dependency>
+            <groupId>org.jasypt</groupId>
+            <artifactId>jasypt</artifactId>
+            <version>1.9.3</version>
+        </dependency>
+
     </dependencies>
 </project>
 

properties/default_config.properties

@@ -28,11 +28,11 @@ buffer.memory=134217728
 #10M
 max.request.size=10485760
 #====================kafka default====================#
-#kafka SASL验证用户名
+#kafka SASL验证用户名-加密
-kafka.user=admin
+kafka.user=nsyGpHKGFA4KW0zro9MDdw==
 
-#kafka SASL及SSL验证密码
+#kafka SASL及SSL验证密码-加密
-kafka.pin=galaxy2019
+kafka.pin=6MleDyA3Z73HSaXiKsDJ2k7Ys8YWLhEJ
 #====================Topology Default====================#
 
 #hbase table name
@@ -42,7 +42,7 @@ hbase.table.name=tsg_galaxy:relation_framedip_account
 mail.default.charset=UTF-8
 
 #0不做任何校验，1弱类型校验
-log.transform.type=0
+log.transform.type=1
 
 #两个输出之间的最大时间(单位milliseconds)
 buffer.timeout=5000

properties/service_flow_config.properties

@@ -17,7 +17,7 @@ hbase.zookeeper.servers=192.168.44.12:2181
 tools.library=D:\\workerspace\\dat\\
 
 #网关的schema位置
-schema.http=http://192.168.44.12:9999/metadata/schema/v1/fields/session_record
+schema.http=http://192.168.44.67:9999/metadata/schema/v1/fields/session_record
 
 #网关APP_ID 获取接口
 app.id.http=http://192.168.44.67:9999/open-api/appDicList

src/main/java/com/zdjizhi/common/FlowWriteConfig.java

@@ -2,12 +2,19 @@ package com.zdjizhi.common;
 
 
 import com.zdjizhi.utils.system.FlowWriteConfigurations;
+import org.jasypt.encryption.pbe.StandardPBEStringEncryptor;
 
 /**
  * @author Administrator
  */
 public class FlowWriteConfig {
 
+    private static StandardPBEStringEncryptor encryptor = new StandardPBEStringEncryptor();
+
+    static {
+        encryptor.setPassword("galaxy");
+    }
+
     public static final int IF_PARAM_LENGTH = 3;
     public static final String VISIBILITY = "disabled";
     public static final String FORMAT_SPLITTER = ",";
@@ -52,8 +59,9 @@ public class FlowWriteConfig {
     public static final String PRODUCER_ACK = FlowWriteConfigurations.getStringProperty(0, "producer.ack");
     public static final String TOOLS_LIBRARY = FlowWriteConfigurations.getStringProperty(0, "tools.library");
     public static final String PRODUCER_KAFKA_COMPRESSION_TYPE = FlowWriteConfigurations.getStringProperty(0, "producer.kafka.compression.type");
-    public static final String KAFKA_USER = FlowWriteConfigurations.getStringProperty(1, "kafka.user");
+    public static final String KAFKA_SASL_JAAS_USER = encryptor.decrypt(FlowWriteConfigurations.getStringProperty(1, "kafka.user"));
-    public static final String KAFKA_PIN = FlowWriteConfigurations.getStringProperty(1, "kafka.pin");
+    public static final String KAFKA_SASL_JAAS_PIN = encryptor.decrypt(FlowWriteConfigurations.getStringProperty(1, "kafka.pin"));
+
 
     /**
      * connection kafka
@@ -71,5 +79,4 @@ public class FlowWriteConfig {
     public static final String SCHEMA_HTTP = FlowWriteConfigurations.getStringProperty(0, "schema.http");
     public static final String APP_ID_HTTP = FlowWriteConfigurations.getStringProperty(0, "app.id.http");
 
-
 }

src/main/java/com/zdjizhi/utils/kafka/CertUtils.java

@@ -33,15 +33,15 @@ class CertUtils {
             properties.put("security.protocol", "SASL_PLAINTEXT");
             properties.put("sasl.mechanism", "PLAIN");
             properties.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username="
-                    + FlowWriteConfig.KAFKA_USER + " password=" + FlowWriteConfig.KAFKA_PIN + ";");
+                    + FlowWriteConfig.KAFKA_SASL_JAAS_USER + " password=" + FlowWriteConfig.KAFKA_SASL_JAAS_PIN + ";");
         } else if (servers.contains(SSL_PORT)) {
             properties.put("security.protocol", "SSL");
             properties.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
             properties.put("ssl.keystore.location", FlowWriteConfig.TOOLS_LIBRARY + "keystore.jks");
-            properties.put("ssl.keystore.password", FlowWriteConfig.KAFKA_PIN);
+            properties.put("ssl.keystore.password", FlowWriteConfig.KAFKA_SASL_JAAS_PIN);
             properties.put("ssl.truststore.location", FlowWriteConfig.TOOLS_LIBRARY + "truststore.jks");
-            properties.put("ssl.truststore.password", FlowWriteConfig.KAFKA_PIN);
+            properties.put("ssl.truststore.password", FlowWriteConfig.KAFKA_SASL_JAAS_PIN);
-            properties.put("ssl.key.password", FlowWriteConfig.KAFKA_PIN);
+            properties.put("ssl.key.password", FlowWriteConfig.KAFKA_SASL_JAAS_PIN);
         }
 
     }