Widen error responses that trigger re-authentication

njhill · njhill · commit c0d3465d15b2 · 2020-06-16T11:23:33.000-07:00
Apart from token expiry, we have encountered at least two other kinds of
persistent error responses that are resolved by re-authentication
(presumably other kinds of token invalidation). The client is
effectively broken once this happens until a re-auth is done.

This change widens the "requires re-auth" criteria used to catch those
and potentially others.

It also includes some related hardening/refactoring of how automatic
re-auths are handled. The original RPC that triggered the auth failure
is now included in the cause chain.
diff --git a/src/main/java/com/ibm/etcd/client/EtcdClient.java b/src/main/java/com/ibm/etcd/client/EtcdClient.java
@@ -22,6 +22,7 @@
 import java.io.InputStream;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.BrokenBarrierException;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CyclicBarrier;
@@ -42,6 +43,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Predicate;
 import com.google.common.base.Predicates;
+import com.google.common.collect.ImmutableSet;
 import com.google.common.io.ByteSource;
 import com.google.common.util.concurrent.ForwardingExecutorService;
 import com.google.common.util.concurrent.Futures;
@@ -289,8 +291,12 @@ public boolean requiresReauth(Throwable t) {
                 return EtcdClient.reauthRequired(t);
             }
             @Override
+            public CallCredentials refreshCredentials(Throwable trigger) {
+                return EtcdClient.this.refreshCredentials(trigger);
+            }
+            @Override
             public CallCredentials refreshCredentials() {
-                return EtcdClient.this.refreshCredentials();
+                return refreshCredentials(null);
             }
         };
 
@@ -385,21 +391,68 @@ public boolean isClosed() {
 
     // ------ authentication logic
 
+    private static final Set<Code> OTHER_AUTH_FAILURE_CODES = ImmutableSet.of(
+            Code.INVALID_ARGUMENT, Code.FAILED_PRECONDITION, Code.PERMISSION_DENIED, Code.UNKNOWN);
+
+    // Various different errors can imply a re-auth is required (sometimes non-obvious),
+    // so we cover most related messages to be safe. This should not cause problems since
+    // re-authentication attempts are rate-limited to once every 15 seconds.
+    //
+    // 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"
+    };
+
+    private Status lastAuthFailStatus;
+    private long lastAuthFailTime;
+
     protected static boolean reauthRequired(Throwable error) {
-        Code statusCode = GrpcClient.codeFromThrowable(error);
-        return statusCode == Code.UNAUTHENTICATED ||
-                (statusCode == Code.INVALID_ARGUMENT
-                && contains(error.getMessage(), "user name is empty") ||
-                (statusCode == Code.CANCELLED
-                && reauthRequired(error.getCause())));
+        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()));
     }
 
