Use consistent view of realms for authentication (#38815)

This change updates the authentication service to use a consistent view
of the realms based on the license state at the start of
authentication. Without this, the license can change during
authentication of a request and it will result in a failure if the
realm that extracted the token is no longer in the realm list. This
manifests in some tests as an authentication failure that should never
really happen; one example would be the test framework's transport
client user should always have a succesful authentication but in the
LicensingTests this can fail and will show up as a
NoNodeAvailableException.

Additionally, the licensing tests have been updated to ensure that
there is consistency when changing the license. The license is changed
by modifying the internal xpack license state on each node, which has
no protection against be changed by some pending cluster action. The
methods to disable and enable now ensure we have a green cluster and
that the cluster is consistent before returning.

Closes #30301

Author: jaymode

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/action/filter/SecurityActionFilter.java

@@ -154,7 +154,15 @@ it to the action without an associated user (not via REST or transport - this is
          */
         final String securityAction = actionMapper.action(action, request);
         authcService.authenticate(securityAction, request, SystemUser.INSTANCE,
-                ActionListener.wrap((authc) -> authorizeRequest(authc, securityAction, request, listener), listener::onFailure));
+                ActionListener.wrap((authc) -> {
+                    if (authc != null) {
+                        authorizeRequest(authc, securityAction, request, listener);
+                    } else if (licenseState.isAuthAllowed() == false) {
+                        listener.onResponse(null);
+                    } else {
+                        listener.onFailure(new IllegalStateException("no authentication present but auth is allowed"));
+                    }
+                }, listener::onFailure));
     }
 
     private <Request extends ActionRequest> void authorizeRequest(Authentication authentication, String securityAction, Request request,

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/action/saml/TransportSamlAuthenticateAction.java

@@ -60,6 +60,7 @@ protected void doExecute(SamlAuthenticateRequest request,
                     listener.onFailure(new IllegalStateException("Cannot find AuthenticationResult on thread context"));
                     return;
                 }
+                assert authentication != null : "authentication should never be null at this point";
                 final Map<String, Object> tokenMeta = (Map<String, Object>) result.getMetadata().get(SamlRealm.CONTEXT_TOKEN_DATA);
                 tokenService.createUserToken(authentication, originatingAuthentication,
                         ActionListener.wrap(tuple -> {

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/action/token/TransportCreateTokenAction.java

@@ -73,7 +73,11 @@ private void authenticateAndCreateToken(CreateTokenRequest request, ActionListen
             authenticationService.authenticate(CreateTokenAction.NAME, request, authToken,
                 ActionListener.wrap(authentication -> {
                     request.getPassword().close();
-                    createToken(request, authentication, originatingAuthentication, true, listener);
+                    if (authentication != null) {
+                        createToken(request, authentication, originatingAuthentication, true, listener);
+                    } else {
+                        listener.onFailure(new UnsupportedOperationException("cannot create token if authentication is not allowed"));
+                    }
                 }, e -> {
                     // clear the request password
                     request.getPassword().close();

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authc/AuthenticationService.java

@@ -135,8 +135,9 @@ class Authenticator {
 
         private final AuditableRequest request;
         private final User fallbackUser;
-
+        private final List<Realm> defaultOrderedRealmList;
         private final ActionListener<Authentication> listener;
+
         private RealmRef authenticatedBy = null;
         private RealmRef lookedupBy = null;
         private AuthenticationToken authenticationToken = null;
@@ -154,6 +155,7 @@ class Authenticator {
         private Authenticator(AuditableRequest auditableRequest, User fallbackUser, ActionListener<Authentication> listener) {
             this.request = auditableRequest;
             this.fallbackUser = fallbackUser;
+            this.defaultOrderedRealmList = realms.asList();
             this.listener = listener;
         }
 
@@ -172,27 +174,33 @@ private Authenticator(AuditableRequest auditableRequest, User fallbackUser, Acti
          * </ol>
          */
         private void authenticateAsync() {
-            lookForExistingAuthentication((authentication) -> {
-                if (authentication != null) {
-                    listener.onResponse(authentication);
-                } else {
-                    tokenService.getAndValidateToken(threadContext, ActionListener.wrap(userToken -> {
-                        if (userToken != null) {
-                            writeAuthToContext(userToken.getAuthentication());
-                        } else {
-                            extractToken(this::consumeToken);
-                        }
-                    }, e -> {
-                        if (e instanceof ElasticsearchSecurityException &&
+            if (defaultOrderedRealmList.isEmpty()) {
+                // this happens when the license state changes between the call to authenticate and the actual invocation
+                // to get the realm list
+                listener.onResponse(null);
+            } else {
+                lookForExistingAuthentication((authentication) -> {
+                    if (authentication != null) {
+                        listener.onResponse(authentication);
+                    } else {
+                        tokenService.getAndValidateToken(threadContext, ActionListener.wrap(userToken -> {
+                            if (userToken != null) {
+                                writeAuthToContext(userToken.getAuthentication());
+                            } else {
+                                extractToken(this::consumeToken);
+                            }
+                        }, e -> {
+                            if (e instanceof ElasticsearchSecurityException &&
                                 tokenService.isExpiredTokenException((ElasticsearchSecurityException) e) == false) {
-                            // intentionally ignore the returned exception; we call this primarily
-                            // for the auditing as we already have a purpose built exception
-                            request.tamperedRequest();
-                        }
-                        listener.onFailure(e);
-                    }));
-                }
-            });
+                                // intentionally ignore the returned exception; we call this primarily
+                                // for the auditing as we already have a purpose built exception
+                                request.tamperedRequest();
+                            }
+                            listener.onFailure(e);
+                        }));
+                    }
+                });
+            }
         }
 
         /**
@@ -233,7 +241,7 @@ void extractToken(Consumer<AuthenticationToken> consumer) {
                 if (authenticationToken != null) {
                     action = () -> consumer.accept(authenticationToken);
                 } else {
-                    for (Realm realm : realms) {
+                    for (Realm realm : defaultOrderedRealmList) {
                         final AuthenticationToken token = realm.token(threadContext);
                         if (token != null) {
                             action = () -> consumer.accept(token);
@@ -260,7 +268,6 @@ private void consumeToken(AuthenticationToken token) {
                 handleNullToken();
             } else {
                 authenticationToken = token;
-                final List<Realm> realmsList = realms.asList();
                 final Map<Realm, Tuple<String, Exception>> messages = new LinkedHashMap<>();
                 final BiConsumer<Realm, ActionListener<User>> realmAuthenticatingConsumer = (realm, userListener) -> {
                     if (realm.supports(authenticationToken)) {
@@ -297,11 +304,12 @@ private void consumeToken(AuthenticationToken token) {
                         userListener.onResponse(null);
                     }
                 };
+
                 final IteratingActionListener<User, Realm> authenticatingListener =
                     new IteratingActionListener<>(ContextPreservingActionListener.wrapPreservingContext(ActionListener.wrap(
                         (user) -> consumeUser(user, messages),
                         (e) -> listener.onFailure(request.exceptionProcessingRequest(e, token))), threadContext),
-                        realmAuthenticatingConsumer, realmsList, threadContext);
+                        realmAuthenticatingConsumer, defaultOrderedRealmList, threadContext);
                 try {
                     authenticatingListener.run();
                 } catch (Exception e) {
@@ -388,7 +396,7 @@ private void consumeUser(User user, Map<Realm, Tuple<String, Exception>> message
          * names of users that exist using a timing attack
          */
         private void lookupRunAsUser(final User user, String runAsUsername, Consumer<User> userConsumer) {
-            final RealmUserLookup lookup = new RealmUserLookup(realms.asList(), threadContext);
+            final RealmUserLookup lookup = new RealmUserLookup(defaultOrderedRealmList, threadContext);
             lookup.lookup(runAsUsername, ActionListener.wrap(tuple -> {
                 if (tuple == null) {
                     // the user does not exist, but we still create a User object, which will later be rejected by authz

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptor.java

@@ -187,12 +187,12 @@ protected Map<String, ServerTransportFilter> initializeProfileFilters(Destructiv
                 case "client":
                     profileFilters.put(entry.getKey(), new ServerTransportFilter.ClientProfile(authcService, authzService,
                             threadPool.getThreadContext(), extractClientCert, destructiveOperations, reservedRealmEnabled,
-                            securityContext));
+                            securityContext, licenseState));
                     break;
                 case "node":
                     profileFilters.put(entry.getKey(), new ServerTransportFilter.NodeProfile(authcService, authzService,
                             threadPool.getThreadContext(), extractClientCert, destructiveOperations, reservedRealmEnabled,
-                            securityContext));
+                            securityContext, licenseState));
                     break;
                 default:
                    throw new IllegalStateException("unknown profile type: " + type);

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/transport/ServerTransportFilter.java

@@ -19,6 +19,7 @@
 import org.elasticsearch.action.admin.indices.open.OpenIndexAction;
 import org.elasticsearch.action.support.DestructiveOperations;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.transport.TaskTransportChannel;
 import org.elasticsearch.transport.TcpTransportChannel;
 import org.elasticsearch.transport.TransportChannel;
@@ -78,17 +79,19 @@ class NodeProfile implements ServerTransportFilter {
         private final DestructiveOperations destructiveOperations;
         private final boolean reservedRealmEnabled;
         private final SecurityContext securityContext;
+        private final XPackLicenseState licenseState;
 
         NodeProfile(AuthenticationService authcService, AuthorizationService authzService,
                     ThreadContext threadContext, boolean extractClientCert, DestructiveOperations destructiveOperations,
-                    boolean reservedRealmEnabled, SecurityContext securityContext) {
+                    boolean reservedRealmEnabled, SecurityContext securityContext, XPackLicenseState licenseState) {
             this.authcService = authcService;
             this.authzService = authzService;
             this.threadContext = threadContext;
             this.extractClientCert = extractClientCert;
             this.destructiveOperations = destructiveOperations;
             this.reservedRealmEnabled = reservedRealmEnabled;
             this.securityContext = securityContext;
+            this.licenseState = licenseState;
         }
 
         @Override
@@ -129,6 +132,7 @@ requests from all the nodes are attached with a user (either a serialize
 
             final Version version = transportChannel.getVersion().equals(Version.V_5_4_0) ? Version.CURRENT : transportChannel.getVersion();
             authcService.authenticate(securityAction, request, (User)null, ActionListener.wrap((authentication) -> {
+                if (authentication != null) {
                     if (reservedRealmEnabled && authentication.getVersion().before(Version.V_5_2_0) &&
                         KibanaUser.NAME.equals(authentication.getUser().authenticatedUser().principal())) {
                         executeAsCurrentVersionKibanaUser(securityAction, request, transportChannel, listener, authentication);
@@ -156,7 +160,12 @@ requests from all the nodes are attached with a user (either a serialize
                                 });
                         asyncAuthorizer.authorize(authzService);
                     }
-                }, listener::onFailure));
+                } else if (licenseState.isAuthAllowed() == false) {
+                    listener.onResponse(null);
+                } else {
+                    listener.onFailure(new IllegalStateException("no authentication present but auth is allowed"));
+                }
+            }, listener::onFailure));
         }
 
         private void executeAsCurrentVersionKibanaUser(String securityAction, TransportRequest request, TransportChannel transportChannel,
@@ -220,9 +229,9 @@ class ClientProfile extends NodeProfile {
 
         ClientProfile(AuthenticationService authcService, AuthorizationService authzService,
                              ThreadContext threadContext, boolean extractClientCert, DestructiveOperations destructiveOperations,
-                             boolean reservedRealmEnabled, SecurityContext securityContext) {
+                             boolean reservedRealmEnabled, SecurityContext securityContext, XPackLicenseState licenseState) {
             super(authcService, authzService, threadContext, extractClientCert, destructiveOperations, reservedRealmEnabled,
-                    securityContext);
+                securityContext, licenseState);
         }
 
         @Override