Merge pull request #31 from b3rnh8rd/develop

schema-registry supports connection via https without cert path validation and without endpoint identification
patschuh · Nov 14, 2022 · 4a18e62 · 4a18e62
2 parents b7c54e1 + 8a82985
commit 4a18e62
Show file tree

Hide file tree

Showing 6 changed files with 71 additions and 23 deletions.
diff --git a/README.md b/README.md
@@ -134,8 +134,16 @@ With default config the API does a host name identification which fails in those
 ```
 java.security.cert.CertificateException: No subject alternative names matching IP address .... found
 ```
-If you select the toggle "No SSL Endpoint Identification" the kafka property "ssl.endpoint.identification.algorithm" 
-is set to an empty string so that this identification is suppressed 
+If you select the toggle "No SSL Endpoint Identification" the kafka property "ssl.endpoint.identification.algorithm" and schema-registry property "schema-registry.ssl.endpoint.identification.algorithm"
+are set to an empty string so that this identification is suppressed 
+
+###### suppress cert path validation
+In some situation you might need to suppress domain name validation for schema-registry.
+With default config the API does a cert path validation which fails when using an ssh tunnel.
+```
+PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
+```
+If you select the toggle "suppress Cert Path Validation" an empty trustmanager is set.
 
 ###### Settings
 

diff --git a/src/main/java/at/esque/kafka/SchemaRegistryBrowserController.java b/src/main/java/at/esque/kafka/SchemaRegistryBrowserController.java
@@ -47,7 +47,7 @@ public class SchemaRegistryBrowserController {
     public void setup(ClusterConfig selectedConfig, ConfigHandler configHandler) {
         schemaRegistryRestService = new RestService(selectedConfig.getSchemaRegistry());
 
-        if (selectedConfig.isSchemaRegistryUseSsl()) {
+        if (selectedConfig.isSchemaRegistrySuppressCertPathValidation() || selectedConfig.isSchemaRegistryUseSsl()) {
             SSLSocketFactory sslSocketFactory = SslSocketFactoryCreator.buildSSlSocketFactory(selectedConfig, configHandler);
             schemaRegistryRestService.setSslSocketFactory(sslSocketFactory);
         }
@@ -84,14 +84,14 @@ public void setup(ClusterConfig selectedConfig, ConfigHandler configHandler) {
                         versionComboBox.getSelectionModel().select(versionComboBox.getItems().size() - 1);
                     }
                 } catch (Exception e) {
-                    ErrorAlert.show(e, getWindow());
+                    ErrorAlert.show(e);
                 }
             });
 
             subjectListView.setItems(FXCollections.observableArrayList(schemaRegistryRestService.getAllSubjects()));
 
         } catch (Exception e) {
-            ErrorAlert.show(e, getWindow());
+            ErrorAlert.show(e);
         }
     }
 

