Repository: provectus/kafka-ui
Issue: Add authentication for Schema Registry
Labels: type/enhancement
scope/backend

It's not possible to add authentication for Schema Registry, so I get this error:

Failed to get Schema for topic appuntamento-claimgen

io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Unexpected character ('<' (code 60)): expected a valid value (JSON String, Number, Array, Object or token 'null', 'true' or 'false')

at [Source: (sun.net.www.protocol.http.HttpURLConnection$HttpInputStream); line: 1, column: 2]; error code: 50005

at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:292) ~[kafka-schema-registry-client-5.5.1.jar!/:?]

Would be appreciated if it will be possibile to add env variables to add auth for Schema Registry.

Issue comments:

Pull request review comments:
url?
could we support both object and string scenarios?
could we check only user & pass fields? why do we need additional boolean for this?
could we set it on webclient level?
Here we can make requests for different clusters and in each cluster can be it's own Schema Registry with it's own settings. So we can't use pre configured web client
Could you please move these properties to constants or replace them with existing ones?
I added method which returns configured WebClient
Resolved
I meant `basic.auth.user.info` and `basic.auth.credentials.source`
Done.
I used constants stored in `io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig`.

Patch that resolved the issue:
diff --git a/README.md b/README.md
index 6feb09572ae..927b67b2800 100644
--- a/README.md
+++ b/README.md
@@ -132,6 +132,9 @@ kafka:
       bootstrapServers: localhost:29091
       zookeeper: localhost:2183
       schemaRegistry: http://localhost:8085
+      schemaRegistryAuth:
+        username: username
+        password: password
 #     schemaNameTemplate: "%s-value"
       jmxPort: 9997
     -
@@ -141,6 +144,8 @@ kafka:
 * `bootstrapServers`: where to connect
 * `zookeeper`: zookeeper service address
 * `schemaRegistry`: schemaRegistry's address
+* `schemaRegistryAuth.username`: schemaRegistry's basic authentication username
+* `schemaRegistryAuth.password`: schemaRegistry's basic authentication password
 * `schemaNameTemplate`: how keys are saved to schemaRegistry
 * `jmxPort`: open jmxPosrts of a broker
 * `readOnly`: enable read only mode
@@ -160,6 +165,8 @@ For example, if you want to use an environment variable to set the `name` parame
 |`KAFKA_CLUSTERS_0_ZOOKEEPER` 	| Zookeper service address 
 |`KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL` 	|Security protocol to connect to the brokers. For SSL connection use "SSL", for plaintext connection don't set this environment variable
 |`KAFKA_CLUSTERS_0_SCHEMAREGISTRY`   	|SchemaRegistry's address
+|`KAFKA_CLUSTERS_0_SCHEMAREGISTRYAUTH_USERNAME`   	|SchemaRegistry's basic authentication username
+|`KAFKA_CLUSTERS_0_SCHEMAREGISTRYAUTH_PASSWORD`   	|SchemaRegistry's basic authentication password
 |`KAFKA_CLUSTERS_0_SCHEMANAMETEMPLATE`  |How keys are saved to schemaRegistry
 |`KAFKA_CLUSTERS_0_JMXPORT`        	|Open jmxPosrts of a broker
 |`KAFKA_CLUSTERS_0_READONLY`        	|Enable read only mode. Default: false
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java
index 82954dff363..f847ca2a80d 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java
@@ -20,6 +20,7 @@ public static class Cluster {
     String bootstrapServers;
     String zookeeper;
     String schemaRegistry;
+    SchemaRegistryAuth schemaRegistryAuth;
     String schemaNameTemplate = "%s-value";
     String keySchemaNameTemplate = "%s-key";
     String protobufFile;
@@ -35,4 +36,10 @@ public static class ConnectCluster {
     String name;
     String address;
   }
+
+  @Data
+  public static class SchemaRegistryAuth {
+    String username;
+    String password;
+  }
 }
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/mapper/ClusterMapper.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/mapper/ClusterMapper.java
index 766f3f24b32..321a6c08713 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/mapper/ClusterMapper.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/mapper/ClusterMapper.java
@@ -15,6 +15,7 @@
 import com.provectus.kafka.ui.model.InternalClusterMetrics;
 import com.provectus.kafka.ui.model.InternalPartition;
 import com.provectus.kafka.ui.model.InternalReplica;
