Mutual TLS support (#6076)

* Mutual TLS support * Kafka test fixes * TeamCity fix * Split integration tests * Use localhost DOCKER_IP * Increase server thread count * Increase SSL handshake timeouts * Add broken pipe retries, use injected client config params * PR comments, Rat license check exclusion
2018-09-19 09:56:15 -07:00 · 2018-09-19 09:56:15 -07:00 · 8972244c68
parent 4fafc2ccc9
commit 8972244c68
59 changed files with 2081 additions and 99 deletions
--- a/.travis.yml
+++ b/.travis.yml
@ -93,8 +93,8 @@ matrix:
      services:
        - docker
      env:
-        - NAME="integration test"
-        - DOCKER_IP=172.17.0.1
+        - NAME="integration test part 1"
+        - DOCKER_IP=127.0.0.1
      install:
        # Only errors will be shown with the -q option. This is to avoid generating too many logs which make travis build failed.
        - mvn install -q -ff -DskipTests -B
@ -108,3 +108,24 @@ matrix:
          echo $v dmesg ======================== ;
          docker exec -it druid-$v sh -c 'dmesg | tail -3' ;
          done
+
+      # run integration tests
+    - sudo: required
+      services:
+        - docker
+      env:
+        - NAME="integration test part 2"
+        - DOCKER_IP=127.0.0.1
+      install:
+        # Only errors will be shown with the -q option. This is to avoid generating too many logs which make travis build failed.
+        - mvn install -q -ff -DskipTests -B
+      script:
+        - $TRAVIS_BUILD_DIR/ci/travis_script_integration_part2.sh
+      after_failure:
+        - for v in ~/shared/logs/*.log ; do
+          echo $v logtail ======================== ; tail -100 $v ;
+          done
+        - for v in broker middlemanager overlord router coordinator historical ; do
+          echo $v dmesg ======================== ;
+          docker exec -it druid-$v sh -c 'dmesg | tail -3' ;
+          done
--- a/ci/travis_script_integration.sh
+++ b/ci/travis_script_integration.sh
@ -21,6 +21,6 @@ set -e

 pushd $TRAVIS_BUILD_DIR/integration-tests

-mvn verify -P integration-tests
+mvn verify -P integration-tests -Dit.test=ITAppenderatorDriverRealtimeIndexTaskTest,ITCompactionTaskTest,ITIndexerTest,ITKafkaIndexingServiceTest,ITKafkaTest,ITParallelIndexTest,ITRealtimeIndexTaskTest

 popd
--- a/ci/travis_script_integration_part2.sh
+++ b/ci/travis_script_integration_part2.sh
@ -0,0 +1,26 @@
+#!/usr/bin/env bash
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+set -e
+
+pushd $TRAVIS_BUILD_DIR/integration-tests
+
+mvn verify -P integration-tests -Dit.test=ITUnionQueryTest,ITTwitterQueryTest,ITWikipediaQueryTest,ITBasicAuthConfigurationTest,ITTLSTest
+
+popd
--- a/docs/content/development/extensions-core/simple-client-sslcontext.md
+++ b/docs/content/development/extensions-core/simple-client-sslcontext.md
@ -18,5 +18,16 @@ Java's SSL support, please refer to [this](http://docs.oracle.com/javase/8/docs/
 |`druid.client.https.trustStoreAlgorithm`|Algorithm to be used by TrustManager to validate certificate chains|`javax.net.ssl.TrustManagerFactory.getDefaultAlgorithm()`|no|
 |`druid.client.https.trustStorePassword`|The [Password Provider](../../operations/password-provider.html) or String password for the Trust Store.|none|yes|

+The following table contains optional parameters for supporting client certificate authentication:
+
+|Property|Description|Default|Required|
+|--------|-----------|-------|--------|
+|`druid.client.https.keyStorePath`|The file path or URL of the TLS/SSL Key store containing the client certificate that Druid will use when communicating with other Druid services. If this is null, the other properties in this table are ignored.|none|yes|
+|`druid.client.https.keyStoreType`|The type of the key store.|none|yes|
+|`druid.client.https.certAlias`|Alias of TLS client certificate in the keystore.|none|yes|
+|`druid.client.https.keyStorePassword`|The [Password Provider](../operations/password-provider.html) or String password for the Key Store.|none|no|
+|`druid.client.https.keyManagerFactoryAlgorithm`|Algorithm to use for creating KeyManager, more details [here](https://docs.oracle.com/javase/7/docs/technotes/guides/security/jsse/JSSERefGuide.html#KeyManager).|`javax.net.ssl.KeyManagerFactory.getDefaultAlgorithm()`|no|
+|`druid.client.https.keyManagerPassword`|The [Password Provider](../operations/password-provider.html) or String password for the Key Manager.|none|no|
+
 This [document](http://docs.oracle.com/javase/8/docs/technotes/guides/security/StandardNames.html) lists all the possible
 values for the above mentioned configs among others provided by Java implementation.
--- a/docs/content/operations/tls-support.md
+++ b/docs/content/operations/tls-support.md
@ -31,7 +31,19 @@ values for the below mentioned configs among others provided by Java implementat
 |`druid.server.https.certAlias`|Alias of TLS/SSL certificate for the connector.|none|yes|
 |`druid.server.https.keyStorePassword`|The [Password Provider](../operations/password-provider.html) or String password for the Key Store.|none|yes|

-Following table contains non-mandatory advanced configuration options, use caution.
+The following table contains configuration options related to client certificate authentication.
+
+|Property|Description|Default|Required|
+|--------|-----------|-------|--------|
+|`druid.server.https.requireClientCertificate`|If set to true, clients must identify themselves by providing a TLS certificate.  If `requireClientCertificate` is false, the rest of the options in this table are ignored.|false|no|
+|`druid.server.https.trustStoreType`|The type of the trust store containing certificates used to validate client certificates. Not needed if `requireClientCertificate` is false.|`java.security.KeyStore.getDefaultType()`|no|
+|`druid.server.https.trustStorePath`|The file path or URL of the trust store containing certificates used to validate client certificates. Not needed if `requireClientCertificate` is false.|none|yes, only if `requireClientCertificate` is true|
+|`druid.server.https.trustStoreAlgorithm`|Algorithm to be used by TrustManager to validate client certificate chains. Not needed if `requireClientCertificate` is false.|`javax.net.ssl.TrustManagerFactory.getDefaultAlgorithm()`|no|
+|`druid.server.https.trustStorePassword`|The [Password Provider](../../operations/password-provider.html) or String password for the Trust Store.  Not needed if `requireClientCertificate` is false.|none|no|
+|`druid.server.https.validateHostnames`|If set to true, check that the client's hostname matches the CN/subjectAltNames in the client certificate.  Not used if `requireClientCertificate` is false.|true|no|
+|`druid.server.https.crlPath`|Specifies a path to a file containing static [Certificate Revocation Lists](https://en.wikipedia.org/wiki/Certificate_revocation_list), used to check if a client certificate has been revoked. Not used if `requireClientCertificate` is false.|null|no|
+
+The following table contains non-mandatory advanced configuration options, use caution.

 |Property|Description|Default|Required|
 |--------|-----------|-------|--------|
--- a/extensions-core/simple-client-sslcontext/src/main/java/org/apache/druid/https/SSLClientConfig.java
+++ b/extensions-core/simple-client-sslcontext/src/main/java/org/apache/druid/https/SSLClientConfig.java
@ -39,6 +39,24 @@ public class SSLClientConfig
  @JsonProperty("trustStorePassword")
  private PasswordProvider trustStorePasswordProvider;

+  @JsonProperty
+  private String keyStorePath;
+
+  @JsonProperty
+  private String keyStoreType;
+
+  @JsonProperty
+  private String certAlias;
+
+  @JsonProperty("keyStorePassword")
+  private PasswordProvider keyStorePasswordProvider;
+
+  @JsonProperty("keyManagerPassword")
+  private PasswordProvider keyManagerPasswordProvider;
+
+  @JsonProperty
+  private String keyManagerFactoryAlgorithm;
+
  public String getProtocol()
  {
    return protocol;
@ -64,6 +82,36 @@ public class SSLClientConfig
    return trustStorePasswordProvider;
  }

+  public String getKeyStorePath()
+  {
+    return keyStorePath;
+  }
+
+  public String getKeyStoreType()
+  {
+    return keyStoreType;
+  }
+
+  public PasswordProvider getKeyStorePasswordProvider()
+  {
+    return keyStorePasswordProvider;
+  }
+
+  public String getCertAlias()
+  {
+    return certAlias;
+  }
+
+  public PasswordProvider getKeyManagerPasswordProvider()
+  {
+    return keyManagerPasswordProvider;
+  }
+
+  public String getKeyManagerFactoryAlgorithm()
+  {
+    return keyManagerFactoryAlgorithm;
+  }
+
  @Override
  public String toString()
  {
@ -72,6 +120,10 @@ public class SSLClientConfig
           ", trustStoreType='" + trustStoreType + '\'' +
           ", trustStorePath='" + trustStorePath + '\'' +
           ", trustStoreAlgorithm='" + trustStoreAlgorithm + '\'' +
+           ", keyStorePath='" + keyStorePath + '\'' +
+           ", keyStoreType='" + keyStoreType + '\'' +
+           ", certAlias='" + certAlias + '\'' +
+           ", keyManagerFactoryAlgorithm='" + keyManagerFactoryAlgorithm + '\'' +
           '}';
  }
 }
--- a/extensions-core/simple-client-sslcontext/src/main/java/org/apache/druid/https/SSLContextProvider.java
+++ b/extensions-core/simple-client-sslcontext/src/main/java/org/apache/druid/https/SSLContextProvider.java
@ -43,12 +43,18 @@ public class SSLContextProvider implements Provider<SSLContext>
  {
    log.info("Creating SslContext for https client using config [%s]", config);

-    return TLSUtils.createSSLContext(
-        config.getProtocol(),
-        config.getTrustStoreType(),
-        config.getTrustStorePath(),
-        config.getTrustStoreAlgorithm(),
-        config.getTrustStorePasswordProvider()
-    );
+    return new TLSUtils.ClientSSLContextBuilder()
+        .setProtocol(config.getProtocol())
+        .setTrustStoreType(config.getTrustStoreType())
+        .setTrustStorePath(config.getTrustStorePath())
+        .setTrustStoreAlgorithm(config.getTrustStoreAlgorithm())
+        .setTrustStorePasswordProvider(config.getTrustStorePasswordProvider())
+        .setKeyStoreType(config.getKeyStoreType())
+        .setKeyStorePath(config.getKeyStorePath())
+        .setKeyStoreAlgorithm(config.getKeyManagerFactoryAlgorithm())
+        .setCertAlias(config.getCertAlias())
+        .setKeyStorePasswordProvider(config.getKeyStorePasswordProvider())
+        .setKeyManagerFactoryPasswordProvider(config.getKeyManagerPasswordProvider())
+        .build();
  }
 }
--- a/integration-tests/.gitignore
+++ b/integration-tests/.gitignore
@ -0,0 +1,6 @@
+client_tls/
+docker/docker_ip
+docker/tls/root.key
+docker/tls/root.pem
+docker/tls/untrusted_root.key
+docker/tls/untrusted_root.pem
--- a/integration-tests/docker/Dockerfile
+++ b/integration-tests/docker/Dockerfile
@ -39,32 +39,43 @@ RUN find /var/lib/mysql -type f -exec touch {} \; && service mysql start \
 # Setup supervisord
 ADD supervisord.conf /etc/supervisor/conf.d/supervisord.conf

+# mysql
+ADD run-mysql.sh /run-mysql.sh
+
 # internal docker_ip:9092 endpoint is used to access Kafka from other Docker containers
 # external docker ip:9093 endpoint is used to access Kafka from test code
 # run this last to avoid rebuilding the image every time the ip changes
 ADD docker_ip docker_ip
-RUN perl -pi -e "s/#listeners=.*/listeners=INTERNAL:\/\/$(resolveip -s $HOSTNAME):9092,EXTERNAL:\/\/$(resolveip -s $HOSTNAME):9093/" /usr/local/kafka/config/server.properties
-RUN perl -pi -e "s/#advertised.listeners=.*/advertised.listeners=INTERNAL:\/\/$(resolveip -s $HOSTNAME):9092,EXTERNAL:\/\/$(cat docker_ip):9093/" /usr/local/kafka/config/server.properties
+RUN perl -pi -e "s/#listeners=.*/listeners=INTERNAL:\/\/172.172.172.2:9092,EXTERNAL:\/\/172.172.172.2:9093/" /usr/local/kafka/config/server.properties
+RUN perl -pi -e "s/#advertised.listeners=.*/advertised.listeners=INTERNAL:\/\/172.172.172.2:9092,EXTERNAL:\/\/$(cat docker_ip):9093/" /usr/local/kafka/config/server.properties
 RUN perl -pi -e "s/#listener.security.protocol.map=.*/listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT\ninter.broker.listener.name=INTERNAL/" /usr/local/kafka/config/server.properties
 RUN perl

+# Add directory with TLS support files
+ADD tls tls
+
+ADD client_tls client_tls
+
 # Expose ports:
-# - 8081: HTTP (coordinator)
-# - 8082: HTTP (broker)
-# - 8083: HTTP (historical)
-# - 8090: HTTP (overlord)
-# - 8091: HTTP (middlemanager)
+# - 8081, 8281: HTTP, HTTPS (coordinator)
+# - 8082, 8282: HTTP, HTTPS (broker)
+# - 8083, 8283: HTTP, HTTPS (historical)
+# - 8090, 8290: HTTP, HTTPS (overlord)
+# - 8091, 8291: HTTP, HTTPS (middlemanager)
 # - 3306: MySQL
 # - 2181 2888 3888: ZooKeeper
-# - 8100 8101 8102 8103 8104 : peon ports
-EXPOSE 8081
-EXPOSE 8082
-EXPOSE 8083
-EXPOSE 8090
-EXPOSE 8091
+# - 8100 8101 8102 8103 8104 8105 : peon ports
+# - 8300 8301 8302 8303 8304 8305 : peon HTTPS ports
+EXPOSE 8081 8281
+EXPOSE 8082 8282
+EXPOSE 8083 8283
+EXPOSE 8090 8290
+EXPOSE 8091 8291
 EXPOSE 3306
 EXPOSE 2181 2888 3888
-EXPOSE 8100 8101 8102 8103 8104
+EXPOSE 8100 8101 8102 8103 8104 8105
+EXPOSE 8300 8301 8302 8303 8304 8305
+EXPOSE 9092 9093

 WORKDIR /var/lib/druid
-ENTRYPOINT export HOST_IP="$(resolveip -s $HOSTNAME)" && exec /usr/bin/supervisord -c /etc/supervisor/conf.d/supervisord.conf
+ENTRYPOINT export HOST_IP="$(resolveip -s $HOSTNAME)" && /tls/generate-server-certs-and-keystores.sh && exec /usr/bin/supervisord -c /etc/supervisor/conf.d/supervisord.conf
--- a/integration-tests/docker/broker.conf
+++ b/integration-tests/docker/broker.conf
@ -36,6 +36,26 @@ command=java
  -Ddruid.auth.authorizer.basic.type=basic
  -Ddruid.sql.enable=true
  -Ddruid.sql.avatica.enable=true
+  -Ddruid.enableTlsPort=true
+  -Ddruid.server.https.keyStorePath=/tls/server.jks
+  -Ddruid.server.https.keyStoreType=jks
+  -Ddruid.server.https.certAlias=druid
+  -Ddruid.server.https.keyManagerPassword=druid123
+  -Ddruid.server.https.keyStorePassword=druid123
+  -Ddruid.server.https.requireClientCertificate=true
+  -Ddruid.server.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.server.https.trustStorePassword=druid123
+  -Ddruid.server.https.trustStoreAlgorithm=PKIX
+  -Ddruid.server.https.validateHostnames=true
+  -Ddruid.server.https.crlPath=/tls/revocations.crl
+  -Ddruid.client.https.trustStoreAlgorithm=PKIX
+  -Ddruid.client.https.protocol=TLSv1.2
+  -Ddruid.client.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.client.https.trustStorePassword=druid123
+  -Ddruid.client.https.keyStorePath=/tls/server.jks
+  -Ddruid.client.https.certAlias=druid
+  -Ddruid.client.https.keyManagerPassword=druid123
+  -Ddruid.client.https.keyStorePassword=druid123
  -cp /shared/docker/lib/*
  org.apache.druid.cli.Main server broker
 redirect_stderr=true
--- a/integration-tests/docker/coordinator.conf
+++ b/integration-tests/docker/coordinator.conf
@ -9,6 +9,7 @@ command=java
  -Duser.timezone=UTC
  -Dfile.encoding=UTF-8
  -Ddruid.host=%(ENV_HOST_IP)s
+  -Ddruid.server.http.numThreads=100
  -Ddruid.metadata.storage.type=mysql
  -Ddruid.metadata.storage.connector.connectURI=jdbc:mysql://druid-metadata-storage/druid
  -Ddruid.metadata.storage.connector.user=druid
@ -29,6 +30,26 @@ command=java
  -Ddruid.auth.authorizers="[\"basic\"]"
  -Ddruid.auth.authorizer.basic.type=basic
  -Ddruid.auth.unsecuredPaths="[\"/druid/coordinator/v1/loadqueue\"]"
+  -Ddruid.enableTlsPort=true
+  -Ddruid.server.https.keyStorePath=/tls/server.jks
+  -Ddruid.server.https.keyStoreType=jks
+  -Ddruid.server.https.certAlias=druid
+  -Ddruid.server.https.keyManagerPassword=druid123
+  -Ddruid.server.https.keyStorePassword=druid123
+  -Ddruid.server.https.requireClientCertificate=true
+  -Ddruid.server.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.server.https.trustStorePassword=druid123
+  -Ddruid.server.https.trustStoreAlgorithm=PKIX
+  -Ddruid.server.https.validateHostnames=true
+  -Ddruid.server.https.crlPath=/tls/revocations.crl
+  -Ddruid.client.https.trustStoreAlgorithm=PKIX
+  -Ddruid.client.https.protocol=TLSv1.2
+  -Ddruid.client.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.client.https.trustStorePassword=druid123
+  -Ddruid.client.https.keyStorePath=/tls/server.jks
+  -Ddruid.client.https.certAlias=druid
+  -Ddruid.client.https.keyManagerPassword=druid123
+  -Ddruid.client.https.keyStorePassword=druid123
  -cp /shared/docker/lib/*
  org.apache.druid.cli.Main server coordinator
 redirect_stderr=true
--- a/integration-tests/docker/historical.conf
+++ b/integration-tests/docker/historical.conf
@ -32,6 +32,26 @@ command=java
  -Ddruid.escalator.authorizerName=basic
  -Ddruid.auth.authorizers="[\"basic\"]"
  -Ddruid.auth.authorizer.basic.type=basic
+  -Ddruid.enableTlsPort=true
+  -Ddruid.server.https.keyStorePath=/tls/server.jks
+  -Ddruid.server.https.keyStoreType=jks
+  -Ddruid.server.https.certAlias=druid
+  -Ddruid.server.https.keyManagerPassword=druid123
+  -Ddruid.server.https.keyStorePassword=druid123
+  -Ddruid.server.https.requireClientCertificate=true
+  -Ddruid.server.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.server.https.trustStorePassword=druid123
+  -Ddruid.server.https.trustStoreAlgorithm=PKIX
+  -Ddruid.server.https.validateHostnames=true
+  -Ddruid.server.https.crlPath=/tls/revocations.crl
+  -Ddruid.client.https.trustStoreAlgorithm=PKIX
+  -Ddruid.client.https.protocol=TLSv1.2
+  -Ddruid.client.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.client.https.trustStorePassword=druid123
+  -Ddruid.client.https.keyStorePath=/tls/server.jks
+  -Ddruid.client.https.certAlias=druid
+  -Ddruid.client.https.keyManagerPassword=druid123
+  -Ddruid.client.https.keyStorePassword=druid123
  -cp /shared/docker/lib/*
  org.apache.druid.cli.Main server historical
 redirect_stderr=true
--- a/integration-tests/docker/metadata-storage.conf
+++ b/integration-tests/docker/metadata-storage.conf
@ -1,6 +1,5 @@
 [program:mysql]
-command=/usr/bin/pidproxy /var/run/mysqld/mysqld.pid /usr/bin/mysqld_safe
-         --bind-address=0.0.0.0
+command=/run-mysql.sh
 user=mysql
 priority=0
 stdout_logfile=/shared/logs/mysql.log
--- a/integration-tests/docker/middlemanager.conf
+++ b/integration-tests/docker/middlemanager.conf
@ -9,11 +9,12 @@ command=java
  -Duser.timezone=UTC
  -Dfile.encoding=UTF-8
  -Ddruid.host=%(ENV_HOST_IP)s
+  -Ddruid.server.http.numThreads=100
  -Ddruid.zk.service.host=druid-zookeeper-kafka
  -Ddruid.worker.capacity=3
  -Ddruid.indexer.logs.directory=/shared/tasklogs
  -Ddruid.storage.storageDirectory=/shared/storage
-  -Ddruid.indexer.runner.javaOpts=-server -Xmx256m -Xms256m -XX:NewSize=128m -XX:MaxNewSize=128m -XX:+UseConcMarkSweepGC -XX:+PrintGCDetails -XX:+PrintGCTimeStamps
+  -Ddruid.indexer.runner.javaOpts="-server -Xmx256m -Xms256m -XX:NewSize=128m -XX:MaxNewSize=128m -XX:+UseConcMarkSweepGC -XX:+PrintGCDetails -XX:+PrintGCTimeStamps"
  -Ddruid.indexer.fork.property.druid.processing.buffer.sizeBytes=25000000
  -Ddruid.indexer.fork.property.druid.processing.numThreads=1
  -Ddruid.indexer.fork.server.http.numThreads=100
@ -35,6 +36,27 @@ command=java
  -Ddruid.escalator.authorizerName=basic
  -Ddruid.auth.authorizers="[\"basic\"]"
  -Ddruid.auth.authorizer.basic.type=basic
+  -Ddruid.enableTlsPort=true
+  -Ddruid.server.https.keyStorePath=/tls/server.jks
+  -Ddruid.server.https.keyStoreType=jks
+  -Ddruid.server.https.certAlias=druid
+  -Ddruid.server.https.keyManagerPassword=druid123
+  -Ddruid.server.https.keyStorePassword=druid123
+  -Ddruid.server.https.requireClientCertificate=true
+  -Ddruid.server.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.server.https.trustStorePassword=druid123
+  -Ddruid.server.https.trustStoreAlgorithm=PKIX
+  -Ddruid.server.https.validateHostnames=true
+  -Ddruid.server.https.crlPath=/tls/revocations.crl
+  -Ddruid.client.https.trustStoreAlgorithm=PKIX
+  -Ddruid.client.https.protocol=TLSv1.2
+  -Ddruid.client.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.client.https.trustStorePassword=druid123
+  -Ddruid.client.https.keyStorePath=/tls/server.jks
+  -Ddruid.client.https.certAlias=druid
+  -Ddruid.client.https.keyManagerPassword=druid123
+  -Ddruid.client.https.keyStorePassword=druid123
+  -Ddruid.startup.logging.logProperties=true
  -cp /shared/docker/lib/*
  org.apache.druid.cli.Main server middleManager
 redirect_stderr=true
--- a/integration-tests/docker/overlord.conf
+++ b/integration-tests/docker/overlord.conf
@ -9,6 +9,7 @@ command=java
  -Duser.timezone=UTC
  -Dfile.encoding=UTF-8
  -Ddruid.host=%(ENV_HOST_IP)s
+  -Ddruid.server.http.numThreads=100
  -Ddruid.metadata.storage.type=mysql
  -Ddruid.metadata.storage.connector.connectURI=jdbc:mysql://druid-metadata-storage/druid
  -Ddruid.metadata.storage.connector.user=druid
@ -30,6 +31,26 @@ command=java
  -Ddruid.escalator.authorizerName=basic
  -Ddruid.auth.authorizers="[\"basic\"]"
  -Ddruid.auth.authorizer.basic.type=basic
+  -Ddruid.enableTlsPort=true
+  -Ddruid.server.https.keyStorePath=/tls/server.jks
+  -Ddruid.server.https.keyStoreType=jks
+  -Ddruid.server.https.certAlias=druid
+  -Ddruid.server.https.keyManagerPassword=druid123
+  -Ddruid.server.https.keyStorePassword=druid123
+  -Ddruid.server.https.requireClientCertificate=true
+  -Ddruid.server.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.server.https.trustStorePassword=druid123
+  -Ddruid.server.https.trustStoreAlgorithm=PKIX
+  -Ddruid.server.https.validateHostnames=true
+  -Ddruid.server.https.crlPath=/tls/revocations.crl
+  -Ddruid.client.https.trustStoreAlgorithm=PKIX
+  -Ddruid.client.https.protocol=TLSv1.2
+  -Ddruid.client.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.client.https.trustStorePassword=druid123
+  -Ddruid.client.https.keyStorePath=/tls/server.jks
+  -Ddruid.client.https.certAlias=druid
+  -Ddruid.client.https.keyManagerPassword=druid123
+  -Ddruid.client.https.keyStorePassword=druid123
  -cp /shared/docker/lib/*
  org.apache.druid.cli.Main server overlord
 redirect_stderr=true
--- a/integration-tests/docker/router-no-client-auth-tls.conf
+++ b/integration-tests/docker/router-no-client-auth-tls.conf
@ -0,0 +1,54 @@
+[program:druid-router-no-client-auth-tls]
+command=java
+  -server
+  -Xmx128m
+  -XX:+UseConcMarkSweepGC
+  -XX:+PrintGCDetails
+  -XX:+PrintGCTimeStamps
+  -Duser.timezone=UTC
+  -Dfile.encoding=UTF-8
+  -Ddruid.host=%(ENV_HOST_IP)s
+  -Ddruid.plaintextPort=8890
+  -Ddruid.tlsPort=9090
+  -Ddruid.zk.service.host=druid-zookeeper-kafka
+  -Ddruid.server.http.numThreads=100
+  -Ddruid.lookup.numLookupLoadingThreads=1
+  -Ddruid.auth.authenticatorChain="[\"basic\"]"
+  -Ddruid.auth.authenticator.basic.type=basic
+  -Ddruid.auth.authenticator.basic.initialAdminPassword=priest
+  -Ddruid.auth.authenticator.basic.initialInternalClientPassword=warlock
+  -Ddruid.auth.authenticator.basic.authorizerName=basic
+  -Ddruid.auth.basic.common.cacheDirectory=/tmp/authCache/router-no-client-auth-tls
+  -Ddruid.escalator.type=basic
+  -Ddruid.escalator.internalClientUsername=druid_system
+  -Ddruid.escalator.internalClientPassword=warlock
+  -Ddruid.escalator.authorizerName=basic
+  -Ddruid.auth.authorizers="[\"basic\"]"
+  -Ddruid.auth.authorizer.basic.type=basic
+  -Ddruid.sql.enable=true
+  -Ddruid.sql.avatica.enable=true
+  -Ddruid.enableTlsPort=true
+  -Ddruid.server.https.keyStorePath=/tls/server.jks
+  -Ddruid.server.https.keyStoreType=jks
+  -Ddruid.server.https.certAlias=druid
+  -Ddruid.server.https.keyManagerPassword=druid123
+  -Ddruid.server.https.keyStorePassword=druid123
+  -Ddruid.server.https.requireClientCertificate=false
+  -Ddruid.server.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.server.https.trustStorePassword=druid123
+  -Ddruid.server.https.trustStoreAlgorithm=PKIX
+  -Ddruid.server.https.validateHostnames=false
+  -Ddruid.client.https.trustStoreAlgorithm=PKIX
+  -Ddruid.client.https.protocol=TLSv1.2
+  -Ddruid.client.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.client.https.trustStorePassword=druid123
+  -Ddruid.client.https.keyStorePath=/tls/server.jks
+  -Ddruid.client.https.certAlias=druid
+  -Ddruid.client.https.keyManagerPassword=druid123
+  -Ddruid.client.https.keyStorePassword=druid123
+  -cp /shared/docker/lib/*
+  org.apache.druid.cli.Main server router
+redirect_stderr=true
+priority=100
+autorestart=false
+stdout_logfile=/shared/logs/router-no-client-auth-tls.log
--- a/integration-tests/docker/router-permissive-tls.conf
+++ b/integration-tests/docker/router-permissive-tls.conf
@ -0,0 +1,54 @@
+[program:druid-router-permissive-tls]
+command=java
+  -server
+  -Xmx128m
+  -XX:+UseConcMarkSweepGC
+  -XX:+PrintGCDetails
+  -XX:+PrintGCTimeStamps
+  -Duser.timezone=UTC
+  -Dfile.encoding=UTF-8
+  -Ddruid.host=%(ENV_HOST_IP)s
+  -Ddruid.plaintextPort=8889
+  -Ddruid.tlsPort=9089
+  -Ddruid.zk.service.host=druid-zookeeper-kafka
+  -Ddruid.server.http.numThreads=100
+  -Ddruid.lookup.numLookupLoadingThreads=1
+  -Ddruid.auth.authenticatorChain="[\"basic\"]"
+  -Ddruid.auth.authenticator.basic.type=basic
+  -Ddruid.auth.authenticator.basic.initialAdminPassword=priest
+  -Ddruid.auth.authenticator.basic.initialInternalClientPassword=warlock
+  -Ddruid.auth.authenticator.basic.authorizerName=basic
+  -Ddruid.auth.basic.common.cacheDirectory=/tmp/authCache/router-permissive-tls
+  -Ddruid.escalator.type=basic
+  -Ddruid.escalator.internalClientUsername=druid_system
+  -Ddruid.escalator.internalClientPassword=warlock
+  -Ddruid.escalator.authorizerName=basic
+  -Ddruid.auth.authorizers="[\"basic\"]"
+  -Ddruid.auth.authorizer.basic.type=basic
+  -Ddruid.sql.enable=true
+  -Ddruid.sql.avatica.enable=true
+  -Ddruid.enableTlsPort=true
+  -Ddruid.server.https.keyStorePath=/tls/server.jks
+  -Ddruid.server.https.keyStoreType=jks
+  -Ddruid.server.https.certAlias=druid
+  -Ddruid.server.https.keyManagerPassword=druid123
+  -Ddruid.server.https.keyStorePassword=druid123
+  -Ddruid.server.https.requireClientCertificate=true
+  -Ddruid.server.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.server.https.trustStorePassword=druid123
+  -Ddruid.server.https.trustStoreAlgorithm=PKIX
+  -Ddruid.server.https.validateHostnames=false
+  -Ddruid.client.https.trustStoreAlgorithm=PKIX
+  -Ddruid.client.https.protocol=TLSv1.2
+  -Ddruid.client.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.client.https.trustStorePassword=druid123
+  -Ddruid.client.https.keyStorePath=/tls/server.jks
+  -Ddruid.client.https.certAlias=druid
+  -Ddruid.client.https.keyManagerPassword=druid123
+  -Ddruid.client.https.keyStorePassword=druid123
+  -cp /shared/docker/lib/*
+  org.apache.druid.cli.Main server router
+redirect_stderr=true
+priority=100
+autorestart=false
+stdout_logfile=/shared/logs/router-permissive-tls.log
--- a/integration-tests/docker/router.conf
+++ b/integration-tests/docker/router.conf
@ -25,6 +25,26 @@ command=java
  -Ddruid.auth.authorizer.basic.type=basic
  -Ddruid.sql.enable=true
  -Ddruid.sql.avatica.enable=true
+  -Ddruid.enableTlsPort=true
+  -Ddruid.server.https.keyStorePath=/tls/server.jks
+  -Ddruid.server.https.keyStoreType=jks
+  -Ddruid.server.https.certAlias=druid
+  -Ddruid.server.https.keyManagerPassword=druid123
+  -Ddruid.server.https.keyStorePassword=druid123
+  -Ddruid.server.https.requireClientCertificate=true
+  -Ddruid.server.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.server.https.trustStorePassword=druid123
+  -Ddruid.server.https.trustStoreAlgorithm=PKIX
+  -Ddruid.server.https.validateHostnames=true
+  -Ddruid.server.https.crlPath=/tls/revocations.crl
+  -Ddruid.client.https.trustStoreAlgorithm=PKIX
+  -Ddruid.client.https.protocol=TLSv1.2
+  -Ddruid.client.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.client.https.trustStorePassword=druid123
+  -Ddruid.client.https.keyStorePath=/tls/server.jks
+  -Ddruid.client.https.certAlias=druid
+  -Ddruid.client.https.keyManagerPassword=druid123
+  -Ddruid.client.https.keyStorePassword=druid123
  -cp /shared/docker/lib/*
  org.apache.druid.cli.Main server router
 redirect_stderr=true
--- a/integration-tests/docker/run-mysql.sh
+++ b/integration-tests/docker/run-mysql.sh
@ -0,0 +1,3 @@
+#!/bin/bash
+
+find /var/lib/mysql -type f -exec touch {} \; && /usr/bin/pidproxy /var/run/mysqld/mysqld.pid /usr/bin/mysqld_safe --bind-address=0.0.0.0
--- a/integration-tests/docker/tls/generate-client-certs-and-keystores.sh
+++ b/integration-tests/docker/tls/generate-client-certs-and-keystores.sh
@ -0,0 +1,21 @@
+#!/bin/bash -eu
+
+./docker/tls/generate-root-certs.sh
+
+mkdir -p client_tls
+rm -f client_tls/*
+cp docker/tls/root.key client_tls/root.key
+cp docker/tls/root.pem client_tls/root.pem
+cp docker/tls/untrusted_root.key client_tls/untrusted_root.key
+cp docker/tls/untrusted_root.pem client_tls/untrusted_root.pem
+cd client_tls
+
+../docker/tls/generate-expired-client-cert.sh
+../docker/tls/generate-good-client-cert.sh
+../docker/tls/generate-incorrect-hostname-client-cert.sh
+../docker/tls/generate-invalid-intermediate-client-cert.sh
+../docker/tls/generate-to-be-revoked-client-cert.sh
+../docker/tls/generate-untrusted-root-client-cert.sh
+../docker/tls/generate-valid-intermediate-client-cert.sh
+
+
--- a/integration-tests/docker/tls/generate-expired-client-cert.sh
+++ b/integration-tests/docker/tls/generate-expired-client-cert.sh
@ -0,0 +1,41 @@
+#!/bin/bash -eu
+
+export DOCKER_HOST_IP=$(resolveip -s $HOSTNAME)
+
+cat <<EOT > expired_csr.conf
+[req]
+default_bits = 1024
+prompt = no
+default_md = sha256
+req_extensions = req_ext
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=IntegrationTests
+emailAddress=integration-test@druid.io
+CN = localhost
+
+[ req_ext ]
+subjectAltName = @alt_names
+basicConstraints=CA:FALSE,pathlen:0
+
+[ alt_names ]
+IP.1 = ${DOCKER_HOST_IP}
+IP.2 = 127.0.0.1
+IP.3 = 172.172.172.1
+DNS.1 = ${HOSTNAME}
+DNS.2 = localhost
+EOT
+
+# Generate a client certificate for this machine
+openssl genrsa -out expired_client.key 1024 -sha256
+openssl req -new -out expired_client.csr -key expired_client.key -reqexts req_ext -config expired_csr.conf
+openssl x509 -req -days -3650 -in expired_client.csr -CA root.pem -CAkey root.key -set_serial 0x11111115 -out expired_client.pem -sha256 -extfile expired_csr.conf -extensions req_ext
+
+# Create a Java keystore containing the generated certificate
+openssl pkcs12 -export -in expired_client.pem -inkey expired_client.key -out expired_client.p12 -name expired_client -CAfile root.pem -caname druid-it-root -password pass:druid123
+keytool -importkeystore -srckeystore expired_client.p12 -srcstoretype PKCS12 -destkeystore expired_client.jks -deststoretype JKS -srcstorepass druid123 -deststorepass druid123
--- a/integration-tests/docker/tls/generate-good-client-cert.sh
+++ b/integration-tests/docker/tls/generate-good-client-cert.sh
@ -0,0 +1,44 @@
+#!/bin/bash -eu
+
+export DOCKER_HOST_IP=$(resolveip -s $HOSTNAME)
+
+cat <<EOT > csr.conf
+[req]
+default_bits = 1024
+prompt = no
+default_md = sha256
+req_extensions = req_ext
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=IntegrationTests
+emailAddress=integration-test@druid.io
+CN = localhost
+
+[ req_ext ]
+subjectAltName = @alt_names
+basicConstraints=CA:FALSE,pathlen:0
+
+[ alt_names ]
+IP.1 = ${DOCKER_HOST_IP}
+IP.2 = 127.0.0.1
+IP.3 = 172.172.172.1
+DNS.1 = ${HOSTNAME}
+DNS.2 = localhost
+EOT
+
+# Generate a client certificate for this machine
+openssl genrsa -out client.key 1024 -sha256
+openssl req -new -out client.csr -key client.key -reqexts req_ext -config csr.conf
+openssl x509 -req -days 3650 -in client.csr -CA root.pem -CAkey root.key -set_serial 0x11111111 -out client.pem -sha256 -extfile csr.conf -extensions req_ext
+
+# Create a Java keystore containing the generated certificate
+openssl pkcs12 -export -in client.pem -inkey client.key -out client.p12 -name druid -CAfile root.pem -caname druid-it-root -password pass:druid123
+keytool -importkeystore -srckeystore client.p12 -srcstoretype PKCS12 -destkeystore client.jks -deststoretype JKS -srcstorepass druid123 -deststorepass druid123
+
+# Create a Java truststore with the imply test cluster root CA
+keytool -import -alias druid-it-root -keystore truststore.jks -file root.pem -storepass druid123 -noprompt
--- a/integration-tests/docker/tls/generate-incorrect-hostname-client-cert.sh
+++ b/integration-tests/docker/tls/generate-incorrect-hostname-client-cert.sh
@ -0,0 +1,40 @@
+#!/bin/bash -eu
+
+export DOCKER_HOST_IP=$(resolveip -s $HOSTNAME)
+
+
+# Generate a client cert with an incorrect hostname for testing
+cat <<EOT > invalid_hostname_csr.conf
+[req]
+default_bits = 1024
+prompt = no
+default_md = sha256
+req_extensions = req_ext
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=IntegrationTests
+emailAddress=integration-test@druid.io
+CN = thisisprobablynottherighthostname
+
+[ req_ext ]
+subjectAltName = @alt_names
+basicConstraints=CA:FALSE,pathlen:0
+
+[ alt_names ]
+DNS.1 = thisisprobablywrongtoo
+
+EOT
+
+openssl genrsa -out invalid_hostname_client.key 1024 -sha256
+openssl req -new -out invalid_hostname_client.csr -key invalid_hostname_client.key -reqexts req_ext -config invalid_hostname_csr.conf
+openssl x509 -req -days 3650 -in invalid_hostname_client.csr -CA root.pem -CAkey root.key -set_serial 0x11111112 -out invalid_hostname_client.pem -sha256 -extfile invalid_hostname_csr.conf -extensions req_ext
+
+# Create a Java keystore containing the generated certificate
+openssl pkcs12 -export -in invalid_hostname_client.pem -inkey invalid_hostname_client.key -out invalid_hostname_client.p12 -name invalid_hostname_client -CAfile root.pem -caname druid-it-root -password pass:druid123
+keytool -importkeystore -srckeystore invalid_hostname_client.p12 -srcstoretype PKCS12 -destkeystore invalid_hostname_client.jks -deststoretype JKS -srcstorepass druid123 -deststorepass druid123
+
--- a/integration-tests/docker/tls/generate-invalid-intermediate-client-cert.sh
+++ b/integration-tests/docker/tls/generate-invalid-intermediate-client-cert.sh
@ -0,0 +1,76 @@
+#!/bin/bash -eu
+
+export DOCKER_HOST_IP=$(resolveip -s $HOSTNAME)
+
+cat <<EOT > invalid_ca_intermediate.conf
+[req]
+default_bits = 1024
+prompt = no
+default_md = sha256
+req_extensions = req_ext
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=IntegrationTests
+emailAddress=bad-intermediate@druid.io
+CN = badintermediate
+
+[ req_ext ]
+subjectAltName = @alt_names
+basicConstraints=CA:FALSE,pathlen:0
+
+[ alt_names ]
+IP.1 = 9.9.9.9
+EOT
+
+# Generate a bad intermediate certificate
+openssl genrsa -out invalid_ca_intermediate.key 1024 -sha256
+openssl req -new -out invalid_ca_intermediate.csr -key invalid_ca_intermediate.key -reqexts req_ext -config invalid_ca_intermediate.conf
+openssl x509 -req -days 3650 -in invalid_ca_intermediate.csr -CA root.pem -CAkey root.key -set_serial 0x33333331 -out invalid_ca_intermediate.pem -sha256 -extfile invalid_ca_intermediate.conf -extensions req_ext
+
+
+cat <<EOT > invalid_ca_client.conf
+[req]
+default_bits = 1024
+prompt = no
+default_md = sha256
+req_extensions = req_ext
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=IntegrationTests
+emailAddress=basic-constraint-fail@druid.io
+CN = localhost
+
+[ req_ext ]
+subjectAltName = @alt_names
+basicConstraints=CA:FALSE,pathlen:0
+
+[ alt_names ]
+IP.1 = ${DOCKER_HOST_IP}
+IP.2 = 127.0.0.1
+IP.3 = 172.172.172.1
+DNS.1 = ${HOSTNAME}
+DNS.2 = localhost
+EOT
+
+# Generate a client certificate for this machine
+openssl genrsa -out invalid_ca_client.key 1024 -sha256
+openssl req -new -out invalid_ca_client.csr -key invalid_ca_client.key -reqexts req_ext -config invalid_ca_client.conf
+openssl x509 -req -days 3650 -in invalid_ca_client.csr -CA invalid_ca_intermediate.pem -CAkey invalid_ca_intermediate.key -set_serial 0x33333333 -out invalid_ca_client.pem -sha256 -extfile invalid_ca_client.conf -extensions req_ext
+
+# Append the signing cert
+printf "\n" >> invalid_ca_client.pem
+cat invalid_ca_intermediate.pem >> invalid_ca_client.pem
+
+# Create a Java keystore containing the generated certificate
+openssl pkcs12 -export -in invalid_ca_client.pem -inkey invalid_ca_client.key -out invalid_ca_client.p12 -name invalid_ca_client -CAfile invalid_ca_intermediate.pem -caname druid-it-root -password pass:druid123
+keytool -importkeystore -srckeystore invalid_ca_client.p12 -srcstoretype PKCS12 -destkeystore invalid_ca_client.jks -deststoretype JKS -srcstorepass druid123 -deststorepass druid123
--- a/integration-tests/docker/tls/generate-root-certs.sh
+++ b/integration-tests/docker/tls/generate-root-certs.sh
@ -0,0 +1,13 @@
+#!/bin/bash -eu
+
+rm -f root.key
+rm -f untrusted_root.key
+rm -f root.pem
+rm -f untrusted_root.pem
+
+openssl genrsa -out docker/tls/root.key 4096
+openssl genrsa -out docker/tls/untrusted_root.key 4096
+
+openssl req -config docker/tls/root.cnf -key docker/tls/root.key -new -x509 -days 3650 -sha256 -extensions v3_ca -out docker/tls/root.pem
+openssl req -config docker/tls/root.cnf -key docker/tls/untrusted_root.key -new -x509 -days 3650 -sha256 -extensions v3_ca -out docker/tls/untrusted_root.pem
+
--- a/integration-tests/docker/tls/generate-server-certs-and-keystores.sh
+++ b/integration-tests/docker/tls/generate-server-certs-and-keystores.sh
@ -0,0 +1,66 @@
+#!/bin/bash -eu
+
+cd /tls
+
+rm -f cert_db.txt
+touch cert_db.txt
+
+export DOCKER_IP=$(cat /docker_ip)
+export MY_HOSTNAME=$(hostname)
+export MY_IP=$(hostname -i)
+
+cat <<EOT > csr.conf
+[req]
+default_bits = 1024
+prompt = no
+default_md = sha256
+req_extensions = req_ext
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=IntegrationTests
+emailAddress=integration-test@druid.io
+CN = ${MY_IP}
+
+[ req_ext ]
+subjectAltName = @alt_names
+basicConstraints=CA:FALSE,pathlen:0
+
+[ alt_names ]
+IP.1 = ${DOCKER_IP}
+IP.2 = ${MY_IP}
+IP.3 = 127.0.0.1
+DNS.1 = ${MY_HOSTNAME}
+DNS.2 = localhost
+
+EOT
+
+# Generate a server certificate for this machine
+openssl genrsa -out server.key 1024 -sha256
+openssl req -new -out server.csr -key server.key -reqexts req_ext -config csr.conf
+openssl x509 -req -days 3650 -in server.csr -CA root.pem -CAkey root.key -set_serial 0x22222222 -out server.pem -sha256 -extfile csr.conf -extensions req_ext
+
+# Create a Java keystore containing the generated certificate
+openssl pkcs12 -export -in server.pem -inkey server.key -out server.p12 -name druid -CAfile root.pem -caname druid-it-root -password pass:druid123
+keytool -importkeystore -srckeystore server.p12 -srcstoretype PKCS12 -destkeystore server.jks -deststoretype JKS -srcstorepass druid123 -deststorepass druid123
+
+# Create a Java truststore with the imply test cluster root CA
+keytool -import -alias druid-it-root -keystore truststore.jks -file root.pem -storepass druid123 -noprompt
+
+# Revoke one of the client certs
+openssl ca -revoke /client_tls/revoked_client.pem -config root.cnf -cert root.pem -keyfile root.key
+
+# Create the CRL
+openssl ca -gencrl -config root.cnf -cert root.pem -keyfile root.key -out /tls/revocations.crl
+
+# Generate empty CRLs for the intermediate cert test case
+rm -f cert_db2.txt
+touch cert_db2.txt
+openssl ca -gencrl -config root2.cnf -cert /client_tls/ca_intermediate.pem -keyfile /client_tls/ca_intermediate.key -out /tls/empty-revocations-intermediate.crl
+
+# Append CRLs
+cat empty-revocations-intermediate.crl >> revocations.crl
--- a/integration-tests/docker/tls/generate-to-be-revoked-client-cert.sh
+++ b/integration-tests/docker/tls/generate-to-be-revoked-client-cert.sh
@ -0,0 +1,43 @@
+#!/bin/bash -eu
+
+export DOCKER_HOST_IP=$(resolveip -s $HOSTNAME)
+
+# Generate a client cert that will be revoked
+cat <<EOT > revoked_csr.conf
+[req]
+default_bits = 1024
+prompt = no
+default_md = sha256
+req_extensions = req_ext
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=RevokedIntegrationTests
+emailAddress=revoked-it-cert@druid.io
+CN = localhost
+
+[ req_ext ]
+subjectAltName = @alt_names
+basicConstraints=CA:FALSE,pathlen:0
+
+[ alt_names ]
+IP.1 = ${DOCKER_HOST_IP}
+IP.2 = 127.0.0.1
+IP.3 = 172.172.172.1
+DNS.1 = ${HOSTNAME}
+DNS.2 = localhost
+
+EOT
+
+# Generate a client certificate for this machine
+openssl genrsa -out revoked_client.key 1024 -sha256
+openssl req -new -out revoked_client.csr -key revoked_client.key -reqexts req_ext -config revoked_csr.conf
+openssl x509 -req -days 3650 -in revoked_client.csr -CA root.pem -CAkey root.key -set_serial 0x11111113 -out revoked_client.pem -sha256 -extfile csr.conf -extensions req_ext
+
+# Create a Java keystore containing the generated certificate
+openssl pkcs12 -export -in revoked_client.pem -inkey revoked_client.key -out revoked_client.p12 -name revoked_druid -CAfile root.pem -caname druid-it-root -password pass:druid123
+keytool -importkeystore -srckeystore revoked_client.p12 -srcstoretype PKCS12 -destkeystore revoked_client.jks -deststoretype JKS -srcstorepass druid123 -deststorepass druid123
--- a/integration-tests/docker/tls/generate-untrusted-root-client-cert.sh
+++ b/integration-tests/docker/tls/generate-untrusted-root-client-cert.sh
@ -0,0 +1,42 @@
+#!/bin/bash -eu
+
+export DOCKER_HOST_IP=$(resolveip -s $HOSTNAME)
+
+cat <<EOT > csr_another_root.conf
+[req]
+default_bits = 1024
+prompt = no
+default_md = sha256
+req_extensions = req_ext
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=IntegrationTests
+emailAddress=integration-test@druid.io
+CN = localhost
+
+[ req_ext ]
+subjectAltName = @alt_names
+basicConstraints=CA:FALSE,pathlen:0
+
+[ alt_names ]
+IP.1 = ${DOCKER_HOST_IP}
+IP.2 = 127.0.0.1
+IP.3 = 172.172.172.1
+DNS.1 = ${HOSTNAME}
+DNS.2 = localhost
+EOT
+
+# Generate a client certificate for this machine
+openssl genrsa -out client_another_root.key 1024 -sha256
+openssl req -new -out client_another_root.csr -key client_another_root.key -reqexts req_ext -config csr_another_root.conf
+openssl x509 -req -days 3650 -in client_another_root.csr -CA untrusted_root.pem -CAkey untrusted_root.key -set_serial 0x11111114 -out client_another_root.pem -sha256 -extfile csr_another_root.conf -extensions req_ext
+
+# Create a Java keystore containing the generated certificate
+openssl pkcs12 -export -in client_another_root.pem -inkey client_another_root.key -out client_another_root.p12 -name druid_another_root -CAfile untrusted_root.pem -caname druid-it-untrusted-root -password pass:druid123
+keytool -importkeystore -srckeystore client_another_root.p12 -srcstoretype PKCS12 -destkeystore client_another_root.jks -deststoretype JKS -srcstorepass druid123 -deststorepass druid123
+
--- a/integration-tests/docker/tls/generate-valid-intermediate-client-cert.sh
+++ b/integration-tests/docker/tls/generate-valid-intermediate-client-cert.sh
@ -0,0 +1,76 @@
+#!/bin/bash -eu
+
+export DOCKER_HOST_IP=$(resolveip -s $HOSTNAME)
+
+cat <<EOT > ca_intermediate.conf
+[req]
+default_bits = 1024
+prompt = no
+default_md = sha256
+req_extensions = req_ext
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=IntegrationTests
+emailAddress=intermediate@druid.io
+CN = intermediate
+
+[ req_ext ]
+subjectAltName = @alt_names
+basicConstraints=CA:TRUE,pathlen:1
+
+[ alt_names ]
+IP.1 = 9.9.9.9
+EOT
+
+# Generate an intermediate certificate
+openssl genrsa -out ca_intermediate.key 1024 -sha256
+openssl req -new -out ca_intermediate.csr -key ca_intermediate.key -reqexts req_ext -config ca_intermediate.conf
+openssl x509 -req -days 3650 -in ca_intermediate.csr -CA root.pem -CAkey root.key -set_serial 0x33333332 -out ca_intermediate.pem -sha256 -extfile ca_intermediate.conf -extensions req_ext
+
+
+cat <<EOT > intermediate_ca_client.conf
+[req]
+default_bits = 1024
+prompt = no
+default_md = sha256
+req_extensions = req_ext
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=IntegrationTests
+emailAddress=intermediate-client@druid.io
+CN = localhost
+
+[ req_ext ]
+subjectAltName = @alt_names
+basicConstraints=CA:FALSE,pathlen:0
+
+[ alt_names ]
+IP.1 = ${DOCKER_HOST_IP}
+IP.2 = 127.0.0.1
+IP.3 = 172.172.172.1
+DNS.1 = ${HOSTNAME}
+DNS.2 = localhost
+EOT
+
+# Generate a client certificate for this machine
+openssl genrsa -out intermediate_ca_client.key 1024 -sha256
+openssl req -new -out intermediate_ca_client.csr -key intermediate_ca_client.key -reqexts req_ext -config intermediate_ca_client.conf
+openssl x509 -req -days 3650 -in intermediate_ca_client.csr -CA ca_intermediate.pem -CAkey ca_intermediate.key -set_serial 0x33333333 -out intermediate_ca_client.pem -sha256 -extfile intermediate_ca_client.conf -extensions req_ext
+
+# Append the signing cert
+printf "\n" >> intermediate_ca_client.pem
+cat ca_intermediate.pem >> intermediate_ca_client.pem
+
+# Create a Java keystore containing the generated certificate
+openssl pkcs12 -export -in intermediate_ca_client.pem -inkey intermediate_ca_client.key -out intermediate_ca_client.p12 -name intermediate_ca_client -CAfile ca_intermediate.pem -caname druid-it-root -password pass:druid123
+keytool -importkeystore -srckeystore intermediate_ca_client.p12 -srcstoretype PKCS12 -destkeystore intermediate_ca_client.jks -deststoretype JKS -srcstorepass druid123 -deststorepass druid123
--- a/integration-tests/docker/tls/root.cnf
+++ b/integration-tests/docker/tls/root.cnf
@ -0,0 +1,35 @@
+[ ca ]
+default_ca = CA_default
+
+[ CA_default ]
+database = /tls/cert_db.txt
+x509_extensions	= usr_cert
+name_opt = ca_default
+cert_opt = ca_default
+default_days = 365
+default_crl_days= 30
+default_md = default
+preserve = no
+policy = policy_match
+
+[req]
+default_bits = 4096
+prompt = no
+default_md = sha256
+req_extensions = v3_ca
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=IntegrationTests
+emailAddress=integration-test@druid.io
+CN = itroot
+
+[ v3_ca ]
+subjectKeyIdentifier = hash
+authorityKeyIdentifier = keyid:always,issuer
+basicConstraints = critical, CA:true
+keyUsage = critical, digitalSignature, cRLSign, keyCertSign
--- a/integration-tests/docker/tls/root2.cnf
+++ b/integration-tests/docker/tls/root2.cnf
@ -0,0 +1,13 @@
+[ ca ]
+default_ca = CA_default
+
+[ CA_default ]
+database = /tls/cert_db2.txt
+x509_extensions	= usr_cert
+name_opt = ca_default
+cert_opt = ca_default
+default_days = 365
+default_crl_days= 30
+default_md = default
+preserve = no
+policy = policy_match
--- a/integration-tests/pom.xml
+++ b/integration-tests/pom.xml
@ -88,6 +88,11 @@
            <artifactId>druid-basic-security</artifactId>
            <version>${project.parent.version}</version>
        </dependency>
+        <dependency>
+            <groupId>org.apache.druid.extensions</groupId>
+            <artifactId>simple-client-sslcontext</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
        <dependency>
            <groupId>org.apache.druid</groupId>
            <artifactId>druid-services</artifactId>
@ -227,6 +232,12 @@
                                -Ddruid.test.config.dockerIp=${env.DOCKER_IP}
                                -Ddruid.test.config.hadoopDir=${env.HADOOP_DIR}
                                -Ddruid.zk.service.host=${env.DOCKER_IP}
+                                -Ddruid.client.https.trustStorePath=client_tls/truststore.jks
+                                -Ddruid.client.https.trustStorePassword=druid123
+                                -Ddruid.client.https.keyStorePath=client_tls/client.jks
+                                -Ddruid.client.https.certAlias=druid
+                                -Ddruid.client.https.keyManagerPassword=druid123
+                                -Ddruid.client.https.keyStorePassword=druid123
                            </argLine>
                            <suiteXmlFiles>
                                <suiteXmlFile>src/test/resources/testng.xml</suiteXmlFile>
@ -276,6 +287,12 @@
                                -Dfile.encoding=UTF-8
                                -Ddruid.test.config.type=configFile
                                -Ddruid.test.config.configFile=${env.CONFIG_FILE}
+                                -Ddruid.client.https.trustStorePath=client_tls/truststore.jks
+                                -Ddruid.client.https.trustStorePassword=druid123
+                                -Ddruid.client.https.keyStorePath=client_tls/client.jks
+                                -Ddruid.client.https.certAlias=druid
+                                -Ddruid.client.https.keyManagerPassword=druid123
+                                -Ddruid.client.https.keyStorePassword=druid123
                            </argLine>
                            <suiteXmlFiles>
                                <suiteXmlFile>src/test/resources/testng.xml</suiteXmlFile>
--- a/integration-tests/run_cluster.sh
+++ b/integration-tests/run_cluster.sh
@ -15,12 +15,14 @@
 # limitations under the License.

 # cleanup
-for node in druid-historical druid-coordinator druid-overlord druid-router druid-broker druid-middlemanager druid-zookeeper-kafka druid-metadata-storage;
+for node in druid-historical druid-coordinator druid-overlord druid-router druid-router-permissive-tls druid-router-no-client-auth-tls druid-broker druid-middlemanager druid-zookeeper-kafka druid-metadata-storage;
 do
 docker stop $node
 docker rm $node
 done

+docker network rm druid-it-net
+
 # environment variables
 DIR=$(cd $(dirname $0) && pwd)
 DOCKERDIR=$DIR/docker
@ -31,6 +33,11 @@ RESOURCEDIR=$DIR/src/test/resources
 # so docker IP addr will be known during docker build
 echo ${DOCKER_IP:=127.0.0.1} > $DOCKERDIR/docker_ip

+# setup client keystore
+./docker/tls/generate-client-certs-and-keystores.sh
+rm -rf docker/client_tls
+cp -r client_tls docker/client_tls
+
 # Make directories if they dont exist
 mkdir -p $SHARED_DIR/logs
 mkdir -p $SHARED_DIR/tasklogs
@ -40,29 +47,37 @@ rm -rf $SHARED_DIR/docker
 cp -R docker $SHARED_DIR/docker
 mvn -B dependency:copy-dependencies -DoutputDirectory=$SHARED_DIR/docker/lib

+docker network create --subnet=172.172.172.0/24 druid-it-net
+
 # Build Druid Cluster Image
 docker build -t druid/cluster $SHARED_DIR/docker

 # Start zookeeper and kafka
-docker run -d --privileged --name druid-zookeeper-kafka -p 2181:2181 -p 9092:9092 -p 9093:9093 -v $SHARED_DIR:/shared -v $DOCKERDIR/zookeeper.conf:$SUPERVISORDIR/zookeeper.conf -v $DOCKERDIR/kafka.conf:$SUPERVISORDIR/kafka.conf druid/cluster
+docker run -d --privileged --net druid-it-net --ip 172.172.172.2 --name druid-zookeeper-kafka -p 2181:2181 -p 9092:9092 -p 9093:9093 -v $SHARED_DIR:/shared -v $DOCKERDIR/zookeeper.conf:$SUPERVISORDIR/zookeeper.conf -v $DOCKERDIR/kafka.conf:$SUPERVISORDIR/kafka.conf druid/cluster

 # Start MYSQL 
-docker run -d --privileged --name druid-metadata-storage -v $SHARED_DIR:/shared -v $DOCKERDIR/metadata-storage.conf:$SUPERVISORDIR/metadata-storage.conf druid/cluster
+docker run -d --privileged --net druid-it-net --ip 172.172.172.3 --name druid-metadata-storage -v $SHARED_DIR:/shared -v $DOCKERDIR/metadata-storage.conf:$SUPERVISORDIR/metadata-storage.conf druid/cluster

 # Start Overlord
-docker run -d --privileged --name druid-overlord -p 8090:8090 -v $SHARED_DIR:/shared -v $DOCKERDIR/overlord.conf:$SUPERVISORDIR/overlord.conf --link druid-metadata-storage:druid-metadata-storage --link druid-zookeeper-kafka:druid-zookeeper-kafka druid/cluster
+docker run -d --privileged --net druid-it-net --ip 172.172.172.4 --name druid-overlord -p 8090:8090 -p 8290:8290 -v $SHARED_DIR:/shared -v $DOCKERDIR/overlord.conf:$SUPERVISORDIR/overlord.conf --link druid-metadata-storage:druid-metadata-storage --link druid-zookeeper-kafka:druid-zookeeper-kafka druid/cluster

 # Start Coordinator
-docker run -d --privileged --name druid-coordinator -p 8081:8081 -v $SHARED_DIR:/shared -v $DOCKERDIR/coordinator.conf:$SUPERVISORDIR/coordinator.conf --link druid-overlord:druid-overlord --link druid-metadata-storage:druid-metadata-storage --link druid-zookeeper-kafka:druid-zookeeper-kafka druid/cluster
+docker run -d --privileged --net druid-it-net --ip 172.172.172.5 --name druid-coordinator -p 8081:8081 -p 8281:8281 -v $SHARED_DIR:/shared -v $DOCKERDIR/coordinator.conf:$SUPERVISORDIR/coordinator.conf --link druid-overlord:druid-overlord --link druid-metadata-storage:druid-metadata-storage --link druid-zookeeper-kafka:druid-zookeeper-kafka druid/cluster

 # Start Historical 
-docker run -d --privileged --name druid-historical -p 8083:8083 -v $SHARED_DIR:/shared -v $DOCKERDIR/historical.conf:$SUPERVISORDIR/historical.conf --link druid-zookeeper-kafka:druid-zookeeper-kafka druid/cluster
+docker run -d --privileged --net druid-it-net --ip 172.172.172.6 --name druid-historical -p 8083:8083 -p 8283:8283 -v $SHARED_DIR:/shared -v $DOCKERDIR/historical.conf:$SUPERVISORDIR/historical.conf --link druid-zookeeper-kafka:druid-zookeeper-kafka druid/cluster

 # Start Middlemanger
-docker run -d --privileged --name druid-middlemanager -p 8091:8091 -p 8100:8100 -p 8101:8101 -p 8102:8102 -p 8103:8103 -p 8104:8104 -p 8105:8105 -v $RESOURCEDIR:/resources -v $SHARED_DIR:/shared -v $DOCKERDIR/middlemanager.conf:$SUPERVISORDIR/middlemanager.conf --link druid-zookeeper-kafka:druid-zookeeper-kafka --link druid-overlord:druid-overlord druid/cluster
+docker run -d --privileged --net druid-it-net --ip 172.172.172.7 --name druid-middlemanager -p 8091:8091 -p 8291:8291 -p 8100:8100 -p 8101:8101 -p 8102:8102 -p 8103:8103 -p 8104:8104 -p 8105:8105 -p 8300:8300 -p 8301:8301 -p 8302:8302 -p 8303:8303 -p 8304:8304 -p 8305:8305 -v $RESOURCEDIR:/resources -v $SHARED_DIR:/shared -v $DOCKERDIR/middlemanager.conf:$SUPERVISORDIR/middlemanager.conf --link druid-zookeeper-kafka:druid-zookeeper-kafka --link druid-overlord:druid-overlord druid/cluster

 # Start Broker 
-docker run -d --privileged --name druid-broker -p 8082:8082 -v $SHARED_DIR:/shared -v $DOCKERDIR/broker.conf:$SUPERVISORDIR/broker.conf --link druid-zookeeper-kafka:druid-zookeeper-kafka --link druid-middlemanager:druid-middlemanager --link druid-historical:druid-historical druid/cluster
+docker run -d --privileged --net druid-it-net --ip 172.172.172.8 --name druid-broker -p 8082:8082 -p 8282:8282 -v $SHARED_DIR:/shared -v $DOCKERDIR/broker.conf:$SUPERVISORDIR/broker.conf --link druid-zookeeper-kafka:druid-zookeeper-kafka --link druid-middlemanager:druid-middlemanager --link druid-historical:druid-historical druid/cluster

-# Start Router 
-docker run -d --privileged --name druid-router -p 8888:8888 -v $SHARED_DIR:/shared -v $DOCKERDIR/router.conf:$SUPERVISORDIR/router.conf --link druid-zookeeper-kafka:druid-zookeeper-kafka --link druid-coordinator:druid-coordinator --link druid-broker:druid-broker druid/cluster
+# Start Router
+docker run -d --privileged --net druid-it-net --ip 172.172.172.9 --name druid-router -p 8888:8888 -p 9088:9088 -v $SHARED_DIR:/shared -v $DOCKERDIR/router.conf:$SUPERVISORDIR/router.conf --link druid-zookeeper-kafka:druid-zookeeper-kafka --link druid-coordinator:druid-coordinator --link druid-broker:druid-broker druid/cluster
+
+# Start Router with permissive TLS settings (client auth enabled, no hostname verification, no revocation check)
+docker run -d --privileged --net druid-it-net --ip 172.172.172.10 --name druid-router-permissive-tls -p 8889:8889 -p 9089:9089 -v $SHARED_DIR:/shared -v $DOCKERDIR/router-permissive-tls.conf:$SUPERVISORDIR/router-permissive-tls.conf --link druid-zookeeper-kafka:druid-zookeeper-kafka --link druid-coordinator:druid-coordinator --link druid-broker:druid-broker druid/cluster
+
+# Start Router with TLS but no client auth
+docker run -d --privileged --net druid-it-net --ip 172.172.172.11 --name druid-router-no-client-auth-tls -p 8890:8890 -p 9090:9090 -v $SHARED_DIR:/shared -v $DOCKERDIR/router-no-client-auth-tls.conf:$SUPERVISORDIR/router-no-client-auth-tls.conf --link druid-zookeeper-kafka:druid-zookeeper-kafka --link druid-coordinator:druid-coordinator --link druid-broker:druid-broker druid/cluster
--- a/integration-tests/src/main/java/org/apache/druid/testing/ConfigFileConfigProvider.java
+++ b/integration-tests/src/main/java/org/apache/druid/testing/ConfigFileConfigProvider.java
@ -38,6 +38,15 @@ public class ConfigFileConfigProvider implements IntegrationTestingConfigProvide
  private String historicalUrl;
  private String coordinatorUrl;
  private String indexerUrl;
+  private String permissiveRouterUrl;
+  private String noClientAuthRouterUrl;
+  private String routerTLSUrl;
+  private String brokerTLSUrl;
+  private String historicalTLSUrl;
+  private String coordinatorTLSUrl;
+  private String indexerTLSUrl;
+  private String permissiveRouterTLSUrl;
+  private String noClientAuthRouterTLSUrl;
  private String middleManagerHost;
  private String zookeeperHosts;        // comma-separated list of host:port
  private String kafkaHost;
@ -70,25 +79,89 @@ public class ConfigFileConfigProvider implements IntegrationTestingConfigProvide
        routerUrl = StringUtils.format("http://%s:%s", routerHost, props.get("router_port"));
      }
    }
+    routerTLSUrl = props.get("router_tls_url");
+    if (routerTLSUrl == null) {
+      String routerHost = props.get("router_host");
+      if (null != routerHost) {
+        routerTLSUrl = StringUtils.format("https://%s:%s", routerHost, props.get("router_tls_port"));
+      }
+    }
+    permissiveRouterUrl = props.get("router_permissive_url");
+    if (permissiveRouterUrl == null) {
+      String permissiveRouterHost = props.get("router_permissive_host");
+      if (null != permissiveRouterHost) {
+        permissiveRouterUrl = StringUtils.format("http://%s:%s", permissiveRouterHost, props.get("router_permissive_port"));
+      }
+    }
+    permissiveRouterTLSUrl = props.get("router_permissive_tls_url");
+    if (permissiveRouterTLSUrl == null) {
+      String permissiveRouterHost = props.get("router_permissive_host");
+      if (null != permissiveRouterHost) {
+        permissiveRouterTLSUrl = StringUtils.format("https://%s:%s", permissiveRouterHost, props.get("router_permissive_tls_port"));
+      }
+    }
+    noClientAuthRouterUrl = props.get("router_no_client_auth_url");
+    if (noClientAuthRouterUrl == null) {
+      String noClientAuthRouterHost = props.get("router_no_client_auth_host");
+      if (null != noClientAuthRouterHost) {
+        noClientAuthRouterUrl = StringUtils.format("http://%s:%s", noClientAuthRouterHost, props.get("router_no_client_auth_port"));
+      }
+    }
+    noClientAuthRouterTLSUrl = props.get("router_no_client_auth_tls_url");
+    if (noClientAuthRouterTLSUrl == null) {
+      String noClientAuthRouterHost = props.get("router_no_client_auth_host");
+      if (null != noClientAuthRouterHost) {
+        noClientAuthRouterTLSUrl = StringUtils.format("https://%s:%s", noClientAuthRouterHost, props.get("router_no_client_auth_tls_port"));
+      }
+    }
    brokerUrl = props.get("broker_url");
    if (brokerUrl == null) {
      brokerUrl = StringUtils.format("http://%s:%s", props.get("broker_host"), props.get("broker_port"));
    }
-
+    brokerTLSUrl = props.get("broker_tls_url");
+    if (brokerTLSUrl == null) {
+      String brokerHost = props.get("broker_host");
+      if (null != brokerHost) {
+        brokerTLSUrl = StringUtils.format("https://%s:%s", brokerHost, props.get("broker_tls_port"));
+      }
+    }
+    
    historicalUrl = props.get("historical_url");
    if (historicalUrl == null) {
      historicalUrl = StringUtils.format("http://%s:%s", props.get("historical_host"), props.get("historical_port"));
    }
+    historicalTLSUrl = props.get("historical_tls_url");
+    if (historicalTLSUrl == null) {
+      String historicalHost = props.get("historical_host");
+      if (null != historicalHost) {
+        historicalTLSUrl = StringUtils.format("https://%s:%s", historicalHost, props.get("historical_tls_port"));
+      }
+    }

    coordinatorUrl = props.get("coordinator_url");
    if (coordinatorUrl == null) {
      coordinatorUrl = StringUtils.format("http://%s:%s", props.get("coordinator_host"), props.get("coordinator_port"));
    }
+    coordinatorTLSUrl = props.get("coordinator_tls_url");
+    if (coordinatorTLSUrl == null) {
+      String coordinatorHost = props.get("coordinator_host");
+      if (null != coordinatorHost) {
+        coordinatorTLSUrl = StringUtils.format("https://%s:%s", coordinatorHost, props.get("coordinator_tls_port"));
+      }
+    }

    indexerUrl = props.get("indexer_url");
    if (indexerUrl == null) {
      indexerUrl = StringUtils.format("http://%s:%s", props.get("indexer_host"), props.get("indexer_port"));
    }
+    indexerTLSUrl = props.get("indexer_tls_url");
+    if (indexerTLSUrl == null) {
+      String indexerHost = props.get("indexer_host");
+      if (null != indexerHost) {
+        indexerTLSUrl = StringUtils.format("https://%s:%s", indexerHost, props.get("indexer_tls_port"));
+      }
+    }
+    
    middleManagerHost = props.get("middlemanager_host");

    zookeeperHosts = props.get("zookeeper_hosts");
@ -98,10 +171,11 @@ public class ConfigFileConfigProvider implements IntegrationTestingConfigProvide

    password = props.get("password");

-    LOG.info("router: [%s]", routerUrl);
-    LOG.info("broker: [%s]", brokerUrl);
-    LOG.info("coordinator: [%s]", coordinatorUrl);
-    LOG.info("overlord: [%s]", indexerUrl);
+    LOG.info("router: [%s], [%s]", routerUrl, routerTLSUrl);
+    LOG.info("broker: [%s], [%s]", brokerUrl, brokerTLSUrl);
+    LOG.info("historical: [%s], [%s]", historicalUrl, historicalTLSUrl);
+    LOG.info("coordinator: [%s], [%s]", coordinatorUrl, coordinatorTLSUrl);
+    LOG.info("overlord: [%s], [%s]", indexerUrl, indexerTLSUrl);
    LOG.info("middle manager: [%s]", middleManagerHost);
    LOG.info("zookeepers: [%s]", zookeeperHosts);
    LOG.info("kafka: [%s]", kafkaHost);
@ -120,30 +194,84 @@ public class ConfigFileConfigProvider implements IntegrationTestingConfigProvide
        return coordinatorUrl;
      }

+      @Override
+      public String getCoordinatorTLSUrl()
+      {
+        return coordinatorTLSUrl;
+      }
+
      @Override
      public String getIndexerUrl()
      {
        return indexerUrl;
      }

+      @Override
+      public String getIndexerTLSUrl()
+      {
+        return indexerTLSUrl;
+      }
+
      @Override
      public String getRouterUrl()
      {
        return routerUrl;
      }

+      @Override
+      public String getRouterTLSUrl()
+      {
+        return routerTLSUrl;
+      }
+
+      @Override
+      public String getPermissiveRouterUrl()
+      {
+        return permissiveRouterUrl;
+      }
+
+      @Override
+      public String getPermissiveRouterTLSUrl()
+      {
+        return permissiveRouterTLSUrl;
+      }
+
+      @Override
+      public String getNoClientAuthRouterUrl()
+      {
+        return noClientAuthRouterUrl;
+      }
+
+      @Override
+      public String getNoClientAuthRouterTLSUrl()
+      {
+        return noClientAuthRouterTLSUrl;
+      }
+
      @Override
      public String getBrokerUrl()
      {
        return brokerUrl;
      }

+      @Override
+      public String getBrokerTLSUrl()
+      {
+        return brokerTLSUrl;
+      }
+
      @Override
      public String getHistoricalUrl()
      {
        return historicalUrl;
      }

+      @Override
+      public String getHistoricalTLSUrl()
+      {
+        return historicalTLSUrl;
+      }
+
      @Override
      public String getMiddleManagerHost()
      {
--- a/integration-tests/src/main/java/org/apache/druid/testing/DockerConfigProvider.java
+++ b/integration-tests/src/main/java/org/apache/druid/testing/DockerConfigProvider.java
@ -48,30 +48,84 @@ public class DockerConfigProvider implements IntegrationTestingConfigProvider
        return "http://" + dockerIp + ":8081";
      }

+      @Override
+      public String getCoordinatorTLSUrl()
+      {
+        return "https://" + dockerIp + ":8281";
+      }
+
      @Override
      public String getIndexerUrl()
      {
        return "http://" + dockerIp + ":8090";
      }

+      @Override
+      public String getIndexerTLSUrl()
+      {
+        return "https://" + dockerIp + ":8290";
+      }
+
      @Override
      public String getRouterUrl()
      {
        return "http://" + dockerIp + ":8888";
      }

+      @Override
+      public String getRouterTLSUrl()
+      {
+        return "https://" + dockerIp + ":9088";
+      }
+
+      @Override
+      public String getPermissiveRouterUrl()
+      {
+        return "http://" + dockerIp + ":8889";
+      }
+
+      @Override
+      public String getPermissiveRouterTLSUrl()
+      {
+        return "https://" + dockerIp + ":9089";
+      }
+
+      @Override
+      public String getNoClientAuthRouterUrl()
+      {
+        return "http://" + dockerIp + ":8890";
+      }
+
+      @Override
+      public String getNoClientAuthRouterTLSUrl()
+      {
+        return "https://" + dockerIp + ":9090";
+      }
+
      @Override
      public String getBrokerUrl()
      {
        return "http://" + dockerIp + ":8082";
      }

+      @Override
+      public String getBrokerTLSUrl()
+      {
+        return "https://" + dockerIp + ":8282";
+      }
+
      @Override
      public String getHistoricalUrl()
      {
        return "http://" + dockerIp + ":8083";
      }

+      @Override
+      public String getHistoricalTLSUrl()
+      {
+        return "https://" + dockerIp + ":8283";
+      }
+
      @Override
      public String getMiddleManagerHost()
      {
--- a/integration-tests/src/main/java/org/apache/druid/testing/IntegrationTestingConfig.java
+++ b/integration-tests/src/main/java/org/apache/druid/testing/IntegrationTestingConfig.java
@ -27,14 +27,32 @@ public interface IntegrationTestingConfig
 {
  String getCoordinatorUrl();

+  String getCoordinatorTLSUrl();
+
  String getIndexerUrl();

+  String getIndexerTLSUrl();
+
  String getRouterUrl();

+  String getRouterTLSUrl();
+
+  String getPermissiveRouterUrl();
+
+  String getPermissiveRouterTLSUrl();
+
+  String getNoClientAuthRouterUrl();
+
+  String getNoClientAuthRouterTLSUrl();
+
  String getBrokerUrl();

+  String getBrokerTLSUrl();
+
  String getHistoricalUrl();

+  String getHistoricalTLSUrl();
+
  String getMiddleManagerHost();

  String getZookeeperHosts();
--- a/integration-tests/src/main/java/org/apache/druid/testing/clients/CoordinatorResourceTestClient.java
+++ b/integration-tests/src/main/java/org/apache/druid/testing/clients/CoordinatorResourceTestClient.java
@ -70,6 +70,11 @@ public class CoordinatorResourceTestClient
    );
  }

+  private String getMetadataSegmentsURL(String dataSource)
+  {
+    return StringUtils.format("%smetadata/datasources/%s/segments", getCoordinatorURL(), dataSource);
+  }
+
  private String getIntervalsURL(String dataSource)
  {
    return StringUtils.format("%sdatasources/%s/intervals", getCoordinatorURL(), dataSource);
@ -80,6 +85,25 @@ public class CoordinatorResourceTestClient
    return StringUtils.format("%s%s", getCoordinatorURL(), "loadstatus");
  }

+  // return a list of the segment dates for the specified datasource
+  public List<String> getMetadataSegments(final String dataSource)
+  {
+    ArrayList<String> segments = null;
+    try {
+      StatusResponseHolder response = makeRequest(HttpMethod.GET, getMetadataSegmentsURL(dataSource));
+
+      segments = jsonMapper.readValue(
+          response.getContent(), new TypeReference<ArrayList<String>>()
+          {
+          }
+      );
+    }
+    catch (Exception e) {
+      throw Throwables.propagate(e);
+    }
+    return segments;
+  }
+
  // return a list of the segment dates for the specified datasource
  public List<String> getSegmentIntervals(final String dataSource)
  {
--- a/integration-tests/src/main/java/org/apache/druid/testing/clients/EventReceiverFirehoseTestClient.java
+++ b/integration-tests/src/main/java/org/apache/druid/testing/clients/EventReceiverFirehoseTestClient.java
@ -71,7 +71,7 @@ public class EventReceiverFirehoseTestClient
  private String getURL()
  {
    return StringUtils.format(
-        "http://%s/druid/worker/v1/chat/%s/push-events/",
+        "https://%s/druid/worker/v1/chat/%s/push-events/",
        host,
        chatID
    );
--- a/integration-tests/src/main/java/org/apache/druid/testing/clients/OverlordResourceTestClient.java
+++ b/integration-tests/src/main/java/org/apache/druid/testing/clients/OverlordResourceTestClient.java
@ -175,7 +175,7 @@ public class OverlordResourceTestClient

  public void waitUntilTaskCompletes(final String taskID)
  {
-    waitUntilTaskCompletes(taskID, 60000, 10);
+    waitUntilTaskCompletes(taskID, 10000, 60);
  }

  public void waitUntilTaskCompletes(final String taskID, final int millisEach, final int numTimes)
--- a/integration-tests/src/main/java/org/apache/druid/testing/utils/RetryUtil.java
+++ b/integration-tests/src/main/java/org/apache/druid/testing/utils/RetryUtil.java
@ -32,9 +32,9 @@ public class RetryUtil

  private static final Logger LOG = new Logger(RetryUtil.class);

-  public static int DEFAULT_RETRY_COUNT = 10;
+  public static int DEFAULT_RETRY_COUNT = 30;

-  public static long DEFAULT_RETRY_SLEEP = TimeUnit.SECONDS.toMillis(30);
+  public static long DEFAULT_RETRY_SLEEP = TimeUnit.SECONDS.toMillis(10);

  public static void retryUntilTrue(Callable<Boolean> callable, String task)
  {
--- a/integration-tests/src/main/java/org/apache/druid/testing/utils/TestQueryHelper.java
+++ b/integration-tests/src/main/java/org/apache/druid/testing/utils/TestQueryHelper.java
@ -42,6 +42,9 @@ public class TestQueryHelper
  private final QueryResourceTestClient queryClient;
  private final ObjectMapper jsonMapper;
  private final String broker;
+  private final String brokerTLS;
+  private final String router;
+  private final String routerTLS;

  @Inject
  TestQueryHelper(
@ -53,11 +56,17 @@ public class TestQueryHelper
    this.jsonMapper = jsonMapper;
    this.queryClient = queryClient;
    this.broker = config.getBrokerUrl();
+    this.brokerTLS = config.getBrokerTLSUrl();
+    this.router = config.getRouterUrl();
+    this.routerTLS = config.getRouterTLSUrl();
  }

  public void testQueriesFromFile(String filePath, int timesToRun) throws Exception
  {
-    testQueriesFromFile(getBrokerURL(), filePath, timesToRun);
+    testQueriesFromFile(getQueryURL(broker), filePath, timesToRun);
+    testQueriesFromFile(getQueryURL(brokerTLS), filePath, timesToRun);
+    testQueriesFromFile(getQueryURL(router), filePath, timesToRun);
+    testQueriesFromFile(getQueryURL(routerTLS), filePath, timesToRun);
  }

  public void testQueriesFromFile(String url, String filePath, int timesToRun) throws Exception
@ -75,7 +84,10 @@ public class TestQueryHelper

  public void testQueriesFromString(String str, int timesToRun) throws Exception
  {
-    testQueriesFromString(getBrokerURL(), str, timesToRun);
+    testQueriesFromString(getQueryURL(broker), str, timesToRun);
+    testQueriesFromString(getQueryURL(brokerTLS), str, timesToRun);
+    testQueriesFromString(getQueryURL(router), str, timesToRun);
+    testQueriesFromString(getQueryURL(routerTLS), str, timesToRun);
  }

  public void testQueriesFromString(String url, String str, int timesToRun) throws Exception
@ -93,6 +105,7 @@ public class TestQueryHelper

  private void testQueries(String url, List<QueryWithResults> queries, int timesToRun) throws Exception
  {
+    LOG.info("Running queries, url [%s]", url);
    for (int i = 0; i < timesToRun; i++) {
      LOG.info("Starting Iteration %d", i);

@ -119,9 +132,9 @@ public class TestQueryHelper
    }
  }

-  private String getBrokerURL()
+  private String getQueryURL(String schemeAndHost)
  {
-    return StringUtils.format("%s/druid/v2?pretty", broker);
+    return StringUtils.format("%s/druid/v2?pretty", schemeAndHost);
  }

  @SuppressWarnings("unchecked")
@ -138,7 +151,7 @@ public class TestQueryHelper
                                  .intervals(interval)
                                  .build();

-    List<Map<String, Object>> results = queryClient.query(getBrokerURL(), query);
+    List<Map<String, Object>> results = queryClient.query(getQueryURL(broker), query);
    if (results.isEmpty()) {
      return 0;
    } else {
--- a/integration-tests/src/test/java/io/druid/tests/security/ITTLSTest.java
+++ b/integration-tests/src/test/java/io/druid/tests/security/ITTLSTest.java
@ -0,0 +1,482 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.druid.tests.security;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Throwables;
+import com.google.inject.Inject;
+import org.apache.druid.guice.annotations.Client;
+import org.apache.druid.guice.http.DruidHttpClientConfig;
+import org.apache.druid.guice.http.LifecycleUtils;
+import org.apache.druid.https.SSLClientConfig;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.lifecycle.Lifecycle;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.java.util.http.client.CredentialedHttpClient;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.HttpClientConfig;
+import org.apache.druid.java.util.http.client.HttpClientInit;
+import org.apache.druid.java.util.http.client.Request;
+import org.apache.druid.java.util.http.client.auth.BasicCredentials;
+import org.apache.druid.java.util.http.client.response.StatusResponseHandler;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.server.security.TLSUtils;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.guice.DruidTestModuleFactory;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+import org.joda.time.Duration;
+import org.testng.Assert;
+import org.testng.annotations.Guice;
+import org.testng.annotations.Test;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLException;
+import javax.ws.rs.core.MediaType;
+import java.io.IOException;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+
+@Guice(moduleFactory = DruidTestModuleFactory.class)
+public class ITTLSTest
+{
+  private static final Logger LOG = new Logger(ITTLSTest.class);
+
+  private static final Duration SSL_HANDSHAKE_TIMEOUT = new Duration(30 * 1000);
+
+  private static final int MAX_BROKEN_PIPE_RETRIES = 30;
+
+  @Inject
+  IntegrationTestingConfig config;
+
+  @Inject
+  ObjectMapper jsonMapper;
+
+  @Inject
+  SSLClientConfig sslClientConfig;
+
+  @Inject
+  @Client
+  HttpClient httpClient;
+
+  @Inject
+  @Client
+  DruidHttpClientConfig httpClientConfig;
+
+  StatusResponseHandler responseHandler = new StatusResponseHandler(StandardCharsets.UTF_8);
+
+  @Test
+  public void testPlaintextAccess()
+  {
+    LOG.info("---------Testing resource access without TLS---------");
+    HttpClient adminClient = new CredentialedHttpClient(
+        new BasicCredentials("admin", "priest"),
+        httpClient
+    );
+    makeRequest(adminClient, HttpMethod.GET, config.getCoordinatorUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getIndexerUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getBrokerUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getHistoricalUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getRouterUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getPermissiveRouterUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getNoClientAuthRouterUrl() + "/status", null);
+  }
+
+  @Test
+  public void testTLSNodeAccess()
+  {
+    LOG.info("---------Testing resource access with TLS enabled---------");
+    HttpClient adminClient = new CredentialedHttpClient(
+        new BasicCredentials("admin", "priest"),
+        httpClient
+    );
+    makeRequest(adminClient, HttpMethod.GET, config.getCoordinatorTLSUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getIndexerTLSUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getBrokerTLSUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getHistoricalTLSUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getRouterTLSUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getPermissiveRouterTLSUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getNoClientAuthRouterTLSUrl() + "/status", null);
+  }
+
+  @Test
+  public void testTLSNodeAccessWithIntermediate()
+  {
+    LOG.info("---------Testing TLS resource access with 3-part cert chain---------");
+    HttpClient intermediateCertClient = makeCustomHttpClient(
+        "client_tls/intermediate_ca_client.jks",
+        "intermediate_ca_client"
+    );
+    makeRequest(intermediateCertClient, HttpMethod.GET, config.getCoordinatorTLSUrl() + "/status", null);
+    makeRequest(intermediateCertClient, HttpMethod.GET, config.getIndexerTLSUrl() + "/status", null);
+    makeRequest(intermediateCertClient, HttpMethod.GET, config.getBrokerTLSUrl() + "/status", null);
+    makeRequest(intermediateCertClient, HttpMethod.GET, config.getHistoricalTLSUrl() + "/status", null);
+    makeRequest(intermediateCertClient, HttpMethod.GET, config.getRouterTLSUrl() + "/status", null);
+    makeRequest(intermediateCertClient, HttpMethod.GET, config.getPermissiveRouterTLSUrl() + "/status", null);
+    makeRequest(intermediateCertClient, HttpMethod.GET, config.getNoClientAuthRouterTLSUrl() + "/status", null);
+  }
+
+  @Test
+  public void checkAccessWithNoCert()
+  {
+    LOG.info("---------Testing TLS resource access without a certificate---------");
+    HttpClient certlessClient = makeCertlessClient();
+    checkFailedAccessNoCert(certlessClient, HttpMethod.GET, config.getCoordinatorTLSUrl());
+    checkFailedAccessNoCert(certlessClient, HttpMethod.GET, config.getIndexerTLSUrl());
+    checkFailedAccessNoCert(certlessClient, HttpMethod.GET, config.getBrokerTLSUrl());
+    checkFailedAccessNoCert(certlessClient, HttpMethod.GET, config.getHistoricalTLSUrl());
+    checkFailedAccessNoCert(certlessClient, HttpMethod.GET, config.getRouterTLSUrl());
+    checkFailedAccessNoCert(certlessClient, HttpMethod.GET, config.getPermissiveRouterTLSUrl());
+    makeRequest(certlessClient, HttpMethod.GET, config.getNoClientAuthRouterTLSUrl() + "/status", null);
+  }
+
+  @Test
+  public void checkAccessWithWrongHostname()
+  {
+    LOG.info("---------Testing TLS resource access when client certificate has non-matching hostnames---------");
+    HttpClient wrongHostnameClient = makeCustomHttpClient(
+        "client_tls/invalid_hostname_client.jks",
+        "invalid_hostname_client"
+    );
+    checkFailedAccessWrongHostname(wrongHostnameClient, HttpMethod.GET, config.getCoordinatorTLSUrl());
+    checkFailedAccessWrongHostname(wrongHostnameClient, HttpMethod.GET, config.getIndexerTLSUrl());
+    checkFailedAccessWrongHostname(wrongHostnameClient, HttpMethod.GET, config.getBrokerTLSUrl());
+    checkFailedAccessWrongHostname(wrongHostnameClient, HttpMethod.GET, config.getHistoricalTLSUrl());
+    checkFailedAccessWrongHostname(wrongHostnameClient, HttpMethod.GET, config.getRouterTLSUrl());
+    makeRequest(wrongHostnameClient, HttpMethod.GET, config.getPermissiveRouterTLSUrl() + "/status", null);
+    makeRequest(wrongHostnameClient, HttpMethod.GET, config.getNoClientAuthRouterTLSUrl() + "/status", null);
+  }
+
+  @Test
+  public void checkAccessWithWrongRoot()
+  {
+    LOG.info("---------Testing TLS resource access when client certificate is signed by a non-trusted root CA---------");
+    HttpClient wrongRootClient = makeCustomHttpClient(
+        "client_tls/client_another_root.jks",
+        "druid_another_root"
+    );
+    checkFailedAccessWrongRoot(wrongRootClient, HttpMethod.GET, config.getCoordinatorTLSUrl());
+    checkFailedAccessWrongRoot(wrongRootClient, HttpMethod.GET, config.getIndexerTLSUrl());
+    checkFailedAccessWrongRoot(wrongRootClient, HttpMethod.GET, config.getBrokerTLSUrl());
+    checkFailedAccessWrongRoot(wrongRootClient, HttpMethod.GET, config.getHistoricalTLSUrl());
+    checkFailedAccessWrongRoot(wrongRootClient, HttpMethod.GET, config.getRouterTLSUrl());
+    checkFailedAccessWrongRoot(wrongRootClient, HttpMethod.GET, config.getPermissiveRouterTLSUrl());
+    makeRequest(wrongRootClient, HttpMethod.GET, config.getNoClientAuthRouterTLSUrl() + "/status", null);
+  }
+
+  @Test
+  public void checkAccessWithRevokedCert()
+  {
+    LOG.info("---------Testing TLS resource access when client certificate has been revoked---------");
+    HttpClient revokedClient = makeCustomHttpClient(
+        "client_tls/revoked_client.jks",
+        "revoked_druid"
+    );
+    checkFailedAccessRevoked(revokedClient, HttpMethod.GET, config.getCoordinatorTLSUrl());
+    checkFailedAccessRevoked(revokedClient, HttpMethod.GET, config.getIndexerTLSUrl());
+    checkFailedAccessRevoked(revokedClient, HttpMethod.GET, config.getBrokerTLSUrl());
+    checkFailedAccessRevoked(revokedClient, HttpMethod.GET, config.getHistoricalTLSUrl());
+    checkFailedAccessRevoked(revokedClient, HttpMethod.GET, config.getRouterTLSUrl());
+    makeRequest(revokedClient, HttpMethod.GET, config.getPermissiveRouterTLSUrl() + "/status", null);
+    makeRequest(revokedClient, HttpMethod.GET, config.getNoClientAuthRouterTLSUrl() + "/status", null);
+  }
+
+  @Test
+  public void checkAccessWithExpiredCert()
+  {
+    LOG.info("---------Testing TLS resource access when client certificate has expired---------");
+    HttpClient expiredClient = makeCustomHttpClient(
+        "client_tls/expired_client.jks",
+        "expired_client"
+    );
+    checkFailedAccessExpired(expiredClient, HttpMethod.GET, config.getCoordinatorTLSUrl());
+    checkFailedAccessExpired(expiredClient, HttpMethod.GET, config.getIndexerTLSUrl());
+    checkFailedAccessExpired(expiredClient, HttpMethod.GET, config.getBrokerTLSUrl());
+    checkFailedAccessExpired(expiredClient, HttpMethod.GET, config.getHistoricalTLSUrl());
+    checkFailedAccessExpired(expiredClient, HttpMethod.GET, config.getRouterTLSUrl());
+    checkFailedAccessExpired(expiredClient, HttpMethod.GET, config.getPermissiveRouterTLSUrl());
+    makeRequest(expiredClient, HttpMethod.GET, config.getNoClientAuthRouterTLSUrl() + "/status", null);
+  }
+
+  @Test
+  public void checkAccessWithNotCASignedCert()
+  {
+    LOG.info(
+        "---------Testing TLS resource access when client certificate is signed by a non-CA intermediate cert---------");
+    HttpClient notCAClient = makeCustomHttpClient(
+        "client_tls/invalid_ca_client.jks",
+        "invalid_ca_client"
+    );
+    checkFailedAccessNotCA(notCAClient, HttpMethod.GET, config.getCoordinatorTLSUrl());
+    checkFailedAccessNotCA(notCAClient, HttpMethod.GET, config.getIndexerTLSUrl());
+    checkFailedAccessNotCA(notCAClient, HttpMethod.GET, config.getBrokerTLSUrl());
+    checkFailedAccessNotCA(notCAClient, HttpMethod.GET, config.getHistoricalTLSUrl());
+    checkFailedAccessNotCA(notCAClient, HttpMethod.GET, config.getRouterTLSUrl());
+    checkFailedAccessNotCA(notCAClient, HttpMethod.GET, config.getPermissiveRouterTLSUrl());
+    makeRequest(notCAClient, HttpMethod.GET, config.getNoClientAuthRouterTLSUrl() + "/status", null);
+  }
+
+  private void checkFailedAccessNoCert(HttpClient httpClient, HttpMethod method, String url)
+  {
+    checkFailedAccess(
+        httpClient,
+        method,
+        url + "/status",
+        "Certless",
+        SSLException.class,
+        "Received fatal alert: bad_certificate"
+    );
+  }
+
+  private void checkFailedAccessWrongHostname(HttpClient httpClient, HttpMethod method, String url)
+  {
+    checkFailedAccess(
+        httpClient,
+        method,
+        url + "/status",
+        "Wrong hostname",
+        SSLException.class,
+        "Received fatal alert: certificate_unknown"
+    );
+  }
+
+  private void checkFailedAccessWrongRoot(HttpClient httpClient, HttpMethod method, String url)
+  {
+    checkFailedAccess(
+        httpClient,
+        method,
+        url + "/status",
+        "Wrong root cert",
+        SSLException.class,
+        "Received fatal alert: certificate_unknown"
+    );
+  }
+
+  private void checkFailedAccessRevoked(HttpClient httpClient, HttpMethod method, String url)
+  {
+    checkFailedAccess(
+        httpClient,
+        method,
+        url + "/status",
+        "Revoked cert",
+        SSLException.class,
+        "Received fatal alert: certificate_unknown"
+    );
+  }
+
+  private void checkFailedAccessExpired(HttpClient httpClient, HttpMethod method, String url)
+  {
+    checkFailedAccess(
+        httpClient,
+        method,
+        url + "/status",
+        "Expired cert",
+        SSLException.class,
+        "Received fatal alert: certificate_unknown"
+    );
+  }
+
+  private void checkFailedAccessNotCA(HttpClient httpClient, HttpMethod method, String url)
+  {
+    checkFailedAccess(
+        httpClient,
+        method,
+        url + "/status",
+        "Cert signed by non-CA",
+        SSLException.class,
+        "Received fatal alert: certificate_unknown"
+    );
+  }
+
+  private HttpClientConfig.Builder getHttpClientConfigBuilder(SSLContext sslContext)
+  {
+    return HttpClientConfig
+        .builder()
+        .withNumConnections(httpClientConfig.getNumConnections())
+        .withReadTimeout(httpClientConfig.getReadTimeout())
+        .withWorkerCount(httpClientConfig.getNumMaxThreads())
+        .withCompressionCodec(
+            HttpClientConfig.CompressionCodec.valueOf(StringUtils.toUpperCase(httpClientConfig.getCompressionCodec()))
+        )
+        .withUnusedConnectionTimeoutDuration(httpClientConfig.getUnusedConnectionTimeout())
+        .withSslHandshakeTimeout(SSL_HANDSHAKE_TIMEOUT)
+        .withSslContext(sslContext);
+  }
+
+  private HttpClient makeCustomHttpClient(String keystorePath, String certAlias)
+  {
+    SSLContext intermediateClientSSLContext = new TLSUtils.ClientSSLContextBuilder()
+        .setProtocol(sslClientConfig.getProtocol())
+        .setTrustStoreType(sslClientConfig.getTrustStoreType())
+        .setTrustStorePath(sslClientConfig.getTrustStorePath())
+        .setTrustStoreAlgorithm(sslClientConfig.getTrustStoreAlgorithm())
+        .setTrustStorePasswordProvider(sslClientConfig.getTrustStorePasswordProvider())
+        .setKeyStoreType(sslClientConfig.getKeyStoreType())
+        .setKeyStorePath(keystorePath)
+        .setKeyStoreAlgorithm(sslClientConfig.getKeyManagerFactoryAlgorithm())
+        .setCertAlias(certAlias)
+        .setKeyStorePasswordProvider(sslClientConfig.getKeyStorePasswordProvider())
+        .setKeyManagerFactoryPasswordProvider(sslClientConfig.getKeyManagerPasswordProvider())
+        .build();
+
+    final HttpClientConfig.Builder builder = getHttpClientConfigBuilder(intermediateClientSSLContext);
+
+    final Lifecycle lifecycle = new Lifecycle();
+
+    HttpClient client = HttpClientInit.createClient(
+        builder.build(),
+        LifecycleUtils.asMmxLifecycle(lifecycle)
+    );
+
+    HttpClient adminClient = new CredentialedHttpClient(
+        new BasicCredentials("admin", "priest"),
+        client
+    );
+    return adminClient;
+  }
+
+  private HttpClient makeCertlessClient()
+  {
+    SSLContext certlessClientSSLContext = new TLSUtils.ClientSSLContextBuilder()
+        .setProtocol(sslClientConfig.getProtocol())
+        .setTrustStoreType(sslClientConfig.getTrustStoreType())
+        .setTrustStorePath(sslClientConfig.getTrustStorePath())
+        .setTrustStoreAlgorithm(sslClientConfig.getTrustStoreAlgorithm())
+        .setTrustStorePasswordProvider(sslClientConfig.getTrustStorePasswordProvider())
+        .build();
+
+    final HttpClientConfig.Builder builder = getHttpClientConfigBuilder(certlessClientSSLContext);
+
+    final Lifecycle lifecycle = new Lifecycle();
+
+    HttpClient client = HttpClientInit.createClient(
+        builder.build(),
+        LifecycleUtils.asMmxLifecycle(lifecycle)
+    );
+
+    HttpClient adminClient = new CredentialedHttpClient(
+        new BasicCredentials("admin", "priest"),
+        client
+    );
+    return adminClient;
+  }
+
+  private void checkFailedAccess(
+      HttpClient httpClient,
+      HttpMethod method,
+      String url,
+      String clientDesc,
+      Class expectedException,
+      String expectedExceptionMsg
+  )
+  {
+    int retries = 0;
+    while (true) {
+      try {
+        makeRequest(httpClient, method, url, null, -1);
+      }
+      catch (RuntimeException re) {
+        Throwable rootCause = Throwables.getRootCause(re);
+
+        if (rootCause instanceof IOException && "Broken pipe".equals(rootCause.getMessage())) {
+          if (retries > MAX_BROKEN_PIPE_RETRIES) {
+            Assert.fail(StringUtils.format(
+                "Broken pipe retries exhausted, test failed, did not get %s.",
+                expectedException
+            ));
+          } else {
+            retries += 1;
+            continue;
+          }
+        }
+
+        Assert.assertTrue(
+            expectedException.isInstance(rootCause),
+            StringUtils.format("Expected %s but found %s instead.", expectedException, rootCause)
+        );
+
+        Assert.assertEquals(
+            rootCause.getMessage(),
+            expectedExceptionMsg
+        );
+
+        LOG.info("%s client [%s] request failed as expected when accessing [%s]", clientDesc, method, url);
+        return;
+      }
+      Assert.fail(StringUtils.format("Test failed, did not get %s.", expectedException));
+    }
+  }
+
+  private StatusResponseHolder makeRequest(HttpClient httpClient, HttpMethod method, String url, byte[] content)
+  {
+    return makeRequest(httpClient, method, url, content, 4);
+  }
+
+  private StatusResponseHolder makeRequest(
+      HttpClient httpClient,
+      HttpMethod method,
+      String url,
+      byte[] content,
+      int maxRetries
+  )
+  {
+    try {
+      Request request = new Request(method, new URL(url));
+      if (content != null) {
+        request.setContent(MediaType.APPLICATION_JSON, content);
+      }
+      int retryCount = 0;
+
+      StatusResponseHolder response;
+
+      while (true) {
+        response = httpClient.go(
+            request,
+            responseHandler
+        ).get();
+
+        if (!response.getStatus().equals(HttpResponseStatus.OK)) {
+          String errMsg = StringUtils.format(
+              "Error while making request to url[%s] status[%s] content[%s]",
+              url,
+              response.getStatus(),
+              response.getContent()
+          );
+          if (retryCount > maxRetries) {
+            throw new ISE(errMsg);
+          } else {
+            LOG.error(errMsg);
+            LOG.error("retrying in 3000ms, retryCount: " + retryCount);
+            retryCount++;
+            Thread.sleep(3000);
+          }
+        } else {
+          LOG.info("[%s] request to [%s] succeeded.", method, url);
+          break;
+        }
+      }
+      return response;
+    }
+    catch (Exception e) {
+      throw Throwables.propagate(e);
+    }
+  }
+}
--- a/integration-tests/src/test/java/org/apache/druid/tests/hadoop/ITHadoopIndexTest.java
+++ b/integration-tests/src/test/java/org/apache/druid/tests/hadoop/ITHadoopIndexTest.java
@ -75,7 +75,7 @@ public class ITHadoopIndexTest extends AbstractIndexerTest
    try {
      final String taskID = indexer.submitTask(indexerSpec);
      LOG.info("TaskID for loading index task %s", taskID);
-      indexer.waitUntilTaskCompletes(taskID, 60000, 20);
+      indexer.waitUntilTaskCompletes(taskID, 10000, 120);
      RetryUtil.retryUntil(
          new Callable<Boolean>()
          {
--- a/integration-tests/src/test/java/org/apache/druid/tests/indexer/AbstractITRealtimeIndexTaskTest.java
+++ b/integration-tests/src/test/java/org/apache/druid/tests/indexer/AbstractITRealtimeIndexTaskTest.java
@ -142,8 +142,8 @@ public abstract class AbstractITRealtimeIndexTaskTest extends AbstractIndexerTes
            }
          },
          true,
-          60000,
-          10,
+          10000,
+          60,
          "Real-time generated segments loaded"
      );

--- a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITCompactionTaskTest.java
+++ b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITCompactionTaskTest.java
@ -42,6 +42,7 @@ public class ITCompactionTaskTest extends AbstractIndexerTest
  {
    loadData();
    final List<String> intervalsBeforeCompaction = coordinator.getSegmentIntervals(INDEX_DATASOURCE);
+    intervalsBeforeCompaction.sort(null);
    final String compactedInterval = "2013-08-31T00:00:00.000Z/2013-09-02T00:00:00.000Z";
    if (intervalsBeforeCompaction.contains(compactedInterval)) {
      throw new ISE("Containing a segment for the compacted interval[%s] before compaction", compactedInterval);
@ -49,12 +50,14 @@ public class ITCompactionTaskTest extends AbstractIndexerTest
    try {
      queryHelper.testQueriesFromFile(INDEX_QUERIES_RESOURCE, 2);
      compactData(false);
+
+      // 4 segments across 2 days, compacted into 1 new segment (5 total)
+      checkCompactionFinished(5);
      queryHelper.testQueriesFromFile(INDEX_QUERIES_RESOURCE, 2);

-      final List<String> intervalsAfterCompaction = coordinator.getSegmentIntervals(INDEX_DATASOURCE);
-      if (!intervalsAfterCompaction.contains(compactedInterval)) {
-        throw new ISE("Compacted segment for interval[%s] does not exist", compactedInterval);
-      }
+      intervalsBeforeCompaction.add(compactedInterval);
+      intervalsBeforeCompaction.sort(null);
+      checkCompactionIntervals(intervalsBeforeCompaction);
    }
    finally {
      unloadAndKillData(INDEX_DATASOURCE);
@ -66,21 +69,16 @@ public class ITCompactionTaskTest extends AbstractIndexerTest
  {
    loadData();
    final List<String> intervalsBeforeCompaction = coordinator.getSegmentIntervals(INDEX_DATASOURCE);
+    intervalsBeforeCompaction.sort(null);
    try {
      queryHelper.testQueriesFromFile(INDEX_QUERIES_RESOURCE, 2);
      compactData(true);
+
+      // 4 segments across 2 days, compacted into 2 new segments (6 total)
+      checkCompactionFinished(6);
      queryHelper.testQueriesFromFile(INDEX_QUERIES_RESOURCE, 2);

-      final List<String> intervalsAfterCompaction = coordinator.getSegmentIntervals(INDEX_DATASOURCE);
-      intervalsBeforeCompaction.sort(null);
-      intervalsAfterCompaction.sort(null);
-      if (!intervalsBeforeCompaction.equals(intervalsAfterCompaction)) {
-        throw new ISE(
-            "Intervals before compaction[%s] should be same with those after compaction[%s]",
-            intervalsBeforeCompaction,
-            intervalsAfterCompaction
-        );
-      }
+      checkCompactionIntervals(intervalsBeforeCompaction);
    }
    finally {
      unloadAndKillData(INDEX_DATASOURCE);
@ -112,4 +110,30 @@ public class ITCompactionTaskTest extends AbstractIndexerTest
        "Segment Compaction"
    );
  }
+
+  private void checkCompactionFinished(int numExpectedSegments)
+  {
+    RetryUtil.retryUntilTrue(
+        () -> {
+          int metadataSegmentCount = coordinator.getMetadataSegments(INDEX_DATASOURCE).size();
+          LOG.info("Current metadata segment count: %d, expected: %d", metadataSegmentCount, numExpectedSegments);
+          return metadataSegmentCount == numExpectedSegments;
+        },
+        "Compaction segment count check"
+    );
+  }
+
+  private void checkCompactionIntervals(List<String> expectedIntervals)
+  {
+    RetryUtil.retryUntilTrue(
+        () -> {
+          final List<String> intervalsAfterCompaction = coordinator.getSegmentIntervals(INDEX_DATASOURCE);
+          intervalsAfterCompaction.sort(null);
+          System.out.println("AFTER: " + intervalsAfterCompaction);
+          System.out.println("EXPECTED: " + expectedIntervals);
+          return intervalsAfterCompaction.equals(expectedIntervals);
+        },
+        "Compaction interval check"
+    );
+  }
 }
--- a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITKafkaIndexingServiceTest.java
+++ b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITKafkaIndexingServiceTest.java
@ -278,8 +278,8 @@ public class ITKafkaIndexingServiceTest extends AbstractIndexerTest
            }
          },
          true,
-          30000,
-          10,
+          10000,
+          30,
          "Real-time generated segments loaded"
      );
    }
--- a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITKafkaTest.java
+++ b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITKafkaTest.java
@ -226,7 +226,7 @@ public class ITKafkaTest extends AbstractIndexerTest
    LOG.info("-------------SUBMITTED TASK");

    // wait for the task to finish
-    indexer.waitUntilTaskCompletes(taskID, 20000, 30);
+    indexer.waitUntilTaskCompletes(taskID, 10000, 60);

    // wait for segments to be handed off
    try {
@ -240,8 +240,8 @@ public class ITKafkaTest extends AbstractIndexerTest
            }
          },
          true,
-          30000,
-          10,
+          10000,
+          30,
          "Real-time generated segments loaded"
      );
    }
--- a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITUnionQueryTest.java
+++ b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITUnionQueryTest.java
@ -25,18 +25,26 @@ import com.google.inject.Inject;
 import org.apache.druid.curator.discovery.ServerDiscoveryFactory;
 import org.apache.druid.curator.discovery.ServerDiscoverySelector;
 import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.Request;
+import org.apache.druid.java.util.http.client.response.StatusResponseHandler;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
 import org.apache.druid.testing.IntegrationTestingConfig;
 import org.apache.druid.testing.clients.EventReceiverFirehoseTestClient;
 import org.apache.druid.testing.guice.DruidTestModuleFactory;
 import org.apache.druid.testing.guice.TestClient;
 import org.apache.druid.testing.utils.RetryUtil;
 import org.apache.druid.testing.utils.ServerDiscoveryUtil;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
 import org.joda.time.DateTime;
 import org.testng.annotations.Guice;
 import org.testng.annotations.Test;

+import java.net.URL;
+import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
@ -173,6 +181,26 @@ public class ITUnionQueryTest extends AbstractIndexerTest

      LOG.info("Event Receiver Found at host [%s]", host);

+      LOG.info("Checking worker /status/health for [%s]", host);
+      final StatusResponseHandler handler = new StatusResponseHandler(StandardCharsets.UTF_8);
+      RetryUtil.retryUntilTrue(
+          () -> {
+            try {
+              StatusResponseHolder response = httpClient.go(
+                  new Request(HttpMethod.GET, new URL(StringUtils.format("https://%s/status/health", host))),
+                  handler
+              ).get();
+              return response.getStatus().equals(HttpResponseStatus.OK);
+            }
+            catch (Throwable e) {
+              LOG.error(e, "");
+              return false;
+            }
+          },
+          StringUtils.format("Checking /status/health for worker [%s]", host)
+      );
+      LOG.info("Finished checking worker /status/health for [%s], success", host);
+
      EventReceiverFirehoseTestClient client = new EventReceiverFirehoseTestClient(
          host,
          EVENT_RECEIVER_SERVICE_PREFIX + id,
--- a/integration-tests/src/test/resources/indexer/kafka_index_task.json
+++ b/integration-tests/src/test/resources/indexer/kafka_index_task.json
@ -61,7 +61,7 @@
      "type" : "realtime",
      "maxRowsInMemory": 500000,
      "intermediatePersistPeriod": "PT3M",
-      "windowPeriod": "PT1M",
+      "windowPeriod": "PT150S",
      "basePersistDirectory": "/home/y/var/druid_state/kafka_test/realtime/basePersist"
    }
  }
--- a/integration-tests/stop_cluster.sh
+++ b/integration-tests/stop_cluster.sh
@ -14,8 +14,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.

-for node in druid-historical druid-coordinator druid-overlord druid-router druid-broker druid-middlemanager druid-zookeeper-kafka druid-metadata-storage;
-do 
+for node in druid-historical druid-coordinator druid-overlord druid-router druid-router-permissive-tls druid-router-no-client-auth-tls druid-broker druid-middlemanager druid-zookeeper-kafka druid-metadata-storage;
+
+do
 docker stop $node
 docker rm $node
 done
+
+docker network rm druid-it-net
--- a/pom.xml
+++ b/pom.xml
@ -1106,6 +1106,7 @@
                        <exclude>**/tutorial/conf/**</exclude>
                        <exclude>**/derby.log</exclude>
                        <exclude>**/docker/**</exclude>
+                        <exclude>**/client_tls/**</exclude>
                        <!--IDE MODULE FILES-->
                        <exclude>**/*.iml</exclude>
                        <!--CRASH LOGS-->
--- a/server/src/main/java/org/apache/druid/server/StatusResource.java
+++ b/server/src/main/java/org/apache/druid/server/StatusResource.java
@ -31,9 +31,11 @@ import org.apache.druid.server.http.security.ConfigResourceFilter;
 import org.apache.druid.server.http.security.StateResourceFilter;

 import javax.inject.Inject;
+import javax.servlet.http.HttpServletRequest;
 import javax.ws.rs.GET;
 import javax.ws.rs.Path;
 import javax.ws.rs.Produces;
+import javax.ws.rs.core.Context;
 import javax.ws.rs.core.MediaType;
 import java.util.ArrayList;
 import java.util.Collection;
@ -47,7 +49,6 @@ import java.util.Set;
@Path("/status")
 public class StatusResource
 {
-
  private final Properties properties;

  private final DruidServerConfig druidServerConfig;
@ -73,7 +74,9 @@ public class StatusResource
  @GET
  @ResourceFilters(StateResourceFilter.class)
  @Produces(MediaType.APPLICATION_JSON)
-  public Status doGet()
+  public Status doGet(
+      @Context final HttpServletRequest req
+  )
  {
    return new Status(Initialization.getLoadedImplementations(DruidModule.class));
  }
--- a/server/src/main/java/org/apache/druid/server/emitter/HttpEmitterModule.java
+++ b/server/src/main/java/org/apache/druid/server/emitter/HttpEmitterModule.java
@ -126,13 +126,13 @@ public class HttpEmitterModule implements Module
      }
    } else if (sslConfig.getTrustStorePath() != null) {
      log.info("Creating SSLContext for HttpEmitter client using config [%s]", sslConfig);
-      effectiveSSLContext = TLSUtils.createSSLContext(
-          sslConfig.getProtocol(),
-          sslConfig.getTrustStoreType(),
-          sslConfig.getTrustStorePath(),
-          sslConfig.getTrustStoreAlgorithm(),
-          sslConfig.getTrustStorePasswordProvider()
-      );
+      effectiveSSLContext = new TLSUtils.ClientSSLContextBuilder()
+          .setProtocol(sslConfig.getProtocol())
+          .setTrustStoreType(sslConfig.getTrustStoreType())
+          .setTrustStorePath(sslConfig.getTrustStorePath())
+          .setTrustStoreAlgorithm(sslConfig.getTrustStoreAlgorithm())
+          .setTrustStorePasswordProvider(sslConfig.getTrustStorePasswordProvider())
+          .build();
    } else {
      effectiveSSLContext = sslContext;
    }
