Sanitize credentials in Kafka Connector configs

* #980: sanitize credentials in Kafka Connector configs. Add a warning to edit page when people try to edit configs with obfuscated field

* #980: make sanitizer patterns configurable

* #980 fix using stream pattern

Co-authored-by: Si Tang <si@indeed.com>

kafka-ui-api/pom.xml

@@ -37,6 +37,10 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-security</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-actuator</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-oauth2-client</artifactId>	

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaConfigSanitizer.java

@@ -0,0 +1,36 @@
+package com.provectus.kafka.ui.service;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.config.SslConfigs;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.actuate.endpoint.Sanitizer;
+import org.springframework.stereotype.Component;
+
+@Component
+class KafkaConfigSanitizer extends Sanitizer {
+  private static final List<String> DEFAULT_PATTERNS_TO_SANITIZE = Arrays.asList(
+      "basic.auth.user.info",  /* For Schema Registry credentials */
+      "password", "secret", "token", "key", ".*credentials.*"  /* General credential patterns */
+  );
+
+  KafkaConfigSanitizer(
+          @Value("${kafka.config.sanitizer.patterns:}") List<String> patternsToSanitize
+  ) {
+    final ConfigDef configDef = new ConfigDef();
+    SslConfigs.addClientSslSupport(configDef);
+    SaslConfigs.addClientSaslSupport(configDef);
+    final Set<String> keysToSanitize = configDef.configKeys().entrySet().stream()
+            .filter(entry -> entry.getValue().type().equals(ConfigDef.Type.PASSWORD))
+            .map(Map.Entry::getKey)
+            .collect(Collectors.toSet());
+    keysToSanitize.addAll(
+            patternsToSanitize.isEmpty() ? DEFAULT_PATTERNS_TO_SANITIZE : patternsToSanitize);
+    this.setKeysToSanitize(keysToSanitize.toArray(new String[0]));
+  }
+}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaConnectService.java

@@ -23,6 +23,7 @@ import com.provectus.kafka.ui.model.KafkaConnectCluster;
 import com.provectus.kafka.ui.model.NewConnectorDTO;
 import com.provectus.kafka.ui.model.TaskDTO;
 import com.provectus.kafka.ui.model.connect.InternalConnectInfo;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
@@ -47,6 +48,7 @@ public class KafkaConnectService {
   private final ClusterMapper clusterMapper;
   private final KafkaConnectMapper kafkaConnectMapper;
   private final ObjectMapper objectMapper;
+  private final KafkaConfigSanitizer kafkaConfigSanitizer;
 
   public Mono<Flux<ConnectDTO>> getConnects(KafkaCluster cluster) {
     return Mono.just(
@@ -187,13 +189,19 @@ public class KafkaConnectService {
                         e -> emptyStatus(connectorName))
                     .map(connectorStatus -> {
                       var status = connectorStatus.getConnector();
+                      final Map<String, Object> obfuscatedConfig = connector.getConfig().entrySet()
+                          .stream()
+                          .collect(Collectors.toMap(
+                              Map.Entry::getKey,
+                              e -> kafkaConfigSanitizer.sanitize(e.getKey(), e.getValue())
+                          ));
                       ConnectorDTO result = (ConnectorDTO) new ConnectorDTO()
                           .connect(connectName)
                           .status(kafkaConnectMapper.fromClient(status))
                           .type(connector.getType())
                           .tasks(connector.getTasks())
                           .name(connector.getName())
-                          .config(connector.getConfig());
+                          .config(obfuscatedConfig);
 
                       if (connectorStatus.getTasks() != null) {
                         boolean isAnyTaskFailed = connectorStatus.getTasks().stream()
@@ -223,7 +231,13 @@ public class KafkaConnectService {
     return getConnectAddress(cluster, connectName)
         .flatMap(connect ->
             KafkaConnectClients.withBaseUrl(connect).getConnectorConfig(connectorName)
-        );
+        )
+        .map(connectorConfig -> {
+          final Map<String, Object> obfuscatedMap = new HashMap<>();
+          connectorConfig.forEach((key, value) ->
+              obfuscatedMap.put(key, kafkaConfigSanitizer.sanitize(key, value)));
+          return obfuscatedMap;
+        });
   }
 
   public Mono<ConnectorDTO> setConnectorConfig(KafkaCluster cluster, String connectName,

kafka-ui-api/src/test/java/com/provectus/kafka/ui/KafkaConnectServiceTests.java

@@ -37,7 +37,8 @@ public class KafkaConnectServiceTests extends AbstractBaseTest {
       "connector.class", "org.apache.kafka.connect.file.FileStreamSinkConnector",
       "tasks.max", "1",
       "topics", "output-topic",
-      "file", "/tmp/test"
+      "file", "/tmp/test",
+      "test.password", "******"
   );
 
   @Autowired
@@ -54,7 +55,8 @@ public class KafkaConnectServiceTests extends AbstractBaseTest {
                 "connector.class", "org.apache.kafka.connect.file.FileStreamSinkConnector",
                 "tasks.max", "1",
                 "topics", "output-topic",
-                "file", "/tmp/test"
+                "file", "/tmp/test",
+                "test.password", "test-credentials"
             ))
         )
         .exchange()
@@ -296,7 +298,8 @@ public class KafkaConnectServiceTests extends AbstractBaseTest {
             "tasks.max", "1",
             "topics", "output-topic",
             "file", "/tmp/test",
-            "name", connectorName
+            "name", connectorName,
+            "test.password", "******"
         ));
   }
 
@@ -324,6 +327,7 @@ public class KafkaConnectServiceTests extends AbstractBaseTest {
             "tasks.max", "1",
             "topics", "output-topic",
             "file", "/tmp/test",
+            "test.password", "******",
             "name", connectorName
         ));
   }

kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/KafkaConfigSanitizerTest.java

@@ -0,0 +1,41 @@
+package com.provectus.kafka.ui.service;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Arrays;
+import java.util.Collections;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.actuate.endpoint.Sanitizer;
+
+class KafkaConfigSanitizerTest {
+
+  @Test
+  void obfuscateCredentials() {
+    final Sanitizer sanitizer = new KafkaConfigSanitizer(Collections.emptyList());
+    assertThat(sanitizer.sanitize("sasl.jaas.config", "secret")).isEqualTo("******");
+    assertThat(sanitizer.sanitize("consumer.sasl.jaas.config", "secret")).isEqualTo("******");
+    assertThat(sanitizer.sanitize("producer.sasl.jaas.config", "secret")).isEqualTo("******");
+    assertThat(sanitizer.sanitize("main.consumer.sasl.jaas.config", "secret")).isEqualTo("******");
+    assertThat(sanitizer.sanitize("database.password", "secret")).isEqualTo("******");
+    assertThat(sanitizer.sanitize("basic.auth.user.info", "secret")).isEqualTo("******");
+  }
+
+  @Test
+  void notObfuscateNormalConfigs() {
+    final Sanitizer sanitizer = new KafkaConfigSanitizer(Collections.emptyList());
+    assertThat(sanitizer.sanitize("security.protocol", "SASL_SSL")).isEqualTo("SASL_SSL");
+    final String[] bootstrapServer = new String[] {"test1:9092", "test2:9092"};
+    assertThat(sanitizer.sanitize("bootstrap.servers", bootstrapServer)).isEqualTo(bootstrapServer);
+  }
+
+  @Test
+  void obfuscateCredentialsWithDefinedPatterns() {
+    final Sanitizer sanitizer = new KafkaConfigSanitizer(Arrays.asList("kafka.ui", ".*test.*"));
+    assertThat(sanitizer.sanitize("consumer.kafka.ui", "secret")).isEqualTo("******");
+    assertThat(sanitizer.sanitize("this.is.test.credentials", "secret")).isEqualTo("******");
+    assertThat(sanitizer.sanitize("this.is.not.credential", "not.credential"))
+            .isEqualTo("not.credential");
+    assertThat(sanitizer.sanitize("database.password", "no longer credential"))
+            .isEqualTo("no longer credential");
+  }
+}