+import com.provectus.kafka.ui.model.InternalSchemaRegistry;
 import com.provectus.kafka.ui.model.InternalTopic;
 import com.provectus.kafka.ui.model.InternalTopicConfig;
 import com.provectus.kafka.ui.model.KafkaCluster;
@@ -49,6 +50,7 @@ public interface ClusterMapper {
 
   @Mapping(target = "protobufFile", source = "protobufFile", qualifiedByName = "resolvePath")
   @Mapping(target = "properties", source = "properties", qualifiedByName = "setProperties")
+  @Mapping(target = "schemaRegistry", source = ".", qualifiedByName = "setSchemaRegistry")
   KafkaCluster toKafkaCluster(ClustersProperties.Cluster clusterProperties);
 
   @Mapping(target = "diskUsage", source = "internalBrokerDiskUsage",
@@ -64,6 +66,24 @@ public interface ClusterMapper {
 
   Partition toPartition(InternalPartition topic);
 
+  default InternalSchemaRegistry setSchemaRegistry(ClustersProperties.Cluster clusterProperties) {
+    if (clusterProperties == null) {
+      return null;
+    }
+
+    InternalSchemaRegistry.InternalSchemaRegistryBuilder internalSchemaRegistry =
+        InternalSchemaRegistry.builder();
+
+    internalSchemaRegistry.url(clusterProperties.getSchemaRegistry());
+
+    if (clusterProperties.getSchemaRegistryAuth() != null) {
+      internalSchemaRegistry.username(clusterProperties.getSchemaRegistryAuth().getUsername());
+      internalSchemaRegistry.password(clusterProperties.getSchemaRegistryAuth().getPassword());
+    }
+
+    return internalSchemaRegistry.build();
+  }
+
   TopicDetails toTopicDetails(InternalTopic topic);
 
   default TopicDetails toTopicDetails(InternalTopic topic, InternalClusterMetrics metrics) {
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalSchemaRegistry.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalSchemaRegistry.java
new file mode 100644
index 00000000000..378f2706f58
--- /dev/null
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalSchemaRegistry.java
@@ -0,0 +1,12 @@
+package com.provectus.kafka.ui.model;
+
+import lombok.Builder;
+import lombok.Data;
+
+@Data
+@Builder(toBuilder = true)
+public class InternalSchemaRegistry {
+  private final String username;
+  private final String password;
+  private final String url;
+}
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/KafkaCluster.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/KafkaCluster.java
index 805f0937e96..68b16073761 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/KafkaCluster.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/KafkaCluster.java
@@ -15,7 +15,7 @@ public class KafkaCluster {
   private final Integer jmxPort;
   private final String bootstrapServers;
   private final String zookeeper;
-  private final String schemaRegistry;
+  private final InternalSchemaRegistry schemaRegistry;
   private final List<KafkaConnectCluster> kafkaConnect;
   private final String schemaNameTemplate;
   private final String keySchemaNameTemplate;
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/schemaregistry/SchemaRegistryAwareRecordSerDe.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/schemaregistry/SchemaRegistryAwareRecordSerDe.java
index 08aef7455ff..03f1431b82d 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/schemaregistry/SchemaRegistryAwareRecordSerDe.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/schemaregistry/SchemaRegistryAwareRecordSerDe.java
@@ -1,6 +1,11 @@
 package com.provectus.kafka.ui.serde.schemaregistry;
 
+
+import static io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig.BASIC_AUTH_CREDENTIALS_SOURCE;
+import static io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig.USER_INFO_CONFIG;
+
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.provectus.kafka.ui.exception.ValidationException;
 import com.provectus.kafka.ui.model.KafkaCluster;
 import com.provectus.kafka.ui.model.MessageSchema;
 import com.provectus.kafka.ui.model.TopicMessageSchema;
@@ -22,6 +27,7 @@
 import java.net.URI;
 import java.nio.ByteBuffer;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -63,14 +69,29 @@ public class SchemaRegistryAwareRecordSerDe implements RecordSerDe {
 
   private static SchemaRegistryClient createSchemaRegistryClient(KafkaCluster cluster) {
     Objects.requireNonNull(cluster.getSchemaRegistry());
+    Objects.requireNonNull(cluster.getSchemaRegistry().getUrl());
     List<SchemaProvider> schemaProviders =
         List.of(new AvroSchemaProvider(), new ProtobufSchemaProvider(), new JsonSchemaProvider());
-    //TODO add auth
+
+    Map<String, String> configs = new HashMap<>();
+    String username = cluster.getSchemaRegistry().getUsername();
+    String password = cluster.getSchemaRegistry().getPassword();
+
+    if (username != null && password != null) {
+      configs.put(BASIC_AUTH_CREDENTIALS_SOURCE, "USER_INFO");
+      configs.put(USER_INFO_CONFIG, username + ":" + password);
+    } else if (username != null) {
+      throw new ValidationException(
+          "You specified username but do not specified password");
+    } else if (password != null) {
+      throw new ValidationException(
+          "You specified password but do not specified username");
+    }
     return new CachedSchemaRegistryClient(
-        Collections.singletonList(cluster.getSchemaRegistry()),
+        Collections.singletonList(cluster.getSchemaRegistry().getUrl()),
         CLIENT_IDENTITY_MAP_CAPACITY,
         schemaProviders,
-        Collections.emptyMap()
+        configs
     );
   }
 
@@ -181,7 +202,8 @@ public TopicMessageSchema getTopicSchema(String topic) {
   private String convertSchema(SchemaMetadata schema) {
 
     String jsonSchema;
-    URI basePath = new URI(cluster.getSchemaRegistry()).resolve(Integer.toString(schema.getId()));
+    URI basePath = new URI(cluster.getSchemaRegistry().getUrl())
+        .resolve(Integer.toString(schema.getId()));
     final ParsedSchema schemaById = Objects.requireNonNull(schemaRegistryClient)
         .getSchemaById(schema.getId());
 
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/SchemaRegistryService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/SchemaRegistryService.java
index 41374120b81..611d6eedf42 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/SchemaRegistryService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/SchemaRegistryService.java
@@ -7,9 +7,11 @@
 import com.provectus.kafka.ui.exception.DuplicateEntityException;
 import com.provectus.kafka.ui.exception.SchemaNotFoundException;
 import com.provectus.kafka.ui.exception.UnprocessableEntityException;
+import com.provectus.kafka.ui.exception.ValidationException;
 import com.provectus.kafka.ui.mapper.ClusterMapper;
 import com.provectus.kafka.ui.model.CompatibilityCheckResponse;
 import com.provectus.kafka.ui.model.CompatibilityLevel;
+import com.provectus.kafka.ui.model.InternalSchemaRegistry;
 import com.provectus.kafka.ui.model.KafkaCluster;
 import com.provectus.kafka.ui.model.NewSchemaSubject;
 import com.provectus.kafka.ui.model.SchemaSubject;
@@ -26,6 +28,8 @@
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
 import org.jetbrains.annotations.NotNull;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Service;
@@ -61,8 +65,10 @@ public Flux<SchemaSubject> getAllLatestVersionSchemas(String clusterName) {
 
   public Mono<String[]> getAllSubjectNames(String clusterName) {
     return clustersStorage.getClusterByName(clusterName)
-        .map(cluster -> webClient.get()
-            .uri(cluster.getSchemaRegistry() + URL_SUBJECTS)
+        .map(cluster -> configuredWebClient(
+            cluster,
+            HttpMethod.GET,
+            URL_SUBJECTS)
             .retrieve()
             .bodyToMono(String[].class)
             .doOnError(log::error)
@@ -77,8 +83,10 @@ public Flux<SchemaSubject> getAllVersionsBySubject(String clusterName, String su
 
   private Flux<Integer> getSubjectVersions(String clusterName, String schemaName) {
     return clustersStorage.getClusterByName(clusterName)
-        .map(cluster -> webClient.get()
-            .uri(cluster.getSchemaRegistry() + URL_SUBJECT_VERSIONS, schemaName)
+        .map(cluster -> configuredWebClient(
+            cluster,
+            HttpMethod.GET,
+            URL_SUBJECT_VERSIONS, schemaName)
             .retrieve()
             .onStatus(NOT_FOUND::equals,
                 throwIfNotFoundStatus(formatted(NO_SUCH_SCHEMA, schemaName))
@@ -99,8 +107,10 @@ public Mono<SchemaSubject> getLatestSchemaVersionBySubject(String clusterName,
   private Mono<SchemaSubject> getSchemaSubject(String clusterName, String schemaName,
                                                String version) {
     return clustersStorage.getClusterByName(clusterName)
-        .map(cluster -> webClient.get()
-            .uri(cluster.getSchemaRegistry() + URL_SUBJECT_BY_VERSION, schemaName, version)
+        .map(cluster -> configuredWebClient(
+            cluster,
+            HttpMethod.GET,
+            URL_SUBJECT_BY_VERSION, schemaName, version)
             .retrieve()
             .onStatus(NOT_FOUND::equals,
                 throwIfNotFoundStatus(formatted(NO_SUCH_SCHEMA_VERSION, schemaName, version))
@@ -140,8 +150,10 @@ public Mono<ResponseEntity<Void>> deleteLatestSchemaSubject(String clusterName,
   private Mono<ResponseEntity<Void>> deleteSchemaSubject(String clusterName, String schemaName,
                                                          String version) {
     return clustersStorage.getClusterByName(clusterName)
-        .map(cluster -> webClient.delete()
-            .uri(cluster.getSchemaRegistry() + URL_SUBJECT_BY_VERSION, schemaName, version)
+        .map(cluster -> configuredWebClient(
+            cluster,
+            HttpMethod.DELETE,
+            URL_SUBJECT_BY_VERSION, schemaName, version)
             .retrieve()
             .onStatus(NOT_FOUND::equals,
                 throwIfNotFoundStatus(formatted(NO_SUCH_SCHEMA_VERSION, schemaName, version))
@@ -152,8 +164,10 @@ private Mono<ResponseEntity<Void>> deleteSchemaSubject(String clusterName, Strin
   public Mono<ResponseEntity<Void>> deleteSchemaSubjectEntirely(String clusterName,
                                                                 String schemaName) {
     return clustersStorage.getClusterByName(clusterName)
-        .map(cluster -> webClient.delete()
-            .uri(cluster.getSchemaRegistry() + URL_SUBJECT, schemaName)
+        .map(cluster -> configuredWebClient(
+            cluster,
+            HttpMethod.DELETE,
+            URL_SUBJECT, schemaName)
             .retrieve()
             .onStatus(NOT_FOUND::equals,
                 throwIfNotFoundStatus(formatted(NO_SUCH_SCHEMA, schemaName))
@@ -178,8 +192,8 @@ public Mono<SchemaSubject> registerNewSchema(String clusterName,
           return clustersStorage.getClusterByName(clusterName)
               .map(KafkaCluster::getSchemaRegistry)
               .map(
-                  schemaRegistryUrl -> checkSchemaOnDuplicate(subject, newSchema, schemaRegistryUrl)
-                      .flatMap(s -> submitNewSchema(subject, newSchema, schemaRegistryUrl))
+                  schemaRegistry -> checkSchemaOnDuplicate(subject, newSchema, schemaRegistry)
+                      .flatMap(s -> submitNewSchema(subject, newSchema, schemaRegistry))
                       .flatMap(resp -> getLatestSchemaVersionBySubject(clusterName, subject))
               )
               .orElse(Mono.error(ClusterNotFoundException::new));
@@ -189,9 +203,11 @@ public Mono<SchemaSubject> registerNewSchema(String clusterName,
   @NotNull
   private Mono<SubjectIdResponse> submitNewSchema(String subject,
                                                   Mono<InternalNewSchema> newSchemaSubject,
-                                                  String schemaRegistryUrl) {
-    return webClient.post()
-        .uri(schemaRegistryUrl + URL_SUBJECT_VERSIONS, subject)
+                                                  InternalSchemaRegistry schemaRegistry) {
+    return configuredWebClient(
+        schemaRegistry,
+        HttpMethod.POST,
+        URL_SUBJECT_VERSIONS, subject)
         .contentType(MediaType.APPLICATION_JSON)
         .body(BodyInserters.fromPublisher(newSchemaSubject, InternalNewSchema.class))
         .retrieve()
@@ -204,9 +220,11 @@ private Mono<SubjectIdResponse> submitNewSchema(String subject,
   @NotNull
   private Mono<SchemaSubject> checkSchemaOnDuplicate(String subject,
                                                      Mono<InternalNewSchema> newSchemaSubject,
-                                                     String schemaRegistryUrl) {
-    return webClient.post()
-        .uri(schemaRegistryUrl + URL_SUBJECT, subject)
+                                                     InternalSchemaRegistry schemaRegistry) {
+    return configuredWebClient(
+        schemaRegistry,
+        HttpMethod.POST,
+        URL_SUBJECT, subject)
         .contentType(MediaType.APPLICATION_JSON)
         .body(BodyInserters.fromPublisher(newSchemaSubject, InternalNewSchema.class))
         .retrieve()
@@ -236,8 +254,10 @@ public Mono<Void> updateSchemaCompatibility(String clusterName, String schemaNam
     return clustersStorage.getClusterByName(clusterName)
         .map(cluster -> {
           String configEndpoint = Objects.isNull(schemaName) ? "/config" : "/config/{schemaName}";
-          return webClient.put()
-              .uri(cluster.getSchemaRegistry() + configEndpoint, schemaName)
+          return configuredWebClient(
+              cluster,
+              HttpMethod.PUT,
+              configEndpoint, schemaName)
               .contentType(MediaType.APPLICATION_JSON)
               .body(BodyInserters.fromPublisher(compatibilityLevel, CompatibilityLevel.class))
               .retrieve()
@@ -257,8 +277,10 @@ public Mono<CompatibilityLevel> getSchemaCompatibilityLevel(String clusterName,
     return clustersStorage.getClusterByName(clusterName)
         .map(cluster -> {
           String configEndpoint = Objects.isNull(schemaName) ? "/config" : "/config/{schemaName}";
-          return webClient.get()
-              .uri(cluster.getSchemaRegistry() + configEndpoint, schemaName)
+          return configuredWebClient(
+              cluster,
+              HttpMethod.GET,
+              configEndpoint, schemaName)
               .retrieve()
               .bodyToMono(InternalCompatibilityLevel.class)
               .map(mapper::toCompatibilityLevel)
@@ -279,9 +301,10 @@ private Mono<CompatibilityLevel> getSchemaCompatibilityInfoOrGlobal(String clust
   public Mono<CompatibilityCheckResponse> checksSchemaCompatibility(
       String clusterName, String schemaName, Mono<NewSchemaSubject> newSchemaSubject) {
     return clustersStorage.getClusterByName(clusterName)
-        .map(cluster -> webClient.post()
-            .uri(cluster.getSchemaRegistry()
-                + "/compatibility/subjects/{schemaName}/versions/latest", schemaName)
+        .map(cluster -> configuredWebClient(
+            cluster,
+            HttpMethod.POST,
+            "/compatibility/subjects/{schemaName}/versions/latest", schemaName)
             .contentType(MediaType.APPLICATION_JSON)
             .body(BodyInserters.fromPublisher(newSchemaSubject, NewSchemaSubject.class))
             .retrieve()
@@ -296,4 +319,32 @@ public Mono<CompatibilityCheckResponse> checksSchemaCompatibility(
   public String formatted(String str, Object... args) {
     return new Formatter().format(str, args).toString();
   }
+
+  private void setBasicAuthIfEnabled(InternalSchemaRegistry schemaRegistry, HttpHeaders headers) {
+    if (schemaRegistry.getUsername() != null && schemaRegistry.getPassword() != null) {
+      headers.setBasicAuth(
+          schemaRegistry.getUsername(),
+          schemaRegistry.getPassword()
+      );
+    } else if (schemaRegistry.getUsername() != null) {
+      throw new ValidationException(
+          "You specified username but do not specified password");
+    } else if (schemaRegistry.getPassword() != null) {
+      throw new ValidationException(
+          "You specified password but do not specified username");
+    }
+  }
+
+  private WebClient.RequestBodySpec configuredWebClient(KafkaCluster cluster, HttpMethod method,
+                                                        String uri, Object... params) {
+    return configuredWebClient(cluster.getSchemaRegistry(), method, uri, params);
+  }
+
+  private WebClient.RequestBodySpec configuredWebClient(InternalSchemaRegistry schemaRegistry,
+                                                        HttpMethod method, String uri,
+                                                        Object... params) {
+    return webClient.method(method)
+        .uri(schemaRegistry.getUrl() + uri, params)
+        .headers(headers -> setBasicAuthIfEnabled(schemaRegistry, headers));
+  }
 }


Test patch: