Properly support TLS host verification with multiple static endpoints (#33)

* Properly support TLS host verification with multiple static endpoints

Also allow explicit override of authority and deprecate
ComposeTrustManagerFactory

Resolves #32

* Update doc to reflect overrideAuthority addition

Author: njhill

etcd-json-schema.md

@@ -4,30 +4,30 @@ Example JSON config doc:
 
 ```json
 {
-  "compose_deployment": "etcd-development-01",
   "endpoints": "https://dev3-9.compose.direct:15182,https://dev3-11.compose.direct:15182",
   "userid": "userid",
   "password": "password",
   "root_prefix": "aka-chroot-or-namespace",
-  "certificate_file": "etcd-dev.pem"
+  "certificate_file": "etcd-dev.pem",
+  "override_authority": "etcd-development-01"
 }
 ```
 
 - All attributes apart from `endpoints` are optional.
 - The `root_prefix` attribute currently has **no effect** on clients created via `EtcdClientConfig.getClient()`. It's included in the configuration for use by application code (to query via `EtcdClientConfig.getRootPrefix()`). In future full chroot-like functionality at the client level might be supported.
 - `certificate_file` is the name of a pem-format (public) cert to use for TLS server-auth, either an absolute path or a filename assumed to be in the same directory as the json config file itself.
 - A `certificate` attribute may be included _instead of_ `certificate_file`, whose value is an embedded string UTF-8 pem format certificate. This allows a single json doc to hold all of the necessary connection info.
-- The `compose_deployment` attribute is only required when using an IBM Compose etcd deployment with provided TLS certificates. It must be set to the name of the deployment, which is the CN of the TLS cert.
+- The `override_authority` is optional and may be used to override the authority used for TLS hostname verification for _all_ endpoints.
 
 Example with embedded (trunctated) TLS cert:
 
 ```json
 {
-  "compose_deployment": "etcd-development-01",
   "endpoints": "https://dev3-9.compose.direct:15182,https://dev3-11.compose.direct:15182",
   "userid": "userid",
   "password": "password",
   "root_prefix": "aka-chroot-or-namespace",
-  "certificate": "-----BEGIN CERTIFICATE-----\nMIIDaTCCA ... MP0u6J/xasx14IW4A==\n-----END CERTIFICATE-----\n"
+  "certificate": "-----BEGIN CERTIFICATE-----\nMIIDaTCCA ... MP0u6J/xasx14IW4A==\n-----END CERTIFICATE-----\n",
+  "override_authority": "etcd-development-01"
 }
 ```

src/main/java/com/ibm/etcd/client/EtcdClient.java

@@ -21,6 +21,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.BrokenBarrierException;
@@ -120,8 +121,12 @@ public class EtcdClient implements KvStoreClient {
     private volatile PersistentLease sessionLease; // lazy-instantiated
 
     public static class Builder {
-        private final NettyChannelBuilder chanBuilder;
+        private final List<String> endpoints;
+        private boolean plainText;
+        private int maxInboundMessageSize;
         private SslContextBuilder sslContextBuilder;
+        private SslContext sslContext;
+        private String overrideAuthority;
         private ByteString name, password;
         private long defaultTimeoutMs = DEFAULT_TIMEOUT_MS;
         private boolean preemptAuth;
@@ -130,8 +135,9 @@ public static class Builder {
         private boolean sendViaEventLoop = true; // default true
         private int sessTimeoutSecs = DEFAULT_SESSION_TIMEOUT_SECS;
 
-        Builder(NettyChannelBuilder chanBuilder) {
-            this.chanBuilder = chanBuilder;
+        Builder(List<String> endpoints) {
+            this.endpoints = Preconditions.checkNotNull(endpoints);
+            Preconditions.checkArgument(!endpoints.isEmpty(), "empty endpoints");
         }
 
         /**
@@ -224,7 +230,16 @@ private SslContextBuilder sslBuilder() {
          * Disable TLS - to connect to insecure servers in development contexts
          */
         public Builder withPlainText() {
-            chanBuilder.usePlaintext();
+            plainText = true;
+            return this;
+        }
+
+        /**
+         * Override the authority used for TLS hostname verification. Applies
+         * to all endpoints and does not otherwise affect DNS name resolution.
+         */
+        public Builder overrideAuthority(String authority) {
+            this.overrideAuthority = authority;
             return this;
         }
 
@@ -237,7 +252,7 @@ public Builder withPlainText() {
          */
         public Builder withCaCert(ByteSource certSource) throws IOException, SSLException {
             try (InputStream cert = certSource.openStream()) {
-                chanBuilder.sslContext(sslBuilder().trustManager(cert).build());
+                sslContext = sslBuilder().trustManager(cert).build();
             }
             return this;
         }
@@ -250,7 +265,7 @@ public Builder withCaCert(ByteSource certSource) throws IOException, SSLExceptio
          * @throws SSLException
          */
         public Builder withTrustManager(TrustManagerFactory tmf) throws SSLException {
-            chanBuilder.sslContext(sslBuilder().trustManager(tmf).build());
+            sslContext = sslBuilder().trustManager(tmf).build();
             return this;
         }
 
@@ -265,7 +280,7 @@ public Builder withTrustManager(TrustManagerFactory tmf) throws SSLException {
         public Builder withTlsConfig(Consumer<SslContextBuilder> contextBuilder) throws SSLException {
             SslContextBuilder sslBuilder = sslBuilder();
             contextBuilder.accept(sslBuilder);
-            chanBuilder.sslContext(sslBuilder.build());
+            sslContext = sslBuilder.build();
             return this;
         }
 
@@ -287,15 +302,36 @@ public Builder withSessionTimeoutSecs(int timeoutSecs) {
          * @param sizeInBytes
          */
         public Builder withMaxInboundMessageSize(int sizeInBytes) {
-            chanBuilder.maxInboundMessageSize(sizeInBytes);
+            this.maxInboundMessageSize = sizeInBytes; 
             return this;
         }
 
         /**
          * @return the built {@link EtcdClient} instance
          */
         public EtcdClient build() {
-            return new EtcdClient(chanBuilder, defaultTimeoutMs, name, password,
+            NettyChannelBuilder ncb;
+            if (endpoints.size() == 1) {
+                ncb = NettyChannelBuilder.forTarget(endpoints.get(0));
+                if (overrideAuthority != null) {
+                    ncb.overrideAuthority(overrideAuthority);
+                }
+            } else {
+                ncb = NettyChannelBuilder
+                        .forTarget(StaticEtcdNameResolverFactory.ETCD)
+                        .nameResolverFactory(new StaticEtcdNameResolverFactory(
+                                endpoints, overrideAuthority));
+            }
+            if (plainText) {
+                ncb.usePlaintext();
+            }
+            if (sslContext != null) {
+                ncb.sslContext(sslContext);
+            }
+            if (maxInboundMessageSize != 0) {
+                ncb.maxInboundMessageSize(maxInboundMessageSize);
+            }
+            return new EtcdClient(ncb, defaultTimeoutMs, name, password,
                     preemptAuth, threads, executor, sendViaEventLoop, sessTimeoutSecs);
         }
     }
@@ -312,7 +348,7 @@ private static int defaultThreadCount() {
      */
     public static Builder forEndpoint(String host, int port) {
         String target = GrpcUtil.authorityFromHostAndPort(host, port);
-        return new Builder(NettyChannelBuilder.forTarget(target));
+        return new Builder(Collections.singletonList(target));
     }
 
     /**
@@ -321,10 +357,7 @@ public static Builder forEndpoint(String host, int port) {
      * @return
      */
     public static Builder forEndpoints(List<String> endpoints) {
-        NettyChannelBuilder ncb = NettyChannelBuilder
-                .forTarget(StaticEtcdNameResolverFactory.ETCD)
-                .nameResolverFactory(new StaticEtcdNameResolverFactory(endpoints));
-        return new Builder(ncb);
+        return new Builder(endpoints);
     }
 
     /**

src/main/java/com/ibm/etcd/client/StaticEtcdNameResolverFactory.java

@@ -15,13 +15,15 @@
  */
 package com.ibm.etcd.client;
 
+import static io.grpc.EquivalentAddressGroup.ATTR_AUTHORITY_OVERRIDE;
+import static java.util.stream.Collectors.toList;
+
 import java.net.URI;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
-import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 import io.grpc.Attributes;
@@ -51,14 +53,32 @@ static class SubResolver {
         public SubResolver(URI uri, NameResolver.Helper helper) {
             this.resolver = DNS_PROVIDER.newNameResolver(uri, helper);
         }
+
+        void updateEagList(List<EquivalentAddressGroup> servers, boolean ownAuthority) {
+            if (ownAuthority) {
+                // Use this endpoint address' authority for its subchannel
+                String authority = resolver.getServiceAuthority();
+                eagList = servers.stream().map(eag -> new EquivalentAddressGroup(
+                        eag.getAddresses(), eag.getAttributes().toBuilder()
+                            .set(ATTR_AUTHORITY_OVERRIDE, authority).build())).collect(toList());
+            } else {
+                eagList = servers;
+            }
+        }
     }
 
     private final URI[] uris;
+    private final String overrideAuthority;
 
     public StaticEtcdNameResolverFactory(List<String> endpoints) {
+        this(endpoints, null);
+    }
+
+    public StaticEtcdNameResolverFactory(List<String> endpoints, String overrideAuthority) {
         if (endpoints == null || endpoints.isEmpty()) {
             throw new IllegalArgumentException("endpoints");
         }
+        this.overrideAuthority = overrideAuthority;
         int count = endpoints.size();
         uris = new URI[count];
         for (int i = 0; i < count; i++) {
@@ -95,9 +115,11 @@ public void start(Listener listener) {
                     @Override
                     public void onAddresses(List<EquivalentAddressGroup> servers, Attributes attributes) {
                         synchronized (resolvers) {
-                            sr.eagList = servers;
+                            // Update this subresolver's servers
+                            sr.updateEagList(servers, overrideAuthority == null);
+                            // Advertise the complete list of EAGs
                             List<EquivalentAddressGroup> newList = Stream.of(resolvers)
-                                    .flatMap(r -> r.eagList.stream()).collect(Collectors.toList());
+                                    .flatMap(r -> r.eagList.stream()).collect(toList());
                             currentCount = newList.size();
                             listener.onAddresses(newList, Attributes.EMPTY);
                         }
@@ -122,7 +144,7 @@ public void refresh() {
             }
             @Override
             public String getServiceAuthority() {
-                return ETCD;
+                return overrideAuthority != null ? overrideAuthority : ETCD;
             }
             @Override
             public void shutdown() {

src/main/java/com/ibm/etcd/client/config/ComposeTrustManagerFactory.java

@@ -37,10 +37,9 @@
 import io.netty.util.internal.EmptyArrays;
 
 /**
- * Custom trust manager to use with multi-endpoint IBM Compose etcd deployments.
- * Works around grpc-java TLS issues.
- * 
+ * @deprecated This is no longer required and shouldn't be used
  */
+@Deprecated
 public class ComposeTrustManagerFactory extends SimpleTrustManagerFactory {
 
     private static final Logger logger = LoggerFactory.getLogger(ComposeTrustManagerFactory.class);

src/main/java/com/ibm/etcd/client/config/EtcdClusterConfig.java

@@ -69,8 +69,10 @@ public enum TlsMode { TLS, PLAINTEXT, AUTO }
     TlsMode tlsMode;
     ByteString user, password;
     ByteString rootPrefix; // a.k.a namespace
+    @Deprecated
     String composeDeployment;
     ByteSource certificate;
+    String overrideAuthority;
 
     protected EtcdClusterConfig() {}
 
@@ -91,6 +93,9 @@ private EtcdClient newClient() throws IOException, CertificateException {
         EtcdClient.Builder builder = EtcdClient.forEndpoints(endpointList)
                 .withCredentials(user, password).withImmediateAuth()
                 .withMaxInboundMessageSize(maxMessageSize);
+        if (overrideAuthority != null) {
+            builder.overrideAuthority(overrideAuthority);
+        }
         TlsMode ssl = tlsMode;
         if (ssl == TlsMode.AUTO || ssl == null) {
             String ep = endpointList.get(0);
@@ -147,6 +152,7 @@ public static EtcdClusterConfig fromProperties(ByteSource source) throws IOExcep
             }
             config.certificate = Files.asByteSource(certFile);
         }
+        config.overrideAuthority = props.getProperty("override_authority");
         return config;
     }
 
@@ -184,6 +190,7 @@ public static EtcdClusterConfig fromJson(ByteSource source, File dir) throws IOE
                 config.certificate = ByteSource.wrap(jsonConfig.certificate.getBytes(UTF_8));
             }
         }
+        config.overrideAuthority = jsonConfig.overrideAuthority;
         return config;
     }
 
@@ -286,11 +293,14 @@ static class JsonConfig {
         String password;
         @SerializedName("root_prefix") // a.k.a namespace
         String rootPrefix;
+        @Deprecated
         @SerializedName("compose_deployment")
         String composeDeployment;
         @SerializedName("certificate")
         String certificate;
         @SerializedName("certificate_file")
         String certificateFile;
+        @SerializedName("override_authority")
+        String overrideAuthority;
     }
 }