fix: Fix re-authentication when used with etcd server 3.5.x (#76)

This unifies the "reauth required" logic between KV and watch requests and ensures that it covers the new response codes/messages.

Thanks to Kei Sugano sgnk@jp.ibm.com for help with the unit test additions.

Signed-off-by: Nick Hill <nickhill@us.ibm.com>

Author: njhill

.travis.yml

@@ -6,7 +6,7 @@ jdk:
 env:
   jobs:
     - TARGET_ETCD_VERSION=v3.4.20
-    - TARGET_ETCD_VERSION=v3.5.9
+    - TARGET_ETCD_VERSION=v3.5.10
   global:
     - secure: "XB13D71N/3iWgKh/+3TDdqLYQN3hBzD+BvUYSbkB/Pgyl1mjI7CVj03IxeVl7AvLkP8O9t4WTVNxutIMXdEap01cc+cvd8S1aCKgRwbMh3dSHbRowjXB9XccX7h7JE09gkmBxHQWu5lIZqdSG2dWat4YjD0i/IdFDDFtmDV1VKVLnjEvxroHBJJDMo5mYFbrQZB0D+5W2v52JpdMw54tF/vsQ6ThlMN9faEuuD+HQBiEvslMmbl+d9eiDxPhHkdf4fUQr0OBz65V/ggvOd0aP8fRpD+G9bKdnjM+9kkvuCFOXTSlxJzhNO4JGIEJErXumGFjjsu5QwFD+RKrKaTl1V5/6IrkKgmyn/ZVFgI8ZiullwTyxei+7mHd1LAt+DW/GDrCG/Q3s3jrgfc9iMemgaMojrK6iOgtHAYYhQwDuOwdxLeOmqvgBEWvbe7nsebLQG6LOhJPEI7UjcSDXH9m7ag/CpP9q3qEOChpQWGkMo0eWLOmOMujdbLWqFaqNs/ot5yZJXb+BNW3opO40KFm0kG1arw3V/4LB+GZ4Wh52ScCZugBshUa8hB1J4pLlFhyi09IaxiqKCqAHfAtwvLQ2SY1VtBkyryViHX1VSIfPa4LpaHUIarFMZk11Vq3SmCvXrKhl1Vt7iVHDQMGBtO+abmfmGSZTgb3twEdcLc1Mc8="
     - secure: "FggRkH5Ttnz/AfHdwcxy5tnkqA42XwuRx4aDEmaz+CuK+hfKbwIQiqli0PzUQqbPJtEReU85vvyRjbC5lAOvFzQKTrCU3shbXXGq9PGiFVvVNsFB+93akAwtEZYl9CDjRzuYqNiS9hrpze4qMLVZsbRUGSXDVGNzK27//zA1WXbb/htqa4xTphmxLG6CrHI8L65tbbC2WY+qIa5eQHJDLwVRVAisXT/LDpj8ky69I9bRZZbyirkwKsgf7UGyneqqjO/d+3IOfc6ppGPQhe1IeL1XuVtiul9OtksfSBsAttgro7nRgbe2szU+Um2JKcv6dxqqFZJi7SN6TfWpN2f16bU88mJHmyZAILr6YQAHkiWM48vwcZR2MVYH0FI0M1c7xmsDeG6bAQ6H0kWKdLr0NfX1/vm+TPc+OHaCsnTlpcnQx5mcTi4Cztr/oTWMYsrocfuoNOy31gosnLspmjFcqUbjreWvArdZNXmVzj0hp6ewNz/bLszqtI57LdoAMc/bDs8rqElpCr8r12vsQQNgmmKigTlZIe8jNoPnJw00atzg7Dn2XqTHxNnFhtMOhEfmX7fEQ4du1LW0H8RntESjdAIeO8kXttraGmHP5Au7b78zN2WKc21NLBVuwAmcm69G6qcGFdBhuY6xup7aipG0ilB7M+/jllcD4mVAPn3lLGQ="

build-env/install-etcd.sh

@@ -3,7 +3,7 @@
 # Fail on any error
 set -e
 
-ETCD_VERSION=${TARGET_ETCD_VERSION:-v3.5.9}
+ETCD_VERSION=${TARGET_ETCD_VERSION:-v3.5.10}
 
 INSTALL_DIR="${1:-$HOME/etcd}"
 

cd/deploy.sh

@@ -1,7 +1,7 @@
 #!/usr/bin/env bash
 
 echo TRAVIS_JDK_VERSION="$TRAVIS_JDK_VERSION"
-if [ "$TRAVIS_BRANCH" = 'main' ] && [ "$TRAVIS_PULL_REQUEST" == 'false' ] && [ "$TRAVIS_JDK_VERSION" == 'openjdk11' ] && [ "$TARGET_ETCD_VERSION" == 'v3.5.9' ]; then
+if [ "$TRAVIS_BRANCH" = 'main' ] && [ "$TRAVIS_PULL_REQUEST" == 'false' ] && [ "$TRAVIS_JDK_VERSION" == 'openjdk17' ] && [ "$TARGET_ETCD_VERSION" == 'v3.5.10' ]; then
     echo "deploying release to central repository"
 
     # prepare key for signing

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

@@ -586,22 +586,23 @@ public boolean isClosed() {
     // See https://godoc.org/github.com/coreos/etcd/auth#pkg-variables
     // and https://godoc.org/github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes#pkg-variables
     private static final String[] AUTH_FAILURE_MESSAGES = {
-            // These will be prefixed with "etcdserver: "
-            "root user does not exist",
-            "root user does not have root role",
-            "user name already exists",
-            "user name is empty",
-            "user name not found",
-            "role name already exists",
-            "role name not found",
-            "role name is empty",
-            "authentication failed, invalid user ID or password",
-            "permission denied",
-            "role is not granted to the user",
-            "permission is not granted to the role",
-            "authentication is not enabled",
-            "invalid auth token",
-            "invalid auth management"
+        // These will be prefixed with "etcdserver: "
+        "root user does not exist",
+        "root user does not have root role",
+        "user name already exists",
+        "user name is empty",
+        "user name not found",
+        "role name already exists",
+        "role name not found",
+        "role name is empty",
+        "authentication failed, invalid user ID or password",
+        "permission denied",
+        "role is not granted to the user",
+        "permission is not granted to the role",
+        "authentication is not enabled",
+        "invalid auth token",
+        "invalid auth management",
+        "revision of auth store is old"
     };
 
     private Status lastAuthFailStatus;
@@ -613,11 +614,13 @@ protected static boolean reauthRequired(Throwable error) {
         }
         Status status = Status.fromThrowable(error);
         Code code = status.getCode();
-        return code == Code.UNAUTHENTICATED
-                || (OTHER_AUTH_FAILURE_CODES.contains(code) &&
-                        (startsWith(status.getDescription(), "auth: ")
-                                || endsWith(status.getDescription(), AUTH_FAILURE_MESSAGES)))
-                || (code == Code.CANCELLED && reauthRequired(error.getCause()));
+        return reauthRequired(code, status.getDescription()) ||
+                (code == Code.CANCELLED && reauthRequired(error.getCause()));
+    }
+
+    public static boolean reauthRequired(Code code, String message) {
+        return code == Code.UNAUTHENTICATED || (OTHER_AUTH_FAILURE_CODES.contains(code) &&
+                (startsWith(message, "auth: ") || endsWith(message, AUTH_FAILURE_MESSAGES)));
     }
 
     // This is only called in a synchronized context

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

@@ -34,6 +34,7 @@
 import java.util.concurrent.locks.LockSupport;
 import java.util.function.Function;
 import java.util.function.Supplier;
+import java.util.regex.Pattern;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -72,6 +73,8 @@ public class GrpcClient {
 
     private static final Logger logger = LoggerFactory.getLogger(GrpcClient.class);
 
+    private static final Pattern CAMELCASE_PATT = Pattern.compile("([a-z])([A-Z]+)");
+
     private final long defaultTimeoutMs;
 
     public interface AuthProvider {
@@ -115,6 +118,15 @@ default CallCredentials refreshCredentials(Throwable trigger) {
     // modified only by reauthenticate() method
     private CallOptions callOptions = CallOptions.DEFAULT; // volatile tbd - lazy probably ok
 
+    // parse gRPC code from go formatted string (camel-case)
+    public static Code parseGoCodeString(String codeString) {
+        try {
+            return Code.valueOf(CAMELCASE_PATT.matcher(codeString).replaceAll("$1_$2").toUpperCase());
+        } catch (IllegalArgumentException _) {
+            return Code.UNKNOWN;
+        }
+    }
+
     /**
      * @deprecated use other constructor
      */

src/main/java/com/ibm/etcd/client/watch/EtcdWatchClient.java

@@ -25,9 +25,12 @@
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Executor;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import javax.annotation.concurrent.GuardedBy;
 
+import com.ibm.etcd.client.EtcdClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -50,15 +53,15 @@
 import io.grpc.stub.StreamObserver;
 
 /**
- * 
+ *
  */
 public final class EtcdWatchClient implements Closeable {
 
     private static final Logger logger = LoggerFactory.getLogger(EtcdWatchClient.class);
 
     private static final Exception CANCEL_EXCEPTION = new CancellationException();
 
-    private static final String UNAUTH_REASON_PREFIX = "rpc error: code = PermissionDenied";
+    private static final Pattern CANCEL_REASON_PATT = Pattern.compile("rpc error: code = (\\w+) desc = (.*)");
 
     private static final MethodDescriptor<WatchRequest,WatchResponse> METHOD_WATCH =
             WatchGrpc.getWatchMethod();
@@ -109,7 +112,7 @@ final class WatcherRecord {
         long upToRevision;
         long watchId = -2L; // -2 only pre-creation, >= -1 after
         boolean lastCreateFailedAuth;
- 
+
         boolean userCancelled, finished;
         volatile boolean vUserCancelled;
 
@@ -176,9 +179,8 @@ public void processWatchEvents(final WatchResponse wr) {
         public boolean processCreatedResponse(WatchResponse wr, boolean cancelled) {
             long newWatchId = wr.getWatchId();
             if (cancelled || newWatchId == -1L) {
-                String reason = wr.getCancelReason();
-                if (reason != null && reason.startsWith(UNAUTH_REASON_PREFIX)
-                        && !lastCreateFailedAuth) {
+                String cancelReason = wr.getCancelReason();
+                if (!lastCreateFailedAuth && reauthRequired(cancelReason)) {
                     // If watch creation fails due to an authentication issue (likely
                     // expired), we trigger a reauth+refresh of the stream by sending
                     // an UNAUTHENTICATED error. This watch must first be re-created
@@ -190,7 +192,7 @@ public boolean processCreatedResponse(WatchResponse wr, boolean cancelled) {
                             if (reqStream != null) {
                                 lastCreateFailedAuth = true;
                                 reqStream.onError(Code.UNAUTHENTICATED
-                                    .toStatus().withDescription(reason).asException());
+                                    .toStatus().withDescription(cancelReason).asException());
                             }
                         }
                     }
@@ -284,6 +286,18 @@ private void completeCreateFuture(boolean created, Exception error) {
         }
     }
 
+    static boolean reauthRequired(String cancelReason) {
+        if (cancelReason != null) {
+            Matcher m = CANCEL_REASON_PATT.matcher(cancelReason);
+            if (m.matches()) {
+                Code code = GrpcClient.parseGoCodeString(m.group(1));
+                String message = m.group(2);
+                return EtcdClient.reauthRequired(code, message);
+            }
+        }
+        return false;
+    }
+
 //  @Override
     public Watch watch(WatchCreateRequest createReq, StreamObserver<WatchUpdate> observer) {
         return watch(createReq, observer, null);
@@ -559,4 +573,4 @@ public void close() {
             }
         });
     }
-}
+}

src/test/java/com/ibm/etcd/client/EtcdTestSuite.java

@@ -19,9 +19,12 @@
 import java.io.IOException;
 import java.io.InputStreamReader;
 import java.io.Reader;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.UUID;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
@@ -50,31 +53,55 @@
     })
 public class EtcdTestSuite {
 
-    static Process etcdProcess, etcdTlsProcess, etcdTlsCaProcess;
+    static Process etcdProcess, etcdTlsProcess, etcdTlsCaProcess, etcdJwtProcess;
 
     static final String etcdCommand;
+    static final String etcdctlCommand;
     static {
          String etcd = System.getenv("ETCD_CMD");
          etcdCommand = etcd != null ? etcd : "etcd";
+         String etcdctl = System.getenv("ETCDCTL_CMD");
+         etcdctlCommand = etcdctl != null ? etcdctl : "etcdctl";
     }
 
     static final String clientKey = EtcdTestSuite.class.getResource("/client.key").getFile();
     static final String clientCert = EtcdTestSuite.class.getResource("/client.crt").getFile();
     static final String serverKey = EtcdTestSuite.class.getResource("/server.key").getFile();
     static final String serverCert = EtcdTestSuite.class.getResource("/server.crt").getFile();
+    static final String jwtKey = EtcdTestSuite.class.getResource("/jwt_RS256.key").getFile(); // openssl genrsa -out jwt_RS256.key 4096
+    static final String jwtPub = EtcdTestSuite.class.getResource("/jwt_RS256.pub").getFile(); // openssl rsa -in jwt_RS256.key -pubout > jwt_RS256.pub
+
+    static final String userName = "root";
+    static final String userPwd = UUID.randomUUID().toString();
 
     @BeforeClass
     public static void setUp() throws Exception {
         etcdProcess = startProcess();
+
         etcdTlsProcess = startProcess("--cert-file=" + serverCert,
                 "--key-file=" + serverKey, "--listen-client-urls=https://localhost:2360",
                 "--listen-peer-urls=http://localhost:2361",
                 "--advertise-client-urls=https://localhost:2360", "--name=tls");
-        etcdTlsCaProcess = null; startProcess("--cert-file=" + serverCert,
+
+        etcdTlsCaProcess = startProcess("--cert-file=" + serverCert,
                 "--key-file=" + serverKey, "--listen-client-urls=https://localhost:2362",
                 "--listen-peer-urls=http://localhost:2363",
                 "--advertise-client-urls=https://localhost:2362", "--name=tls-ca", 
                 "--trusted-ca-file=" + clientCert, "--client-cert-auth");
+
+        Path tmpDir = Files.createTempDirectory(null);
+        tmpDir.toFile().deleteOnExit();
+        etcdJwtProcess = startProcess("--auth-token=jwt,pub-key=" + jwtPub + ",priv-key=" + jwtKey + ",sign-method=RS256,ttl=4s",
+                "--listen-client-urls=http://localhost:2365",
+                "--listen-peer-urls=http://localhost:2364",
+                "--advertise-client-urls=http://localhost:2365",
+                "--data-dir=" + tmpDir);
+        executeCtlCommand("--endpoints=localhost:2365",
+                "user", "add", userName, "--new-user-password", userPwd);
+        executeCtlCommand("--endpoints=localhost:2365",
+                "user", "grant-role", userName, "root");
+        executeCtlCommand("--endpoints=localhost:2365",
+                "auth", "enable");
     }
 
     private static Process startProcess(String... cmdline) throws Exception {
@@ -98,11 +125,34 @@ private static Process startProcess(String... cmdline) throws Exception {
         }
     }
 
+    private static void executeCtlCommand(String... cmdline) throws Exception {
+        boolean ok = false;
+        Process p = null;
+        try {
+            List<String> cmd = new ArrayList<>();
+            cmd.add(etcdctlCommand);
+            cmd.addAll(Arrays.asList(cmdline));
+            p = new ProcessBuilder(cmd)
+                    .redirectErrorStream(true).start();
+            waitForStartup(p);
+            p.waitFor(30L, TimeUnit.SECONDS);
+            System.out.println("etcdctl exit value:" + p.exitValue());
+            ok = true;
+        } catch (IOException e) {
+            System.out.println("Failed to execute etcdctl: " + e);
+        } finally {
+            if (!ok) {
+                tearDown(p);
+            }
+        }
+    }
+
     @AfterClass
     public static void tearDown() {
         tearDown(etcdProcess);
         tearDown(etcdTlsProcess);
         tearDown(etcdTlsCaProcess);
+        tearDown(etcdJwtProcess);
     }
 
     public static void tearDown(Process process) {

src/test/java/com/ibm/etcd/client/JsonConfigTest.java

@@ -109,6 +109,7 @@ public void testLifecycle() throws Exception {
 
         EtcdClient c4 = config.getClient();
         assertNotSame(c1, c4);
+        c4.close();
     }
 
     void runBasicTests(EtcdClusterConfig config) throws Exception {