--- a/server/src/main/java/org/apache/druid/server/http/OverlordProxyServlet.java
+++ b/server/src/main/java/org/apache/druid/server/http/OverlordProxyServlet.java
@ -23,12 +23,17 @@ import com.google.common.base.Throwables;
 import com.google.inject.Inject;
 import org.apache.druid.client.indexing.IndexingService;
 import org.apache.druid.discovery.DruidLeaderClient;
+import org.apache.druid.guice.annotations.Global;
+import org.apache.druid.guice.http.DruidHttpClientConfig;
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.server.security.AuthConfig;
+import com.google.inject.Provider;
+import org.eclipse.jetty.client.HttpClient;
 import org.eclipse.jetty.client.api.Request;
 import org.eclipse.jetty.proxy.ProxyServlet;

+import javax.servlet.ServletException;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 import java.net.URI;
@ -40,13 +45,19 @@ import java.net.URISyntaxException;
 public class OverlordProxyServlet extends ProxyServlet
 {
  private final DruidLeaderClient druidLeaderClient;
+  private final Provider<HttpClient> httpClientProvider;
+  private final DruidHttpClientConfig httpClientConfig;

  @Inject
  OverlordProxyServlet(
-      @IndexingService DruidLeaderClient druidLeaderClient
+      @IndexingService DruidLeaderClient druidLeaderClient,
+      @Global Provider<HttpClient> httpClientProvider,
+      @Global DruidHttpClientConfig httpClientConfig
  )
  {
    this.druidLeaderClient = druidLeaderClient;
+    this.httpClientProvider = httpClientProvider;
+    this.httpClientConfig = httpClientConfig;
  }

  @Override
@ -71,6 +82,22 @@ public class OverlordProxyServlet extends ProxyServlet
    }
  }

+  @Override
+  protected HttpClient newHttpClient()
+  {
+    return httpClientProvider.get();
+  }
+
+  @Override
+  protected HttpClient createHttpClient() throws ServletException
+  {
+    HttpClient client = super.createHttpClient();
+    // override timeout set in ProxyServlet.createHttpClient
+    setTimeout(httpClientConfig.getReadTimeout().getMillis());
+    return client;
+  }
+
+
  @Override
  protected void sendProxyRequest(
      HttpServletRequest clientRequest,
--- a/server/src/main/java/org/apache/druid/server/initialization/TLSServerConfig.java
+++ b/server/src/main/java/org/apache/druid/server/initialization/TLSServerConfig.java
@ -55,6 +55,27 @@ public class TLSServerConfig
  @JsonProperty
  private List<String> excludeProtocols;

+  @JsonProperty
+  private boolean requireClientCertificate = false;
+
+  @JsonProperty
+  private String trustStoreType;
+
+  @JsonProperty
+  private String trustStorePath;
+
+  @JsonProperty
+  private String trustStoreAlgorithm;
+
+  @JsonProperty("trustStorePassword")
+  private PasswordProvider trustStorePasswordProvider;
+
+  @JsonProperty
+  private boolean validateHostnames = true;
+
+  @JsonProperty
+  private String crlPath;
+
  public String getKeyStorePath()
  {
    return keyStorePath;
@ -105,6 +126,41 @@ public class TLSServerConfig
    return excludeProtocols;
  }

+  public boolean isRequireClientCertificate()
+  {
+    return requireClientCertificate;
+  }
+
+  public String getTrustStoreType()
+  {
+    return trustStoreType;
+  }
+
+  public String getTrustStorePath()
+  {
+    return trustStorePath;
+  }
+
+  public String getTrustStoreAlgorithm()
+  {
+    return trustStoreAlgorithm;
+  }
+
+  public PasswordProvider getTrustStorePasswordProvider()
+  {
+    return trustStorePasswordProvider;
+  }
+
+  public boolean isValidateHostnames()
+  {
+    return validateHostnames;
+  }
+
+  public String getCrlPath()
+  {
+    return crlPath;
+  }
+
  @Override
  public String toString()
  {
@ -117,6 +173,12 @@ public class TLSServerConfig
           ", excludeCipherSuites=" + excludeCipherSuites +
           ", includeProtocols=" + includeProtocols +
           ", excludeProtocols=" + excludeProtocols +
+           ", requireClientCertificate=" + requireClientCertificate +
+           ", trustStoreType='" + trustStoreType + '\'' +
+           ", trustStorePath='" + trustStorePath + '\'' +
+           ", trustStoreAlgorithm='" + trustStoreAlgorithm + '\'' +
+           ", validateHostnames='" + validateHostnames + '\'' +
+           ", crlPath='" + crlPath + '\'' +
           '}';
  }
 }