diff --git a/src/main/java/at/esque/kafka/cluster/ClusterConfig.java b/src/main/java/at/esque/kafka/cluster/ClusterConfig.java
@@ -16,6 +16,7 @@ public class ClusterConfig {
     private StringProperty schemaRegistryBasicAuthUserInfo = new SimpleStringProperty();
     private BooleanProperty schemaRegistryUseSsl = new SimpleBooleanProperty();
     private BooleanProperty sslEnabled = new SimpleBooleanProperty();
+    private BooleanProperty certPathValidationSuppressed = new SimpleBooleanProperty();
     private StringProperty keyStoreLocation = new SimpleStringProperty();
     private StringProperty keyStorePassword = new SimpleStringProperty();
     private StringProperty trustStoreLocation = new SimpleStringProperty();
@@ -43,6 +44,7 @@ public void update(ClusterConfig existingConfig) {
             this.setSchemaRegistry(existingConfig.getSchemaRegistry());
             this.setSchemaRegistryBasicAuthUserInfo(existingConfig.getSchemaRegistryBasicAuthUserInfo());
             this.setSchemaRegistryUseSsl(existingConfig.isSchemaRegistryUseSsl());
+            this.setSchemaRegistrySuppressCertPathValidation(existingConfig.isSchemaRegistrySuppressCertPathValidation());
             this.setSslEnabled(existingConfig.isSslEnabled());
             this.setKeyStoreLocation(existingConfig.getKeyStoreLocation());
             this.setKeyStorePassword(existingConfig.getKeyStorePassword());
@@ -111,6 +113,18 @@ public void setSslEnabled(boolean sslEnabled) {
         this.sslEnabled.set(sslEnabled);
     }
 
+    @JsonProperty("certPathValidationSuppressed")
+    public boolean isSchemaRegistrySuppressCertPathValidation() {
+        return certPathValidationSuppressed.get();
+    }
+    public BooleanProperty suppressCertPathValidation() {
+        return certPathValidationSuppressed;
+    }
+
+    public void setSchemaRegistrySuppressCertPathValidation(boolean suppressCertPathValidation) {
+        this.certPathValidationSuppressed.set(suppressCertPathValidation);
+    }
+
     @JsonProperty("keyStoreLocation")
     public String getKeyStoreLocation() {
         return keyStoreLocation.get();

diff --git a/src/main/java/at/esque/kafka/cluster/SslSocketFactoryCreator.java b/src/main/java/at/esque/kafka/cluster/SslSocketFactoryCreator.java
@@ -3,10 +3,7 @@
 import at.esque.kafka.handlers.ConfigHandler;
 import org.apache.kafka.common.config.SslConfigs;
 
-import javax.net.ssl.KeyManagerFactory;
-import javax.net.ssl.SSLContext;
-import javax.net.ssl.SSLSocketFactory;
-import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.*;
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.security.KeyManagementException;
@@ -15,36 +12,59 @@
 import java.security.NoSuchAlgorithmException;
 import java.security.UnrecoverableKeyException;
 import java.security.cert.CertificateException;
+import java.security.cert.X509Certificate;
 import java.util.Map;
 
 public class SslSocketFactoryCreator {
 
     private SslSocketFactoryCreator() {
     }
 
+    private static final TrustManager[] UNQUESTIONING_TRUST_MANAGER = new TrustManager[]{
+            new X509TrustManager() {
+                public X509Certificate[] getAcceptedIssuers() {
+                    return null;
+                }
+
+                public void checkClientTrusted(X509Certificate[] certs, String authType) {
+                }
+
+                public void checkServerTrusted(X509Certificate[] certs, String authType) {
+                }
+            }
+    };
+
     public static SSLSocketFactory buildSSlSocketFactory(ClusterConfig clusterConfig, ConfigHandler configHandler) {
         try {
-            KeyStore ks = KeyStore.getInstance(KeyStore.getDefaultType());
+            SSLContext sc = SSLContext.getInstance("TLSv1.2");
             Map<String, String> sslProperties = configHandler.getSslProperties(clusterConfig);
-            ks.load(new FileInputStream(sslProperties.get(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG)), sslProperties.get(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG).toCharArray());
-
-            KeyStore ts = KeyStore.getInstance(KeyStore.getDefaultType());
-            ts.load(new FileInputStream(sslProperties.get(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG)), sslProperties.get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG).toCharArray());
-
             KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
-            kmf.init(ks, sslProperties.get(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG).toCharArray());
 
-            TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
-            tmf.init(ts);
+            if (useKeyStore(configHandler.getSslProperties(clusterConfig))) {
+                KeyStore ks = KeyStore.getInstance(KeyStore.getDefaultType());
+                ks.load(new FileInputStream(sslProperties.get(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG)), sslProperties.get(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG).toCharArray());
+                kmf.init(ks, sslProperties.get(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG).toCharArray());
+            }
+            if (clusterConfig.isSchemaRegistrySuppressCertPathValidation()) {
+                sc.init(kmf.getKeyManagers(), UNQUESTIONING_TRUST_MANAGER, null);
+            } else {
+                KeyStore ts = KeyStore.getInstance(KeyStore.getDefaultType());
+                ts.load(new FileInputStream(sslProperties.get(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG)), sslProperties.get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG).toCharArray());
+                TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+                tmf.init(ts);
 
-            SSLContext sc = SSLContext.getInstance("TLSv1.2");
-            sc.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);
+                sc.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);
+            }
 
             return sc.getSocketFactory();
-
-        } catch (KeyStoreException | IOException | NoSuchAlgorithmException | CertificateException | UnrecoverableKeyException | KeyManagementException e) {
+        } catch (KeyStoreException | IOException | NoSuchAlgorithmException | CertificateException |
+                 UnrecoverableKeyException | KeyManagementException e) {
             at.esque.kafka.alerts.ErrorAlert.show(e);
             return null;
         }
     }
+
+    private static boolean useKeyStore(Map<String, String> sslProperties) {
+        return sslProperties.get(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG) != null && sslProperties.get(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG) != null;
+    }
 }
diff --git a/src/main/java/at/esque/kafka/dialogs/ClusterConfigDialog.java b/src/main/java/at/esque/kafka/dialogs/ClusterConfigDialog.java
@@ -33,6 +33,7 @@ public class ClusterConfigDialog {
     public static final String LABEL_KAFKA_CONNECT_BASIC_AUTH_USER = "Kafka Connect Basic Auth User";
     public static final String LABEL_KAFKA_CONNECT_BASIC_AUTH_PASSWORD = "Kafka Connect Basic Auth Password";
     public static final String LABEL_USE_SSL_CONFIGURATION = "use SSL Configuration";
+    public static final String LABEL_SUPPRESS_CERT_PATH_VALIDATION = "suppress Cert Path Validation";
     public static final String LABEL_SUPPRESS_SSL_ENDPOINT_IDENTIFICATION = "no SSL Endpoint Identification";
 
     private ClusterConfigDialog(){}
@@ -80,7 +81,11 @@ public static Optional<ClusterConfig> show(ClusterConfig existingConfig) {
                         Field.ofBooleanType(copy.isSchemaRegistryUseSsl())
                                 .label(LABEL_USE_SSL_CONFIGURATION)
                                 .tooltip(LABEL_USE_SSL_CONFIGURATION)
-                                .bind(copy.schemaRegistryUseSslProperty())
+                                .bind(copy.schemaRegistryUseSslProperty()),
+                        Field.ofBooleanType(copy.isSchemaRegistrySuppressCertPathValidation())
+                                .label(LABEL_SUPPRESS_CERT_PATH_VALIDATION)
+                                .tooltip(LABEL_SUPPRESS_CERT_PATH_VALIDATION)
+                                .bind(copy.suppressCertPathValidation())
                 ),
                 Group.of(
                         Field.ofStringType(copy.getkafkaConnectUrl()==null?"":copy.getkafkaConnectUrl())

diff --git a/src/main/java/at/esque/kafka/handlers/ConfigHandler.java b/src/main/java/at/esque/kafka/handlers/ConfigHandler.java
@@ -257,6 +257,7 @@ public Map<String, String> getSslProperties(ClusterConfig config) {
 
         if (config.issuppressSslEndPointIdentification()) {
             props.put("ssl.endpoint.identification.algorithm", "");
+            props.put("schema-registry.ssl.endpoint.identification.algorithm", "");
         }
 
         if (config.isSchemaRegistryHttps()) {