BE: Add skip ssl for SR

api/src/main/java/io/kafbat/ui/KafkaUiApplication.java

@@ -1,10 +1,11 @@
 package io.kafbat.ui;
 
+import io.kafbat.ui.service.ssl.SkipSecurityProvider;
 import io.kafbat.ui.util.DynamicConfigOperations;
+import java.security.Security;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.autoconfigure.ldap.LdapAutoConfiguration;
 import org.springframework.boot.builder.SpringApplicationBuilder;
-import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.scheduling.annotation.EnableAsync;
 import org.springframework.scheduling.annotation.EnableScheduling;
 
@@ -17,8 +18,10 @@ public static void main(String[] args) {
     startApplication(args);
   }
 
-  public static ConfigurableApplicationContext startApplication(String[] args) {
-    return new SpringApplicationBuilder(KafkaUiApplication.class)
+  public static void startApplication(String[] args) {
+    Security.addProvider(new SkipSecurityProvider());
+
+    new SpringApplicationBuilder(KafkaUiApplication.class)
         .initializers(DynamicConfigOperations.dynamicConfigPropertiesInitializer())
         .build()
         .run(args);

api/src/main/java/io/kafbat/ui/config/ClustersProperties.java

@@ -159,7 +159,7 @@ public static class SchemaRegistryAuth {
   public static class TruststoreConfig {
     String truststoreLocation;
     String truststorePassword;
-    boolean verifySsl = true;
+    boolean verify = true;
   }
 
   @Data
@@ -254,8 +254,8 @@ public boolean use(Boolean request) {
       if (enabled) {
         if (Boolean.TRUE.equals(request)) {
           return true;
-        } else if (request == null && defaultEnabled) {
-          return true;
+        } else {
+          return request == null && defaultEnabled;
         }
       }
       return false;
@@ -287,7 +287,6 @@ private void flattenClusterProperties() {
     }
   }
 
-  @SuppressWarnings("unchecked")
   private Map<String, Object> flattenClusterProperties(@Nullable String prefix,
                                                        @Nullable Map<String, Object> propertiesMap) {
     Map<String, Object> flattened = new HashMap<>();
@@ -306,8 +305,8 @@ private Map<String, Object> flattenClusterProperties(@Nullable String prefix,
 
   private void validateClusterNames() {
     // if only one cluster provided it is ok not to set name
-    if (clusters.size() == 1 && !StringUtils.hasText(clusters.get(0).getName())) {
-      clusters.get(0).setName("Default");
+    if (clusters.size() == 1 && !StringUtils.hasText(clusters.getFirst().getName())) {
+      clusters.getFirst().setName("Default");
       return;
     }
 

api/src/main/java/io/kafbat/ui/serdes/builtin/sr/SchemaRegistrySerde.java

@@ -24,6 +24,7 @@
 import io.kafbat.ui.serde.api.PropertyResolver;
 import io.kafbat.ui.serde.api.SchemaDescription;
 import io.kafbat.ui.serdes.BuiltInSerde;
+import io.kafbat.ui.service.ssl.SkipSecurityProvider;
 import io.kafbat.ui.util.jsonschema.AvroJsonSchemaConverter;
 import io.kafbat.ui.util.jsonschema.ProtobufSchemaConverter;
 import java.net.URI;
@@ -76,7 +77,8 @@
             kafkaClusterProperties.getProperty("schemaRegistrySsl.keystoreLocation", String.class).orElse(null),
             kafkaClusterProperties.getProperty("schemaRegistrySsl.keystorePassword", String.class).orElse(null),
             kafkaClusterProperties.getProperty("ssl.truststoreLocation", String.class).orElse(null),
-            kafkaClusterProperties.getProperty("ssl.truststorePassword", String.class).orElse(null)
+            kafkaClusterProperties.getProperty("ssl.truststorePassword", String.class).orElse(null),
+            kafkaClusterProperties.getProperty("ssl.verify", Boolean.class).orElse(true)
         ),
         kafkaClusterProperties.getProperty("schemaRegistryKeySchemaNameTemplate", String.class).orElse("%s-key"),
         kafkaClusterProperties.getProperty("schemaRegistrySchemaNameTemplate", String.class).orElse("%s-value"),
@@ -102,7 +104,8 @@
             serdeProperties.getProperty("keystoreLocation", String.class).orElse(null),
             serdeProperties.getProperty("keystorePassword", String.class).orElse(null),
             kafkaClusterProperties.getProperty("ssl.truststoreLocation", String.class).orElse(null),
-            kafkaClusterProperties.getProperty("ssl.truststorePassword", String.class).orElse(null)
+            kafkaClusterProperties.getProperty("ssl.truststorePassword", String.class).orElse(null),
+            kafkaClusterProperties.getProperty("ssl.verify", Boolean.class).orElse(true)
         ),
         serdeProperties.getProperty("keySchemaNameTemplate", String.class).orElse("%s-key"),
         serdeProperties.getProperty("schemaNameTemplate", String.class).orElse("%s-value"),
@@ -126,13 +129,14 @@
     this.checkSchemaExistenceForDeserialize = checkTopicSchemaExistenceForDeserialize;
   }
 
   private static SchemaRegistryClient createSchemaRegistryClient(List<String> urls,
                                                                  @Nullable String username,
                                                                  @Nullable String password,
                                                                  @Nullable String keyStoreLocation,
                                                                  @Nullable String keyStorePassword,
                                                                  @Nullable String trustStoreLocation,
-                                                                 @Nullable String trustStorePassword) {
+                                                                 @Nullable String trustStorePassword,
+                                                                 boolean verifySsl) {
     Map<String, String> configs = new HashMap<>();
     if (username != null && password != null) {
       configs.put(BASIC_AUTH_CREDENTIALS_SOURCE, "USER_INFO");
@@ -145,6 +149,13 @@
           "You specified password but do not specified username");
     }
 
+    if (!verifySsl) {
+      configs.put(
+          SchemaRegistryClientConfig.CLIENT_NAMESPACE + SslConfigs.SSL_TRUSTMANAGER_ALGORITHM_CONFIG,
+          SkipSecurityProvider.NAME
+      );
+    }
+
     // We require at least a truststore. The logic is done similar to SchemaRegistryService.securedWebClientOnTLS
     if (trustStoreLocation != null && trustStorePassword != null) {
       configs.put(SchemaRegistryClientConfig.CLIENT_NAMESPACE + SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,

api/src/main/java/io/kafbat/ui/service/ssl/SkipSecurityProvider.java

@@ -0,0 +1,12 @@
+package io.kafbat.ui.service.ssl;
+
+import java.security.Provider;
+
+public class SkipSecurityProvider extends Provider {
+  public static final String NAME = "Skip";
+
+  public SkipSecurityProvider() {
+    super(NAME, "1.0", "Skip TrustManagerFactory Provider");
+    put("TrustManagerFactory." + NAME, "io.kafbat.ui.service.ssl.SkipTrustManagerFactorySpi");
+  }
+}

api/src/main/java/io/kafbat/ui/service/ssl/SkipTrustManagerFactorySpi.java

@@ -0,0 +1,26 @@
+package io.kafbat.ui.service.ssl;
+
+import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
+import java.security.KeyStore;
+import javax.net.ssl.ManagerFactoryParameters;
+import javax.net.ssl.TrustManager;
+
+@SuppressWarnings("unused")
+public class SkipTrustManagerFactorySpi extends javax.net.ssl.TrustManagerFactorySpi {
+
+  public SkipTrustManagerFactorySpi() {
+  }
+
+  @Override
+  protected void engineInit(KeyStore ks) {
+  }
+
+  @Override
+  protected void engineInit(ManagerFactoryParameters spec) {
+  }
+
+  @Override
+  protected TrustManager[] engineGetTrustManagers() {
+    return InsecureTrustManagerFactory.INSTANCE.getTrustManagers();
+  }
+}

api/src/main/java/io/kafbat/ui/util/KafkaClientSslPropertiesUtil.java

@@ -16,7 +16,7 @@ public static void addKafkaSslProperties(@Nullable ClustersProperties.Truststore
       return;
     }
 
-    if (!truststoreConfig.isVerifySsl()) {
+    if (!truststoreConfig.isVerify()) {
       sink.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
     }
 

api/src/main/java/io/kafbat/ui/util/WebClientConfigurator.java

@@ -47,7 +47,7 @@ private static ObjectMapper defaultOM() {
 
   public WebClientConfigurator configureSsl(@Nullable ClustersProperties.TruststoreConfig truststoreConfig,
                                             @Nullable ClustersProperties.KeystoreConfig keystoreConfig) {
-    if (truststoreConfig != null && !truststoreConfig.isVerifySsl()) {
+    if (truststoreConfig != null && !truststoreConfig.isVerify()) {
       return configureNoSsl();
     }
 
@@ -130,14 +130,13 @@ public WebClientConfigurator configureBufferSize(DataSize maxBuffSize) {
     return this;
   }
 
-  public WebClientConfigurator configureObjectMapper(ObjectMapper mapper) {
+  public void configureObjectMapper(ObjectMapper mapper) {
     builder.codecs(codecs -> {
       codecs.defaultCodecs()
           .jackson2JsonEncoder(new Jackson2JsonEncoder(mapper, MediaType.APPLICATION_JSON));
       codecs.defaultCodecs()
           .jackson2JsonDecoder(new Jackson2JsonDecoder(mapper, MediaType.APPLICATION_JSON));
     });
-    return this;
   }
 
   public WebClientConfigurator configureCodecs(Consumer<ClientCodecConfigurer> configurer) {

contract-typespec/api/config.tsp

@@ -146,7 +146,7 @@ model ApplicationConfig {
         ssl?: {
           truststoreLocation?: string;
           truststorePassword?: string;
-          verifySsl?: boolean = true;
+          verify?: boolean = true;
         };
         schemaRegistry?: string;
         schemaRegistryAuth?: {