--- a/server/src/main/java/org/apache/druid/server/initialization/jetty/JettyServerModule.java
+++ b/server/src/main/java/org/apache/druid/server/initialization/jetty/JettyServerModule.java
@ -75,6 +75,8 @@ import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler;

 import javax.net.ssl.KeyManagerFactory;
 import javax.net.ssl.SSLEngine;
+import javax.net.ssl.TrustManagerFactory;
+import java.security.KeyStore;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@ -231,6 +233,7 @@ public class JettyServerModule extends JerseyServletModule
      if (sslContextFactoryBinding == null) {
        // Never trust all certificates by default
        sslContextFactory = new SslContextFactory(false);
+
        sslContextFactory.setKeyStorePath(tlsServerConfig.getKeyStorePath());
        sslContextFactory.setKeyStoreType(tlsServerConfig.getKeyStoreType());
        sslContextFactory.setKeyStorePassword(tlsServerConfig.getKeyStorePasswordProvider().getPassword());
@ -256,6 +259,37 @@ public class JettyServerModule extends JerseyServletModule
          sslContextFactory.setExcludeProtocols(
              tlsServerConfig.getExcludeProtocols().toArray(new String[0]));
        }
+
+        sslContextFactory.setNeedClientAuth(tlsServerConfig.isRequireClientCertificate());
+        if (tlsServerConfig.isRequireClientCertificate()) {
+          if (tlsServerConfig.getCrlPath() != null) {
+            // setValidatePeerCerts is used just to enable revocation checking using a static CRL file.
+            // Certificate validation is always performed when client certificates are required.
+            sslContextFactory.setValidatePeerCerts(true);
+            sslContextFactory.setCrlPath(tlsServerConfig.getCrlPath());
+          }
+          if (tlsServerConfig.isValidateHostnames()) {
+            sslContextFactory.setEndpointIdentificationAlgorithm("HTTPS");
+          }
+          if (tlsServerConfig.getTrustStorePath() != null) {
+            sslContextFactory.setTrustStorePath(tlsServerConfig.getTrustStorePath());
+            sslContextFactory.setTrustStoreType(
+                tlsServerConfig.getTrustStoreType() == null
+                ? KeyStore.getDefaultType()
+                : tlsServerConfig.getTrustStoreType()
+            );
+            sslContextFactory.setTrustManagerFactoryAlgorithm(
+                tlsServerConfig.getTrustStoreAlgorithm() == null
+                ? TrustManagerFactory.getDefaultAlgorithm()
+                : tlsServerConfig.getTrustStoreAlgorithm()
+            );
+            sslContextFactory.setTrustStorePassword(
+                tlsServerConfig.getTrustStorePasswordProvider() == null
+                ? null
+                : tlsServerConfig.getTrustStorePasswordProvider().getPassword()
+            );
+          }
+        }
      } else {
        sslContextFactory = sslContextFactoryBinding.getProvider().get();
      }