kafka-ui-react-app/src/components/Connect/Edit/Edit.tsx

@@ -99,34 +99,45 @@ const Edit: React.FC<EditProps> = ({
 
   if (isConfigFetching) return <PageLoader />;
 
+  const hasCredentials = JSON.stringify(config, null, '\t').includes(
+    '"******"'
+  );
   return (
-    <div className="box">
-      <form onSubmit={handleSubmit(onSubmit)}>
-        <div className="field">
-          <div className="control">
-            <Controller
-              control={control}
-              name="config"
-              render={({ field }) => (
-                <JSONEditor {...field} readOnly={isSubmitting} />
-              )}
-            />
-          </div>
-          <p className="help is-danger">
-            <ErrorMessage errors={errors} name="config" />
-          </p>
+    <>
+      {hasCredentials && (
+        <div className="notification is-danger is-light">
+          Please replace ****** with the real credential values to avoid
+          accidentally breaking your connector config!
         </div>
-        <div className="field">
-          <div className="control">
-            <input
-              type="submit"
-              className="button is-primary"
-              disabled={!isValid || isSubmitting || !isDirty}
-            />
+      )}
+      <div className="box">
+        <form onSubmit={handleSubmit(onSubmit)}>
+          <div className="field">
+            <div className="control">
+              <Controller
+                control={control}
+                name="config"
+                render={({ field }) => (
+                  <JSONEditor {...field} readOnly={isSubmitting} />
+                )}
+              />
+            </div>
+            <p className="help is-danger">
+              <ErrorMessage errors={errors} name="config" />
+            </p>
           </div>
-        </div>
-      </form>
-    </div>
+          <div className="field">
+            <div className="control">
+              <input
+                type="submit"
+                className="button is-primary"
+                disabled={!isValid || isSubmitting || !isDirty}
+              />
+            </div>
+          </div>
+        </form>
+      </div>
+    </>
   );
 };
 

kafka-ui-react-app/src/components/Connect/Edit/__tests__/Edit.spec.tsx

@@ -61,6 +61,13 @@ describe('Edit', () => {
       expect(wrapper.toJSON()).toMatchSnapshot();
     });
 
+    it('matches snapshot when config has credentials', () => {
+      const wrapper = create(
+        setupWrapper({ config: { ...connector.config, password: '******' } })
+      );
+      expect(wrapper.toJSON()).toMatchSnapshot();
+    });
+
     it('fetches config on mount', () => {
       const fetchConfig = jest.fn();
       mount(setupWrapper({ fetchConfig }));

kafka-ui-react-app/src/components/Connect/Edit/__tests__/__snapshots__/Edit.spec.tsx.snap

@@ -47,4 +47,59 @@ exports[`Edit view matches snapshot 1`] = `
 </div>
 `;
 
+exports[`Edit view matches snapshot when config has credentials 1`] = `
+Array [
+  <div
+    className="notification is-danger is-light"
+  >
+    Please replace ****** with the real credential values to avoid accidentally breaking your connector config!
+  </div>,
+  <div
+    className="box"
+  >
+    <form
+      onSubmit={[Function]}
+    >
+      <div
+        className="field"
+      >
+        <div
+          className="control"
+        >
+          <mock-JSONEditor
+            name="config"
+            onBlur={[Function]}
+            onChange={[Function]}
+            readOnly={false}
+            value="{
+	\\"connector.class\\": \\"FileStreamSource\\",
+	\\"tasks.max\\": \\"10\\",
+	\\"topic\\": \\"test-topic\\",
+	\\"file\\": \\"/some/file\\",
+	\\"password\\": \\"******\\"
+}"
+          />
+        </div>
+        <p
+          className="help is-danger"
+        />
+      </div>
+      <div
+        className="field"
+      >
+        <div
+          className="control"
+        >
+          <input
+            className="button is-primary"
+            disabled={true}
+            type="submit"
+          />
+        </div>
+      </div>
+    </form>
+  </div>,
+]
+`;
+
 exports[`Edit view matches snapshot when fetching config 1`] = `<mock-PageLoader />`;