-    private CallCredentials refreshCredentials() {
+    // This is only called in a synchronized context
+    private CallCredentials refreshCredentials(Throwable trigger) {
+        long lastAuthFailure = System.currentTimeMillis() - lastAuthFailTime;
+        if (lastAuthFailure < 15_000L) {
+            // Rate-limit how often re-auth attempts are made,
+            // return last auth failure in the meantime
+            return new CallCredentials() {
+                @Override
+                public void applyRequestMetadata(RequestInfo requestInfo,
+                        Executor appExecutor, MetadataApplier applier) {
+                    applier.fail(lastAuthFailStatus);
+                }
+                //@Override
+                public void thisUsesUnstableApi() {}
+            };
+        }
         return new CallCredentials() {
-            private Metadata tokenHeader; //TODO volatile TBD
-            private final long authTime = System.currentTimeMillis();
+            private volatile Metadata tokenHeader; //TODO volatile TBD
             private final ListenableFuture<Metadata> futureTokenHeader =
                     Futures.transform(authenticate(), ar -> tokenHeader = tokenHeader(ar), directExecutor());
+
             @Override
             public void applyRequestMetadata(RequestInfo requestInfo,
                     Executor appExecutor, MetadataApplier applier) {
@@ -411,11 +464,14 @@ public void applyRequestMetadata(RequestInfo requestInfo,
                         applier.apply(futureTokenHeader.get());
                     } catch (ExecutionException | InterruptedException ee) { // (IE won't be thrown)
                         Status failStatus = Status.fromThrowable(ee.getCause());
-                        Code code = failStatus != null ? failStatus.getCode() : null;
-                        if (code != Code.INVALID_ARGUMENT
-                                && (System.currentTimeMillis() - authTime) > 15_000L) {
-                            // this will force another auth attempt
-                            failStatus = Status.UNAUTHENTICATED.withDescription("re-attempt re-auth");
+                        lastAuthFailStatus = failStatus;
+                        lastAuthFailTime = System.currentTimeMillis();
+                        if (trigger != null) {
+                            if (failStatus.getCause() != null) {
+                                failStatus.getCause().addSuppressed(trigger);
+                            } else {
+                                failStatus = failStatus.withCause(trigger);
+                            }
                         }
                         applier.fail(failStatus);
                     }
@@ -439,16 +495,15 @@ private ListenableFuture<AuthenticateResponse> authenticate() {
         CallOptions callOpts = CallOptions.DEFAULT;
         return Futures.catchingAsync(
                 grpc.fuCall(METHOD_AUTHENTICATE, request, callOpts, 0L),
-                Exception.class, ex -> !retryAuthRequest(ex)
+                    Exception.class, ex -> !retryAuthRequest(ex)
                         ? Futures.immediateFailedFuture(ex)
                         : grpc.fuCall(METHOD_AUTHENTICATE, request, callOpts, 0L),
                 directExecutor());
     }
 
     protected static boolean retryAuthRequest(Throwable error) {
-        Status status = Status.fromThrowable(error);
-        Code statusCode = status != null ? status.getCode() : null;
-        return statusCode == Code.UNAVAILABLE && GrpcClient.isConnectException(error);
+        return GrpcClient.codeFromThrowable(error) == Code.UNAVAILABLE
+                && GrpcClient.isConnectException(error);
     }
 
     // ------ individual client getters
@@ -518,6 +573,19 @@ protected static boolean contains(String str, String subStr) {
         return str != null && str.contains(subStr);
     }
 
+    protected static boolean startsWith(String str, String prefix) {
+        return str != null && str.startsWith(prefix);
+    }
+
+    protected static boolean endsWith(String str, String... suffixes) {
+        if (str != null) {
+            for (String suffix : suffixes) {
+                if (str.endsWith(suffix)) return true;
+            }
+        }
+        return false;
+    }
+
     protected static final class EtcdEventThread extends FastThreadLocalThread {
         public EtcdEventThread(Runnable r) {
             super(r);
diff --git a/src/main/java/com/ibm/etcd/client/GrpcClient.java b/src/main/java/com/ibm/etcd/client/GrpcClient.java
@@ -75,7 +75,16 @@ public class GrpcClient {
     private final long defaultTimeoutMs;
  
     public interface AuthProvider {
+        /**
+         * Called from a synchronized context
+         */
         CallCredentials refreshCredentials();
+        default CallCredentials refreshCredentials(Throwable trigger) {
+            return refreshCredentials();
+        }
+        /**
+         * Not called from a synchronized context
+         */
         boolean requiresReauth(Throwable t);
     }
 
@@ -166,7 +175,7 @@ public Executor getResponseExecutor() {
 
     public void authenticateNow() {
         assert authProvider != NO_AUTH;
-        reauthenticate(getCallOptions());
+        reauthenticate(getCallOptions(), null);
     }
 
     protected CallOptions getCallOptions() {
@@ -201,13 +210,13 @@ public static <R> RetryDecision<R> retryDecision(boolean idempotent) {
     public <ReqT,R> ListenableFuture<R> call(MethodDescriptor<ReqT,R> method,
             ReqT request, boolean idempotent) {
         return call(method, null, request, null, retryDecision(idempotent),
-                0, false, null, 0L);
+                0, false, false, null, 0L);
     }
 
     public <ReqT,R> ListenableFuture<R> call(MethodDescriptor<ReqT,R> method,
             ReqT request, boolean idempotent, long timeoutMillis, Executor executor) {
         return call(method, null, request, executor, retryDecision(idempotent),
-                0, false, null, timeoutMillis);
+                0, false, false, null, timeoutMillis);
     }
 
     //TODO probably move this
@@ -218,14 +227,14 @@ public static interface RetryDecision<ReqT> {
     public <ReqT,R> ListenableFuture<R> call(MethodDescriptor<ReqT,R> method, Condition precondition,
             ReqT request, Executor executor, RetryDecision<ReqT> retry, boolean backoff,
             Deadline deadline, long timeoutMs) {
-        return call(method, precondition, request, executor, retry, 0, backoff, deadline, timeoutMs);
+        return call(method, precondition, request, executor, retry, 0, false, backoff, deadline, timeoutMs);
     }
 
     // deadline is for entire request (including retry pauses),
     // timeout is per-attempt and 0 means not specified
     private <ReqT,R> ListenableFuture<R> call(MethodDescriptor<ReqT,R> method,
             Condition precondition, ReqT request, Executor executor, RetryDecision<ReqT> retry,
-            int attempt, boolean backoff, Deadline deadline, long timeoutMs) {
+            int attempt, boolean afterReauth, boolean backoff, Deadline deadline, long timeoutMs) {
         if (precondition != null && !precondition.satisfied()) {
             return failInExecutor(new CancellationException("precondition false"), executor);
         }
@@ -237,18 +246,29 @@ private <ReqT,R> ListenableFuture<R> call(MethodDescriptor<ReqT,R> method,
             callOpts = callOpts.withExecutor(executor);
         }
         return Futures.catchingAsync(fuCall(method, request, callOpts, timeoutMs), Exception.class, t -> {
-            boolean reauth;
-            // first case: if no retries to do then fail
-            if ((!backoff && attempt > 0) || (deadline != null && deadline.isExpired())
-                    || (!(reauth = reauthIfRequired(t, baseCallOpts))
-                            && !retry.retry(t, request))) {
+            // first cases: determine if we fail immediately
+            if ((!backoff && attempt > 0) || (deadline != null && deadline.isExpired())) {
+                // multiple retries disabled or deadline expired
+                return Futures.immediateFailedFuture(t);
+            }
+            boolean reauth = false;
+            if (authProvider.requiresReauth(t)) {
+                if (afterReauth) {
+                    // if we have an auth failure immediately following a reauth, give up
+                    // (important to avoid infinite loop of auth failures)
+                    return Futures.immediateFailedFuture(t);
+                }
+                reauthenticate(baseCallOpts, t);
+                reauth = true;
+            } else if (!retry.retry(t, request)) {
+                // retry predicate says no (non retryable request and/or error)
                 return Futures.immediateFailedFuture(t);
             }
 
             // second case: immediate retry (first failure or after auth failure + reauth)
             if (reauth || attempt == 0 && immediateRetryLimiter.tryAcquire()) {
                 return call(method, precondition, request, executor, retry,
-                        reauth ? attempt : 1, backoff, deadline, timeoutMs);
+                        reauth ? attempt : 1, reauth, backoff, deadline, timeoutMs);
             }
             int nextAttempt = attempt <= 1 ? 2 : attempt + 1; // skip attempt if we were rate-limited
 
@@ -257,8 +277,8 @@ private <ReqT,R> ListenableFuture<R> call(MethodDescriptor<ReqT,R> method,
             if (deadline != null && deadline.timeRemaining(MILLISECONDS) < delayMs) {
                 return Futures.immediateFailedFuture(t);
             }
-            return Futures.scheduleAsync(() -> call(method, precondition, request, executor,
-                    retry, nextAttempt, backoff, deadline, timeoutMs), delayMs, MILLISECONDS, ses);
+            return Futures.scheduleAsync(() -> call(method, precondition, request, executor, retry,
+                    nextAttempt, false, backoff, deadline, timeoutMs), delayMs, MILLISECONDS, ses);
         }, executor != null ? executor : directExecutor());
     }
 
@@ -313,19 +333,15 @@ protected <ReqT,R> ListenableFuture<R> fuCall(MethodDescriptor<ReqT,R> method, R
 
     protected boolean retryableStreamError(Throwable error) {
         return (Status.fromThrowable(error).getCode() != Code.INVALID_ARGUMENT
-                && !causedByJavaError(error));
-    }
-
-    protected static boolean causedByJavaError(Throwable t) {
-        return causedBy(t, Error.class);
+                && !causedBy(error, Error.class));
     }
 
     /**
      * @return true if reauthentication was required and attempted
      */
     protected boolean reauthIfRequired(Throwable error, CallOptions callOpts) {
         if (authProvider.requiresReauth(error)) {
-            reauthenticate(callOpts);
+            reauthenticate(callOpts, error);
             return true;
         }
         return false;
@@ -336,18 +352,18 @@ public static boolean isConnectException(Throwable t) {
     }
 
     public static Code codeFromThrowable(Throwable t) {
-        Status status = Status.fromThrowable(t);
-        return status != null ? status.getCode() : null;
+        return Status.fromThrowable(t).getCode(); // fromThrowable won't return null
     }
 
 
-    private void reauthenticate(CallOptions failedOpts) {
+    private void reauthenticate(CallOptions failedOpts, Throwable authFailure) {
         // assert name != null && password != null;
         if (getCallOptions() == failedOpts) { // obj identity comparison intentional
             synchronized (this) {
                 CallOptions callOpts = getCallOptions();
                 if (callOpts == failedOpts) {
-                    callOptions = callOpts.withCallCredentials(authProvider.refreshCredentials());
+                    callOptions = callOpts.withCallCredentials(
+                            authProvider.refreshCredentials(authFailure));
                 }
             }
         }
@@ -413,6 +429,8 @@ final class ResilientBiDiStream<ReqT,RespT> {
         private boolean finished;
         private Throwable error;
 
+        private boolean lastAuthFailed;
+
         /**
          * 
          * @param method
@@ -605,6 +623,7 @@ public void beforeStart(ClientCallStreamObserver<ReqT> rs) {
             // called from grpc response thread
             @Override
             public void onNext(RespT value) {
+                lastAuthFailed = false;
                 respStream.onNext(value);
             }
             // called from grpc response thread
@@ -614,9 +633,10 @@ public void onError(Throwable t) {
                 if (finished) {
                     finalError = true;
                 } else {
-                    reauthed = reauthIfRequired(t, sentCallOptions);
+                    reauthed = !lastAuthFailed && reauthIfRequired(t, sentCallOptions);
                     finalError = !reauthed && !retryableStreamError(t);
                 }
+                lastAuthFailed = reauthed;
                 if (!finalError) {
                     int errCount = -1;
                     String msg;
@@ -670,6 +690,7 @@ public void onError(Throwable t) {
             // called from grpc response thread
             @Override
             public void onCompleted() {
+                lastAuthFailed = false;
                 if (!finished) {
                     logger.warn("Unexpected onCompleted received"
                             + " for stream of method " + method.getFullMethodName());
diff --git a/src/main/java/com/ibm/etcd/client/watch/EtcdWatchClient.java b/src/main/java/com/ibm/etcd/client/watch/EtcdWatchClient.java