--- a/server/src/main/java/org/apache/druid/server/security/TLSUtils.java
+++ b/server/src/main/java/org/apache/druid/server/security/TLSUtils.java
@ -19,48 +19,205 @@

 package org.apache.druid.server.security;

+import com.google.common.base.Preconditions;
 import com.google.common.base.Throwables;
 import org.apache.druid.metadata.PasswordProvider;

+import org.eclipse.jetty.util.ssl.AliasedX509ExtendedKeyManager;
+
+import javax.annotation.Nullable;
+import javax.net.ssl.KeyManager;
+import javax.net.ssl.KeyManagerFactory;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509ExtendedKeyManager;
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.security.KeyManagementException;
 import java.security.KeyStore;
 import java.security.KeyStoreException;
 import java.security.NoSuchAlgorithmException;
+import java.security.UnrecoverableKeyException;
 import java.security.cert.CertificateException;

 public class TLSUtils
 {
+  public static class ClientSSLContextBuilder
+  {
+    private String protocol;
+    private String trustStoreType;
+    private String trustStorePath;
+    private String trustStoreAlgorithm;
+    private PasswordProvider trustStorePasswordProvider;
+    private String keyStoreType;
+    private String keyStorePath;
+    private String keyStoreAlgorithm;
+    private String certAlias;
+    private PasswordProvider keyStorePasswordProvider;
+    private PasswordProvider keyManagerFactoryPasswordProvider;
+
+    public ClientSSLContextBuilder setProtocol(String protocol)
+    {
+      this.protocol = protocol;
+      return this;
+    }
+
+    public ClientSSLContextBuilder setTrustStoreType(String trustStoreType)
+    {
+      this.trustStoreType = trustStoreType;
+      return this;
+    }
+
+    public ClientSSLContextBuilder setTrustStorePath(String trustStorePath)
+    {
+      this.trustStorePath = trustStorePath;
+      return this;
+    }
+
+    public ClientSSLContextBuilder setTrustStoreAlgorithm(String trustStoreAlgorithm)
+    {
+      this.trustStoreAlgorithm = trustStoreAlgorithm;
+      return this;
+    }
+
+    public ClientSSLContextBuilder setTrustStorePasswordProvider(PasswordProvider trustStorePasswordProvider)
+    {
+      this.trustStorePasswordProvider = trustStorePasswordProvider;
+      return this;
+    }
+
+    public ClientSSLContextBuilder setKeyStoreType(String keyStoreType)
+    {
+      this.keyStoreType = keyStoreType;
+      return this;
+    }
+
+    public ClientSSLContextBuilder setKeyStorePath(String keyStorePath)
+    {
+      this.keyStorePath = keyStorePath;
+      return this;
+    }
+
+    public ClientSSLContextBuilder setKeyStoreAlgorithm(String keyStoreAlgorithm)
+    {
+      this.keyStoreAlgorithm = keyStoreAlgorithm;
+      return this;
+    }
+
+    public ClientSSLContextBuilder setCertAlias(String certAlias)
+    {
+      this.certAlias = certAlias;
+      return this;
+    }
+
+    public ClientSSLContextBuilder setKeyStorePasswordProvider(PasswordProvider keyStorePasswordProvider)
+    {
+      this.keyStorePasswordProvider = keyStorePasswordProvider;
+      return this;
+    }
+
+    public ClientSSLContextBuilder setKeyManagerFactoryPasswordProvider(PasswordProvider keyManagerFactoryPasswordProvider)
+    {
+      this.keyManagerFactoryPasswordProvider = keyManagerFactoryPasswordProvider;
+      return this;
+    }
+
+    public SSLContext build()
+    {
+      Preconditions.checkNotNull(trustStorePath, "must specify a trustStorePath");
+
+      return createSSLContext(
+        protocol,
+        trustStoreType,
+        trustStorePath,
+        trustStoreAlgorithm,
+        trustStorePasswordProvider,
+        keyStoreType,
+        keyStorePath,
+        keyStoreAlgorithm,
+        certAlias,
+        keyStorePasswordProvider,
+        keyManagerFactoryPasswordProvider
+      );
+    }
+  }
+
  public static SSLContext createSSLContext(
-      String protocol,
-      String trustStoreType,
+      @Nullable String protocol,
+      @Nullable String trustStoreType,
      String trustStorePath,
-      String trustStoreAlgorithm,
-      PasswordProvider trustStorePasswordProvider
+      @Nullable String trustStoreAlgorithm,
+      @Nullable PasswordProvider trustStorePasswordProvider,
+      @Nullable String keyStoreType,
+      @Nullable String keyStorePath,
+      @Nullable String keyStoreAlgorithm,
+      @Nullable String certAlias,
+      @Nullable PasswordProvider keyStorePasswordProvider,
+      @Nullable PasswordProvider keyManagerFactoryPasswordProvider
  )
  {
    SSLContext sslContext = null;
    try {
      sslContext = SSLContext.getInstance(protocol == null ? "TLSv1.2" : protocol);
-      KeyStore keyStore = KeyStore.getInstance(trustStoreType == null
+      KeyStore trustStore = KeyStore.getInstance(trustStoreType == null
                                               ? KeyStore.getDefaultType()
                                               : trustStoreType);
-      keyStore.load(
+      trustStore.load(
          new FileInputStream(trustStorePath),
-          trustStorePasswordProvider.getPassword().toCharArray()
+          trustStorePasswordProvider == null ? null : trustStorePasswordProvider.getPassword().toCharArray()
      );
      TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance(trustStoreAlgorithm == null
                                                                                ? TrustManagerFactory.getDefaultAlgorithm()
                                                                                : trustStoreAlgorithm);
-      trustManagerFactory.init(keyStore);
-      sslContext.init(null, trustManagerFactory.getTrustManagers(), null);
+      trustManagerFactory.init(trustStore);
+
+
+      KeyManager[] keyManagers;
+      if (keyStorePath != null) {
+        KeyStore keyStore = KeyStore.getInstance(keyStoreType == null
+                                                 ? KeyStore.getDefaultType()
+                                                 : keyStoreType);
+        keyStore.load(
+            new FileInputStream(keyStorePath),
+            keyStorePasswordProvider == null ? null : keyStorePasswordProvider.getPassword().toCharArray()
+        );
+
+        KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(
+            keyStoreAlgorithm == null ?
+            KeyManagerFactory.getDefaultAlgorithm() : keyStoreAlgorithm
+        );
+        keyManagerFactory.init(
+            keyStore,
+            keyManagerFactoryPasswordProvider == null ? null : keyManagerFactoryPasswordProvider.getPassword().toCharArray()
+        );
+        keyManagers = createAliasedKeyManagers(keyManagerFactory.getKeyManagers(), certAlias);
+      } else {
+        keyManagers = null;
+      }
+
+      sslContext.init(
+          keyManagers,
+          trustManagerFactory.getTrustManagers(),
+          null
+      );
    }
-    catch (CertificateException | KeyManagementException | IOException | KeyStoreException | NoSuchAlgorithmException e) {
+    catch (CertificateException | KeyManagementException | IOException | KeyStoreException | NoSuchAlgorithmException | UnrecoverableKeyException e) {
      Throwables.propagate(e);
    }
    return sslContext;
  }
+
+  // Use Jetty's aliased KeyManager for consistency between server/client TLS configs
+  private static KeyManager[] createAliasedKeyManagers(KeyManager[] delegates, String certAlias)
+  {
+    KeyManager[] aliasedManagers = new KeyManager[delegates.length];
+    for (int i = 0; i < delegates.length; i++) {
+      if (delegates[i] instanceof X509ExtendedKeyManager) {
+        aliasedManagers[i] = new AliasedX509ExtendedKeyManager((X509ExtendedKeyManager) delegates[i], certAlias);
+      } else {
+        aliasedManagers[i] = delegates[i];
+      }
+    }
+    return aliasedManagers;
+  }
 }
--- a/server/src/test/java/org/apache/druid/server/http/OverlordProxyServletTest.java
+++ b/server/src/test/java/org/apache/druid/server/http/OverlordProxyServletTest.java
@ -41,7 +41,7 @@ public class OverlordProxyServletTest

    EasyMock.replay(druidLeaderClient, request);

-    URI uri = URI.create(new OverlordProxyServlet(druidLeaderClient).rewriteTarget(request));
+    URI uri = URI.create(new OverlordProxyServlet(druidLeaderClient, null, null).rewriteTarget(request));
    Assert.assertEquals("https://overlord:port/druid/overlord/worker?param1=test&param2=test2", uri.toString());
  }

--- a/services/src/main/java/org/apache/druid/cli/CliCoordinator.java
+++ b/services/src/main/java/org/apache/druid/cli/CliCoordinator.java
@ -45,6 +45,7 @@ import org.apache.druid.guice.LifecycleModule;
 import org.apache.druid.guice.ManageLifecycle;
 import org.apache.druid.guice.annotations.CoordinatorIndexingServiceHelper;
 import org.apache.druid.guice.annotations.EscalatedGlobal;
+import org.apache.druid.guice.http.JettyHttpClientModule;
 import org.apache.druid.java.util.common.concurrent.ScheduledExecutorFactory;
 import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.java.util.http.client.HttpClient;
@ -127,6 +128,8 @@ public class CliCoordinator extends ServerRunnable
  {
    List<Module> modules = new ArrayList<>();

+    modules.add(JettyHttpClientModule.global());
+
    modules.add(
        new Module()
        {