Ability to configure SSL truststore and keystore for SR Clients

client/src/main/java/io/confluent/kafka/schemaregistry/client/security/SslFactory.java

@@ -65,7 +65,7 @@ public SslFactory(Map<String, ?> configs) {
 
       this.sslContext = createSslContext();
     } catch (Exception e) {
-      throw new RuntimeException("Error initializing the ssl context for RestService" + e);
+      throw new RuntimeException("Error initializing the ssl context for RestService" , e);
     }
   }
 

core/src/test/java/io/confluent/kafka/schemaregistry/rest/RestApiSslTest.java

@@ -0,0 +1,144 @@
+/*
+ * Copyright 2019 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ * http://www.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.kafka.schemaregistry.rest;
+
+import io.confluent.kafka.schemaregistry.ClusterTestHarness;
+import io.confluent.kafka.schemaregistry.avro.AvroCompatibilityLevel;
+import io.confluent.kafka.schemaregistry.avro.AvroUtils;
+import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient;
+import io.confluent.kafka.schemaregistry.client.SchemaRegistryClientConfig;
+import org.apache.avro.Schema;
+import org.apache.kafka.common.config.types.Password;
+import org.junit.Test;
+
+import javax.net.ssl.HostnameVerifier;
+import javax.net.ssl.HttpsURLConnection;
+import javax.net.ssl.SSLSession;
+import javax.security.auth.login.Configuration;
+import java.io.File;
+import java.security.KeyPair;
+import java.security.cert.X509Certificate;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+
+public class RestApiSslTest extends ClusterTestHarness {
+
+  Properties props = new Properties();
+
+  public RestApiSslTest() {
+    super(1, true, AvroCompatibilityLevel.BACKWARD.name);
+  }
+
+
+  @Test
+  public void testRegisterWithClientSecurity() throws Exception {
+
+    setupHostNameVerifier();
+
+    String subject = "testSubject";
+    Schema schema = AvroUtils.parseSchema(
+        "{\"type\":\"record\","
+        + "\"name\":\"myrecord\","
+        + "\"fields\":"
+        + "[{\"type\":\"string\",\"name\":\"f1\"}]}").schemaObj;
+
+    int expectedIdSchema1 = 1;
+
+    Map clientsslConfigs = new HashMap();
+    clientsslConfigs.put(
+        SchemaRegistryClientConfig.CLIENT_NAMESPACE + SchemaRegistryConfig.SSL_PROTOCOL_CONFIG,
+        "TLS");
+    clientsslConfigs.put(
+        SchemaRegistryClientConfig.CLIENT_NAMESPACE + SchemaRegistryConfig.SSL_KEYSTORE_LOCATION_CONFIG,
+        props.get(SchemaRegistryConfig.SSL_KEYSTORE_LOCATION_CONFIG));
+    clientsslConfigs.put(
+        SchemaRegistryClientConfig.CLIENT_NAMESPACE + SchemaRegistryConfig.SSL_KEYSTORE_PASSWORD_CONFIG,
+        props.get(SchemaRegistryConfig.SSL_KEYSTORE_PASSWORD_CONFIG));
+    clientsslConfigs.put(
+        SchemaRegistryClientConfig.CLIENT_NAMESPACE + SchemaRegistryConfig.SSL_KEY_PASSWORD_CONFIG,
+        props.get(SchemaRegistryConfig.SSL_KEYSTORE_PASSWORD_CONFIG));
+    clientsslConfigs.put(
+        SchemaRegistryClientConfig.CLIENT_NAMESPACE + SchemaRegistryConfig.SSL_KEYSTORE_TYPE_CONFIG,
+        props.get(SchemaRegistryConfig.SSL_KEYSTORE_TYPE_CONFIG));
+    clientsslConfigs.put(
+        SchemaRegistryClientConfig.CLIENT_NAMESPACE + SchemaRegistryConfig.SSL_TRUSTSTORE_LOCATION_CONFIG,
+        props.get(SchemaRegistryConfig.SSL_TRUSTSTORE_LOCATION_CONFIG));
+    clientsslConfigs.put(
+        SchemaRegistryClientConfig.CLIENT_NAMESPACE + SchemaRegistryConfig.SSL_TRUSTSTORE_PASSWORD_CONFIG,
+        props.get(SchemaRegistryConfig.SSL_TRUSTSTORE_PASSWORD_CONFIG));
+    clientsslConfigs.put(
+        SchemaRegistryClientConfig.CLIENT_NAMESPACE + SchemaRegistryConfig.SSL_TRUSTSTORE_TYPE_CONFIG,
+        props.get(SchemaRegistryConfig.SSL_TRUSTSTORE_TYPE_CONFIG));
+    CachedSchemaRegistryClient schemaRegistryClient = new CachedSchemaRegistryClient(restApp.restClient, 10, clientsslConfigs);
+
+    assertEquals(
+        "Registering should succeed",
+        expectedIdSchema1,
+        schemaRegistryClient.register(subject, schema)
+    );
+
+  }
+
+
+  @Override
+  protected Properties getSchemaRegistryProperties() {
+    Configuration.setConfiguration(null);
+    props.put(
+        SchemaRegistryConfig.SCHEMAREGISTRY_INTER_INSTANCE_PROTOCOL_CONFIG,
+        "https"
+    );
+    props.put(SchemaRegistryConfig.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
+    try {
+      File trustStoreFile = File.createTempFile("truststore", ".jks");
+      trustStoreFile.deleteOnExit();
+      List<X509Certificate> clientCerts = new ArrayList<>();
+
+      List<KeyPair> keyPairs = new ArrayList<>();
+      props.putAll(
+          SecureTestUtils.clientSslConfigsWithKeyStore(1, trustStoreFile, new Password
+                  ("TrustPassword"), clientCerts,
+              keyPairs
+          ));
+      props.put(SchemaRegistryConfig.SSL_CLIENT_AUTHENTICATION_CONFIG, SchemaRegistryConfig.SSL_CLIENT_AUTHENTICATION_REQUIRED);
+
+    } catch (Exception e) {
+      e.printStackTrace();
+    }
+    return props;
+  }
+
+  @Override
+  protected String getSchemaRegistryProtocol() {
+    return "https";
+  }
+
+  private void setupHostNameVerifier() {
+      // Create all-trusting host name verifier
+      HostnameVerifier allHostsValid = new HostnameVerifier() {
+        public boolean verify(String hostname, SSLSession session) {
+          return true;
+        }
+      };
+      // Install the all-trusting host verifier
+      HttpsURLConnection.setDefaultHostnameVerifier(allHostsValid);
+  }
+
+}

core/src/test/java/io/confluent/kafka/schemaregistry/rest/SecureTestUtils.java

@@ -0,0 +1,118 @@
+/*
+ * Copyright 2019 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ * http://www.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.kafka.schemaregistry.rest;
+
+import org.apache.kafka.common.config.types.Password;
+import org.apache.kafka.test.TestSslUtils;
+
+import javax.net.ssl.TrustManagerFactory;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.security.KeyPair;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.cert.Certificate;
+import java.security.cert.CertificateException;
+import java.security.cert.X509Certificate;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+public class SecureTestUtils {
+
+  public static Properties clientSslConfigsWithKeyStore(
+      int numberOfCerts,
+      File trustStoreFile,
+      Password trustPassword,
+      List<X509Certificate> clientCerts,
+      List<KeyPair> keyPairs
+  ) throws GeneralSecurityException, IOException {
+
+    Map<String, X509Certificate> certificateMap = new HashMap<>();
+
+    File clientKSFile = File.createTempFile("CKeystore", ".jks");
+    String keyStorePassword = new Password("Client-KS-Password").value();
+
+    for (int i = 0; i < numberOfCerts; i++) {
+      KeyPair kp = TestSslUtils.generateKeyPair("RSA");
+      X509Certificate cert = TestSslUtils.generateCertificate(
+          "CN=localhost, O=Client" + i, kp, 30, "SHA1withRSA");
+
+      clientCerts.add(cert);
+      keyPairs.add(kp);
+      certificateMap.put("client-" + i, cert);
+    }
+
+    createKeyStore(clientKSFile, keyStorePassword, clientCerts, keyPairs);
+
+    TestSslUtils.createTrustStore(trustStoreFile.toString(), trustPassword, certificateMap);
+
+    Properties sslConfigs =
+        getClientSslConfigs(trustStoreFile, trustPassword.value(), clientKSFile, keyStorePassword);
+
+    return sslConfigs;
+  }
+
+  public static void createKeyStore(
+      File keyStoreFile,
+      String keyStorePassword,
+      List<X509Certificate> clientCerts,
+      List<KeyPair> keyPairs
+  ) throws CertificateException, NoSuchAlgorithmException, IOException, KeyStoreException {
+    KeyStore keyStore = KeyStore.getInstance("JKS");
+    keyStore.load(null, null);
+
+    for (int i = 0; i < clientCerts.size(); i++) {
+      keyStore.setKeyEntry(
+          "client-" + i,
+          keyPairs.get(i).getPrivate(),
+          keyStorePassword.toCharArray(),
+          new Certificate[]{clientCerts.get(i)}
+      );
+    }
+
+    FileOutputStream out = new FileOutputStream(keyStoreFile);
+    keyStore.store(out, keyStorePassword.toCharArray());
+    out.close();
+
+  }
+
+  private static Properties getClientSslConfigs(
+      File trustStoreFile, String trustPassword,
+      File clientKSFile, String keyStorePassword
+  ) {
+    //makes sure client also runs TLSv1.2
+    System.setProperty("https.protocols", "TLSv1.2");
+    Properties sslConfigs = new Properties();
+    sslConfigs.put("ssl.protocol", "TLSv1.2");
+    sslConfigs.put("ssl.keystore.location", clientKSFile.getPath());
+    sslConfigs.put("ssl.keystore.type", "JKS");
+    sslConfigs.put("ssl.keymanager.algorithm", TrustManagerFactory.getDefaultAlgorithm());
+    sslConfigs.put("ssl.keystore.password", keyStorePassword);
+    sslConfigs.put("ssl.key.password", keyStorePassword);
+    sslConfigs.put("ssl.truststore.location", trustStoreFile.getPath());
+    sslConfigs.put("ssl.truststore.password", trustPassword);
+    sslConfigs.put("ssl.truststore.type", "JKS");
+    sslConfigs.put("ssl.trustmanager.algorithm", TrustManagerFactory.getDefaultAlgorithm());
+    sslConfigs.put("ssl.enabled.protocols", "TLSv1.2");
+    sslConfigs.put("security.protocol", "SSL");
+    return sslConfigs;
+  }
+}