Add support for specifying client TLS key/cert in json-based config (#41)

* Add support for specifying client TLS key/cert in json-based config

Includes updates to EtcdClient/EtcdClusterConfig to ensure that
cached/shared client instances are appropriately reference counted.

* Fix PersistentLeaseKeyTest timing flake

Author: njhill

etcd-json-schema.md

@@ -9,15 +9,18 @@ Example JSON config doc:
   "password": "password",
   "root_prefix": "aka-chroot-or-namespace",
   "certificate_file": "etcd-dev.pem",
-  "override_authority": "etcd-development-01"
+  "override_authority": "etcd-development-01",
+  "client_key_file": "etcd-client-key.key",
+  "client_certificate_file": "etcd-client-cert.crt"
 }
 ```
 
 - 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 `override_authority` is optional and may be used to override the authority used for TLS hostname verification for _all_ endpoints.
+- `client_key_file` and `client_certificate_file` form an optional key/cert pair for TLS client-auth. Either may also be embedded in a similar way by instead including `client_key` and/or `client_certificate` string attributes.
+- The `override_authority` attribute is optional and may be used to override the authority used for TLS hostname verification for _all_ endpoints.
 
 Example with embedded (trunctated) TLS cert:
 

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

@@ -89,6 +89,9 @@
 import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.util.AbstractReferenceCounted;
+import io.netty.util.IllegalReferenceCountException;
+import io.netty.util.ReferenceCounted;
 import io.netty.util.concurrent.FastThreadLocalThread;
 
 public class EtcdClient implements KvStoreClient {
@@ -111,6 +114,8 @@ public class EtcdClient implements KvStoreClient {
     protected static final Pattern ADDR_PATT =
             Pattern.compile("(?:https?://|dns:///)?([a-zA-Z0-9\\-.]+)(?::(\\d+))?");
 
+    private final ReferenceCounted refCount; // may be null
+
     private final int sessionTimeoutSecs;
 
     private final ByteString name, password;
@@ -140,6 +145,7 @@ public static class Builder {
         private Executor executor; // for call-backs
         private boolean sendViaEventLoop = true; // default true
         private int sessTimeoutSecs = DEFAULT_SESSION_TIMEOUT_SECS;
+        private boolean refCounted;
 
         Builder(List<String> endpoints) {
             this.endpoints = Preconditions.checkNotNull(endpoints);
@@ -337,8 +343,8 @@ public EtcdClient build() {
             if (maxInboundMessageSize != 0) {
                 ncb.maxInboundMessageSize(maxInboundMessageSize);
             }
-            return new EtcdClient(ncb, defaultTimeoutMs, name, password,
-                    preemptAuth, threads, executor, sendViaEventLoop, sessTimeoutSecs);
+            return new EtcdClient(ncb, defaultTimeoutMs, name, password, preemptAuth,
+                    threads, executor, sendViaEventLoop, sessTimeoutSecs, refCounted);
         }
     }
 
@@ -391,7 +397,7 @@ public static Builder forEndpoints(String endpoints) {
     EtcdClient(NettyChannelBuilder chanBuilder, long defaultTimeoutMs,
             ByteString name, ByteString password, boolean initialAuth,
             int threads, Executor userExecutor, boolean sendViaEventLoop,
-            int sessTimeoutSecs) {
+            int sessTimeoutSecs, boolean refCounted) {
 
         if (name == null && password != null) {
             throw new IllegalArgumentException("password without name");
@@ -400,6 +406,17 @@ public static Builder forEndpoints(String endpoints) {
         this.password = password;
         this.sessionTimeoutSecs = sessTimeoutSecs;
 
+        this.refCount = !refCounted ? null : new AbstractReferenceCounted() {
+            @Override
+            public ReferenceCounted touch(Object hint) {
+                return null;
+            }
+            @Override
+            protected void deallocate() {
+                doClose();
+            }
+        };
+        
         chanBuilder.keepAliveTime(10L, SECONDS);
         chanBuilder.keepAliveTimeout(8L, SECONDS);
 
@@ -462,6 +479,22 @@ public CallCredentials refreshCredentials() {
 
     @Override
     public void close() {
+        if (refCount != null) {
+            refCount.release();
+        } else {
+            doClose();
+        }
+    }
+
+    boolean retain() {
+        if (refCount != null) try {
+            refCount.retain();
+            return true;
+        } catch (IllegalReferenceCountException irce) {}
+        return false;
+    }
+
+    void doClose() {
         if (leaseClient == CLOSED) {
             return;
         }
@@ -792,4 +825,22 @@ MultithreadEventLoopGroup getInternalExecutor() {
         return internalExecutor;
     }
 
+    // Public for access by EtcdClusterConfig from config package
+    public static final class Internal {
+        private Internal() {}
+
+        public static boolean retain(EtcdClient client) {
+            return client != null && client.retain();
+        }
+
+        public static void cleanup(EtcdClient client) {
+            if (client != null) {
+                client.doClose();
+            }
+        }
+
+        public static void makeRefCounted(Builder builder) {
+            builder.refCounted = true;
+        }
+    }
 }

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

@@ -73,6 +73,9 @@ public enum TlsMode { TLS, PLAINTEXT, AUTO }
     String composeDeployment;
     ByteSource certificate;
     String overrideAuthority;
+    // either both or neither of these must be set
+    ByteSource clientKey;
+    ByteSource clientCertificate;
 
     protected EtcdClusterConfig() {}
 
@@ -93,23 +96,34 @@ private EtcdClient newClient() throws IOException, CertificateException {
         EtcdClient.Builder builder = EtcdClient.forEndpoints(endpointList)
                 .withCredentials(user, password).withImmediateAuth()
                 .withMaxInboundMessageSize(maxMessageSize);
+        EtcdClient.Internal.makeRefCounted(builder);
         if (overrideAuthority != null) {
             builder.overrideAuthority(overrideAuthority);
         }
         TlsMode ssl = tlsMode;
         if (ssl == TlsMode.AUTO || ssl == null) {
             String ep = endpointList.get(0);
-            ssl = ep.startsWith("https://")
-                    || (!ep.startsWith("http://") && certificate != null)
-                    ? TlsMode.TLS : TlsMode.PLAINTEXT;
+            if (ep.startsWith("http://")) ssl = TlsMode.PLAINTEXT;
+            else if (certificate != null || clientCertificate != null
+                    || ep.startsWith("https://")) {
+                ssl = TlsMode.TLS;
+            }
         }
         if (ssl == TlsMode.PLAINTEXT) {
             builder.withPlainText();
-        } else if (composeDeployment != null) {
-            builder.withTrustManager(new ComposeTrustManagerFactory(composeDeployment,
-                    composeDeployment, certificate));
-        } else if (certificate != null) {
-            builder.withCaCert(certificate);
+        } else {
+            if (composeDeployment != null) {
+                builder.withTrustManager(new ComposeTrustManagerFactory(composeDeployment,
+                        composeDeployment, certificate));
+            } else if (certificate != null) {
+                builder.withCaCert(certificate);
+            }
+            if (clientCertificate != null && clientKey != null) {
+                try (InputStream keyStream = clientKey.openStream();
+                     InputStream certStream = clientCertificate.openStream()) {
+                    builder.withTlsConfig(b -> b.keyManager(certStream, keyStream));
+                }
+            }
         }
         if (isShutdown) {
             throw new IllegalStateException("shutdown");
@@ -141,19 +155,30 @@ public static EtcdClusterConfig fromProperties(ByteSource source) throws IOExcep
         config.composeDeployment = props.getProperty("compose_deployment");
         config.rootPrefix = bs(props.getProperty("root_prefix")); // a.k.a namespace
         String tlsMode = props.getProperty("tls_mode");
-        if (tlsMode != null) {
+        if (tlsMode != null) try {
             config.tlsMode = TlsMode.valueOf(tlsMode);
+        } catch (IllegalArgumentException iae) {
+            throw new IOException("Invalid value "
+                    + tlsMode + " for etcd tls_mode config property");
         }
-        String certPath = props.getProperty("certificate_file");
-        if (certPath != null) {
-            File certFile = new File(certPath);
-            if (!certFile.exists()) {
-                throw new IOException("cant find certificate file: " + certPath);
-            }
-            config.certificate = Files.asByteSource(certFile);
-        }
+        config.clientKey = certFromProperties("client_key", props);
+        config.clientCertificate = certFromProperties("client_certificate", props);
+        config.certificate = certFromProperties("certificate_file", props);
         config.overrideAuthority = props.getProperty("override_authority");
-        return config;
+        return validateConfig(config);
+    }
+
+    private static ByteSource certFromProperties(String certFilePropName,
+            Properties props) throws IOException {
+        String certPath = props.getProperty(certFilePropName);
+        if (certPath == null) {
+            return null;
+        }
+        File certFile = new File(certPath);
+        if (certFile.exists()) {
+            return Files.asByteSource(certFile);
+        }
+        throw new IOException("cant find certificate file: " + certPath);
     }
 
     public static EtcdClusterConfig fromJson(ByteSource source, File dir) throws IOException {
@@ -170,27 +195,53 @@ public static EtcdClusterConfig fromJson(ByteSource source, File dir) throws IOE
         config.password = bs(jsonConfig.password);
         config.composeDeployment = jsonConfig.composeDeployment;
         config.rootPrefix = bs(jsonConfig.rootPrefix);
-        if (jsonConfig.certificateFile != null) {
-            File certFile = new File(jsonConfig.certificateFile);
+        if (jsonConfig.tlsMode != null) try {
+            config.tlsMode = TlsMode.valueOf(jsonConfig.tlsMode);
+        } catch (IllegalArgumentException iae) {
+            throw new IOException("Invalid value "
+                    + jsonConfig.tlsMode + " for etcd tls_mode config field");
+        }
+        config.clientKey = certFromJson(jsonConfig.clientKeyFile, jsonConfig.clientKey, dir);
+        config.clientCertificate = certFromJson(
+                jsonConfig.clientCertificateFile, jsonConfig.clientCertificate, dir);
+        config.certificate = certFromJson(jsonConfig.certificateFile, jsonConfig.certificate, dir);
+        config.overrideAuthority = jsonConfig.overrideAuthority;
+        return validateConfig(config);
+    }
+
+    private static ByteSource certFromJson(String certFileName, String literalCert,
+            File dir) throws IOException {
+        if (certFileName != null) {
+            File certFile = new File(certFileName);
             if (dir != null && !certFile.exists()) {
                 // try same dir as the config file
-                certFile = new File(dir, jsonConfig.certificateFile);
+                certFile = new File(dir, certFileName);
             }
             if (certFile.exists()) {
-                config.certificate = Files.asByteSource(certFile);
-            } else {
+                if (literalCert != null) {
+                    logger.warn("Ignoring json-embedded cert because file "
+                            + certFileName + " was also provided");
+                }
+                return Files.asByteSource(certFile);
+            } else if (literalCert != null) {
                 // will fall back to embedded if present
-                logger.warn("Can't find certificate file: " + jsonConfig.certificateFile);
-            }
-        }
-        if (jsonConfig.certificate != null) {
-            if (config.certificate != null) {
-                logger.warn("Ignoring json-embedded cert because file was also provided");
+                logger.warn("Can't find certificate file: " + certFileName);
             } else {
-                config.certificate = ByteSource.wrap(jsonConfig.certificate.getBytes(UTF_8));
+                throw new IOException("Can't find certificate file: " + certFileName);
             }
         }
-        config.overrideAuthority = jsonConfig.overrideAuthority;
+        return literalCert != null ? ByteSource.wrap(literalCert.getBytes(UTF_8)) : null;
+    }
+
+    private static EtcdClusterConfig validateConfig(EtcdClusterConfig config) throws IOException {
+        if (config.composeDeployment != null) {
+            logger.warn("compose_deployment config param is deprecated," +
+                    " use override_authority to set a specific name for TLS SNI");
+        }
+        if ((config.clientKey == null) != (config.clientCertificate == null)) {
+            throw new IOException("Must specify either both or neither of TLS client_key "
+                    + "and client_certificate attributes");
+        }
         return config;
     }
 
@@ -225,12 +276,26 @@ public static EtcdClusterConfig fromJsonFileOrSimple(String fileOrSimpleString)
         throw new FileNotFoundException("etcd config json file not found: " + f);
     }
 
-    private static final Cache<CacheKey,EtcdClient> clientCache = CacheBuilder.newBuilder().weakValues()
-            .<CacheKey,EtcdClient>removalListener(rn -> rn.getValue().close()).build();
+    private static final Cache<CacheKey, EtcdClient> clientCache = CacheBuilder.newBuilder().weakValues()
+            .<CacheKey, EtcdClient>removalListener(rn -> EtcdClient.Internal.cleanup(rn.getValue())).build();
 
     public static EtcdClient getClient(EtcdClusterConfig config) throws IOException, CertificateException {
         try {
-            return clientCache.get(new CacheKey(config), config::newClient);
+            // Share equivalent ref-counted client from cache if possible
+            final CacheKey key = new CacheKey(config);
+            while (true) {
+                EtcdClient client = clientCache.getIfPresent(key);
+                if (client == null) {
+                    EtcdClient[] maybeNew = new EtcdClient[1];
+                    client  = clientCache.get(key, () -> maybeNew[0] = config.newClient());
+                    // if client != maybeNew[0] then we got a pre-existing client
+                    if (client != maybeNew[0] && !EtcdClient.Internal.retain(client)) {
+                        clientCache.invalidate(key);
+                        continue;
+                    }
+                }
+                return client;
+            }
         } catch (ExecutionException ee) {
             Throwables.throwIfInstanceOf(ee.getCause(), IOException.class);
             Throwables.throwIfInstanceOf(ee.getCause(), CertificateException.class);
@@ -267,12 +332,16 @@ public boolean equals(Object obj) {
             return Objects.equals(config.endpoints, other.endpoints)
                     && Objects.equals(config.composeDeployment, other.composeDeployment)
                     && Objects.equals(config.user, other.user)
-                    && Objects.equals(config.tlsMode, other.tlsMode);
+                    && Objects.equals(config.tlsMode, other.tlsMode)
+                    && (config.certificate == null) == (other.certificate == null)
+                    && (config.clientKey == null) == (other.clientKey == null)
+                    && (config.clientCertificate == null) == (other.clientCertificate == null);
         }
         @Override
         public int hashCode() {
             return Objects.hash(config.endpoints, config.user,
-                    config.composeDeployment, config.tlsMode);
+                    config.composeDeployment, config.tlsMode, config.certificate == null,
+                    config.clientKey == null, config.clientCertificate == null);
         }
     }
 
@@ -296,11 +365,21 @@ static class JsonConfig {
         @Deprecated
         @SerializedName("compose_deployment")
         String composeDeployment;
+        @SerializedName("tls_mode")
+        String tlsMode;
         @SerializedName("certificate")
         String certificate;
         @SerializedName("certificate_file")
         String certificateFile;
         @SerializedName("override_authority")
         String overrideAuthority;
+        @SerializedName("client_key")
+        String clientKey;
+        @SerializedName("client_key_file")
+        String clientKeyFile;
+        @SerializedName("client_certificate")
+        String clientCertificate;
+        @SerializedName("client_certificate_file")
+        String clientCertificateFile;
     }
 }