Merge branch 'master' into ccr

* master:
  Integrates soft-deletes into Elasticsearch (#33222)
  Revert "Integrates soft-deletes into Elasticsearch (#33222)"
  Add support for "authorization_realms" (#33262)

docs/reference/settings/security-settings.asciidoc

@@ -246,6 +246,13 @@ This setting is multivalued; you can specify multiple user contexts.
 Required to operate in user template mode. If `user_search.base_dn` is specified, 
 this setting is not valid. For more information on
 the different modes, see {xpack-ref}/ldap-realm.html[LDAP realms].
+
+`authorization_realms`::
+The names of the realms that should be consulted for delegate authorization.
+If this setting is used, then the LDAP realm does not perform role mapping and
+instead loads the user from the listed realms. The referenced realms are
+consulted in the order that they are defined in this list.
+See {stack-ov}/realm-chains.html#authorization_realms[Delegating authorization to another realm] 
 +
 --
 NOTE: If any settings starting with `user_search` are specified, the 
@@ -733,6 +740,12 @@ Specifies the {xpack-ref}/security-files.html[location] of the
 {xpack-ref}/mapping-roles.html[YAML role  mapping configuration file].
 Defaults to `ES_PATH_CONF/role_mapping.yml`.
 
+`authorization_realms`::
+The names of the realms that should be consulted for delegate authorization.
+If this setting is used, then the PKI realm does not perform role mapping and
+instead loads the user from the listed realms.
+See {stack-ov}/realm-chains.html#authorization_realms[Delegating authorization to another realm] 
+
 `cache.ttl`::
 Specifies the time-to-live for cached user entries. A user and a hash of its 
 credentials are cached for this period of time. Use the
@@ -856,6 +869,12 @@ Defaults to `false`.
 Specifies whether to populate the {es} user's metadata with the values that are 
 provided by the SAML attributes. Defaults to `true`.
 
+`authorization_realms`::
+The names of the realms that should be consulted for delegate authorization.
+If this setting is used, then the SAML realm does not perform role mapping and
+instead loads the user from the listed realms.
+See {stack-ov}/realm-chains.html#authorization_realms[Delegating authorization to another realm] 
+
 `allowed_clock_skew`::
 The maximum amount of skew that can be tolerated between the IdP's clock and the
 {es} node's clock.

x-pack/docs/en/security/authentication/configuring-kerberos-realm.asciidoc

@@ -166,5 +166,10 @@ POST _xpack/security/role_mapping/kerbrolemapping
 // CONSOLE
 
 For more information, see {stack-ov}/mapping-roles.html[Mapping users and groups to roles].
+
+NOTE: The Kerberos realm supports
+{stack-ov}/realm-chains.html#authorization_realms[authorization realms] as an
+alternative to role mapping.
+
 --
 

x-pack/docs/en/security/authentication/configuring-ldap-realm.asciidoc

@@ -189,6 +189,11 @@ For more information, see
 {xpack-ref}/ldap-realm.html#mapping-roles-ldap[Mapping LDAP Groups to Roles] 
 and 
 {xpack-ref}/mapping-roles.html[Mapping Users and Groups to Roles].
+
+NOTE: The LDAP realm supports
+{stack-ov}/realm-chains.html#authorization_realms[authorization realms] as an
+alternative to role mapping.
+
 --
 
 . (Optional) Configure the `metadata` setting on the LDAP realm to include extra 

x-pack/docs/en/security/authentication/configuring-pki-realm.asciidoc

@@ -10,7 +10,8 @@ NOTE: You cannot use PKI certificates to authenticate users in {kib}.
 
 To use PKI in {es}, you configure a PKI realm, enable client authentication on
 the desired network layers (transport or http), and map the Distinguished Names
-(DNs) from the user certificates to {security} roles in the role mapping file.
+(DNs) from the user certificates to {security} roles in the 
+<<security-api-role-mapping,role-mapping API>> or role-mapping file.
 
 You can also use a combination of PKI and username/password authentication. For
 example, you can enable SSL/TLS on the transport layer and define a PKI realm to
@@ -173,4 +174,9 @@ key. You can also use the authenticate API to validate your role mapping.
 
 For more information, see 
 {xpack-ref}/mapping-roles.html[Mapping Users and Groups to Roles].
+
+NOTE: The PKI realm supports
+{stack-ov}/realm-chains.html#authorization_realms[authorization realms] as an
+alternative to role mapping.
+
 --

x-pack/docs/en/security/authentication/configuring-saml-realm.asciidoc

@@ -219,6 +219,11 @@ access any data.
 
 Your SAML users cannot do anything until they are mapped to {security}
 roles. See {stack-ov}/saml-role-mapping.html[Configuring role mappings]. 
+
+NOTE: The SAML realm supports
+{stack-ov}/realm-chains.html#authorization_realms[authorization realms] as an
+alternative to role mapping.
+
 --
 
 . {stack-ov}/saml-kibana.html[Configure {kib} to use SAML SSO]. 

x-pack/docs/en/security/authentication/saml-guide.asciidoc

@@ -473,7 +473,7 @@ or separate keys used for each of those.
 
 The Elastic Stack uses X.509 certificates with RSA private keys for SAML
 cryptography. These keys can be generated using any standard SSL tool, including
-the `elasticsearch-certutil` tool that ships with X-Pack.
+the `elasticsearch-certutil` tool that ships with {xpack}.
 
 Your IdP may require that the Elastic Stack have a cryptographic key for signing
 SAML messages, and that you provide the corresponding signing certificate within
@@ -624,9 +624,10 @@ When a user authenticates using SAML, they are identified to the Elastic Stack,
 but this does not automatically grant them access to perform any actions or
 access any data.
 
-Your SAML users cannot do anything until they are mapped to {security}
+Your SAML users cannot do anything until they are assigned {security}
-roles. This mapping is performed through the
+roles. This is done through either the
-{ref}/security-api-put-role-mapping.html[add role mapping API].
+{ref}/security-api-put-role-mapping.html[add role mapping API], or with
+<<authorization_realms, authorization realms>>.
 
 This is an example of a simple role mapping that grants the `kibana_user` role
 to any user who authenticates against the `saml1` realm:
@@ -683,6 +684,18 @@ PUT /_xpack/security/role_mapping/saml-finance
 // CONSOLE
 // TEST
 
+If your users also exist in a repository that can be directly accessed by {security}
+(such as an LDAP directory) then you can use
+<<authorization_realms, authorization realms>> instead of role mappings.
+
+In this case, you perform the following steps:
+1. In your SAML realm, assigned a SAML attribute to act as the lookup userid,
+   by configuring the `attributes.principal` setting.
+2. Create a new realm that can lookup users from your local repository (e.g. an
+   `ldap` realm)
+3. In your SAML realm, set `authorization_realms` to the name of the realm you
+   created in step 2.
+
 [[saml-user-metadata]]
 === User metadata
 

x-pack/docs/en/security/authorization/mapping-roles.asciidoc

@@ -24,6 +24,9 @@ either role management method. For example, when you use the role mapping API,
 you are able to map users to both API-managed roles and file-managed roles
 (and likewise for file-based role-mappings).
 
+NOTE: The PKI, LDAP, Kerberos and SAML realms support using
+<<authorization_realms, authorization realms>> as an alternative to role mapping.
+
 [[mapping-roles-api]]
 ==== Using the role mapping API
 

x-pack/docs/en/security/authorization/run-as-privilege.asciidoc

@@ -12,7 +12,7 @@ the realm you use to authenticate. Both the internal `native` and `file` realms
 support this out of the box. The LDAP realm must be configured to run in
 <<ldap-user-search, _user search_ mode>>. The Active Directory realm must be
 <<ad-settings,configured with a `bind_dn` and `secure_bind_password`>> to support
-_run as_. The PKI realm does not support _run as_.
+_run as_. The PKI, Kerberos, and SAML realms do not support _run as_.
 
 To submit requests on behalf of other users, you need to have the `run_as`
 permission. For example, the following role grants permission to submit request

x-pack/plugin/core/src/main/java/org/elasticsearch/license/XPackLicenseState.java

@@ -410,10 +410,20 @@ public class XPackLicenseState {
      */
     public boolean isCustomRoleProvidersAllowed() {
         final Status localStatus = status;
-        return (localStatus.mode == OperationMode.PLATINUM || localStatus.mode == OperationMode.TRIAL )
+        return (localStatus.mode == OperationMode.PLATINUM || localStatus.mode == OperationMode.TRIAL)
                 && localStatus.active;
     }
 
+    /**
+     * @return whether "authorization_realms" are allowed based on the license {@link OperationMode}
+     * @see org.elasticsearch.xpack.core.security.authc.support.DelegatedAuthorizationSettings
+     */
+    public boolean isAuthorizationRealmAllowed() {
+        final Status localStatus = status;
+        return (localStatus.mode == OperationMode.PLATINUM || localStatus.mode == OperationMode.TRIAL)
+            && localStatus.active;
+    }
+
     /**
      * Determine if Watcher is available based on the current license.
      * <p>

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

@@ -8,6 +8,8 @@ package org.elasticsearch.xpack.core.security.authc;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.xpack.core.security.authc.support.DelegatedAuthorizationSettings;
 import org.elasticsearch.xpack.core.XPackField;
 import org.elasticsearch.xpack.core.security.user.User;
 
@@ -146,6 +148,14 @@ public abstract class Realm implements Comparable<Realm> {
         return type + "/" + config.name;
     }
 
+    /**
+     * This is no-op in the base class, but allows realms to be aware of what other realms are configured
+     *
+     * @see DelegatedAuthorizationSettings
+     */
+    public void initialize(Iterable<Realm> realms, XPackLicenseState licenseState) {
+    }
+
     /**
      * A factory interface to construct a security realm.
      */

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authc/kerberos/KerberosRealmSettings.java

@@ -10,6 +10,7 @@ import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.set.Sets;
+import org.elasticsearch.xpack.core.security.authc.support.DelegatedAuthorizationSettings;
 
 import java.util.Set;
 
@@ -44,7 +45,9 @@ public final class KerberosRealmSettings {
      * @return the valid set of {@link Setting}s for a {@value #TYPE} realm
      */
     public static Set<Setting<?>> getSettings() {
-        return Sets.newHashSet(HTTP_SERVICE_KEYTAB_PATH, CACHE_TTL_SETTING, CACHE_MAX_USERS_SETTING, SETTING_KRB_DEBUG_ENABLE,
+        final Set<Setting<?>> settings = Sets.newHashSet(HTTP_SERVICE_KEYTAB_PATH, CACHE_TTL_SETTING, CACHE_MAX_USERS_SETTING,
-                SETTING_REMOVE_REALM_NAME);
+                SETTING_KRB_DEBUG_ENABLE, SETTING_REMOVE_REALM_NAME);
+        settings.addAll(DelegatedAuthorizationSettings.getSettings());
+        return settings;
     }
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authc/ldap/LdapRealmSettings.java

@@ -9,6 +9,7 @@ import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.xpack.core.security.authc.ldap.support.LdapMetaDataResolverSettings;
 import org.elasticsearch.xpack.core.security.authc.support.CachingUsernamePasswordRealmSettings;
+import org.elasticsearch.xpack.core.security.authc.support.DelegatedAuthorizationSettings;
 import org.elasticsearch.xpack.core.security.authc.support.mapper.CompositeRoleMapperSettings;
 
 import java.util.HashSet;
@@ -37,6 +38,7 @@ public final class LdapRealmSettings {
             assert LDAP_TYPE.equals(type) : "type [" + type + "] is unknown. expected one of [" + AD_TYPE + ", " + LDAP_TYPE + "]";
             settings.addAll(LdapSessionFactorySettings.getSettings());
             settings.addAll(LdapUserSearchSessionFactorySettings.getSettings());
+            settings.addAll(DelegatedAuthorizationSettings.getSettings());
         }
         settings.addAll(LdapMetaDataResolverSettings.getSettings());
         return settings;

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authc/pki/PkiRealmSettings.java

@@ -7,6 +7,7 @@ package org.elasticsearch.xpack.core.security.authc.pki;
 
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.xpack.core.security.authc.support.DelegatedAuthorizationSettings;
 import org.elasticsearch.xpack.core.security.authc.support.mapper.CompositeRoleMapperSettings;
 import org.elasticsearch.xpack.core.ssl.SSLConfigurationSettings;
 
@@ -43,6 +44,7 @@ public final class PkiRealmSettings {
         settings.add(SSL_SETTINGS.truststoreAlgorithm);
         settings.add(SSL_SETTINGS.caPaths);
 
+        settings.addAll(DelegatedAuthorizationSettings.getSettings());
         settings.addAll(CompositeRoleMapperSettings.getSettings());
 
         return settings;

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authc/saml/SamlRealmSettings.java

@@ -8,6 +8,7 @@ package org.elasticsearch.xpack.core.security.authc.saml;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.set.Sets;
+import org.elasticsearch.xpack.core.security.authc.support.DelegatedAuthorizationSettings;
 import org.elasticsearch.xpack.core.ssl.SSLConfigurationSettings;
 import org.elasticsearch.xpack.core.ssl.X509KeyPairSettings;
 
@@ -89,6 +90,7 @@ public class SamlRealmSettings {
         set.addAll(DN_ATTRIBUTE.settings());
         set.addAll(NAME_ATTRIBUTE.settings());
         set.addAll(MAIL_ATTRIBUTE.settings());
+        set.addAll(DelegatedAuthorizationSettings.getSettings());
         return set;
     }
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authc/support/DelegatedAuthorizationSettings.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.core.security.authc.support;
+
+import org.elasticsearch.common.settings.Setting;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Function;
+
+/**
+ * Settings related to "Delegated Authorization" (aka Lookup Realms)
+ */
+public class DelegatedAuthorizationSettings {
+
+    public static final Setting<List<String>> AUTHZ_REALMS = Setting.listSetting("authorization_realms",
+        Collections.emptyList(), Function.identity(), Setting.Property.NodeScope);
+
+    public static Collection<Setting<?>> getSettings() {
+        return Collections.singleton(AUTHZ_REALMS);
+    }
+}

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

@@ -34,10 +34,12 @@ import org.elasticsearch.xpack.core.security.user.SystemUser;
 import org.elasticsearch.xpack.core.security.user.User;
 import org.elasticsearch.xpack.security.audit.AuditTrail;
 import org.elasticsearch.xpack.security.audit.AuditTrailService;
+import org.elasticsearch.xpack.security.authc.support.RealmUserLookup;
 
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
@@ -381,33 +383,18 @@ public class AuthenticationService extends AbstractComponent {
          * names of users that exist using a timing attack
          */
         private void lookupRunAsUser(final User user, String runAsUsername, Consumer<User> userConsumer) {
-            final List<Realm> realmsList = realms.asList();
+            final RealmUserLookup lookup = new RealmUserLookup(realms.asList(), threadContext);
-            final BiConsumer<Realm, ActionListener<User>> realmLookupConsumer = (realm, lookupUserListener) ->
+            lookup.lookup(runAsUsername, ActionListener.wrap(tuple -> {
-                    realm.lookupUser(runAsUsername, ActionListener.wrap((lookedupUser) -> {
+                if (tuple == null) {
-                        if (lookedupUser != null) {
+                    // the user does not exist, but we still create a User object, which will later be rejected by authz
-                            lookedupBy = new RealmRef(realm.name(), realm.type(), nodeName);
+                    userConsumer.accept(new User(runAsUsername, null, user));
-                            lookupUserListener.onResponse(lookedupUser);
+                } else {
-                        } else {
+                    User foundUser = Objects.requireNonNull(tuple.v1());
-                            lookupUserListener.onResponse(null);
+                    Realm realm = Objects.requireNonNull(tuple.v2());
-                        }
+                    lookedupBy = new RealmRef(realm.name(), realm.type(), nodeName);
-                    }, lookupUserListener::onFailure));
+                    userConsumer.accept(new User(foundUser, user));
-
+                }
-            final IteratingActionListener<User, Realm> userLookupListener =
+            }, exception -> listener.onFailure(request.exceptionProcessingRequest(exception, authenticationToken))));
-                    new IteratingActionListener<>(ActionListener.wrap((lookupUser) -> {
-                                if (lookupUser == null) {
-                                    // the user does not exist, but we still create a User object, which will later be rejected by authz
-                                    userConsumer.accept(new User(runAsUsername, null, user));
-                                } else {
-                                    userConsumer.accept(new User(lookupUser, user));
-                                }
-                            },
-                            (e) -> listener.onFailure(request.exceptionProcessingRequest(e, authenticationToken))),
-                            realmLookupConsumer, realmsList, threadContext);
-            try {
-                userLookupListener.run();
-            } catch (Exception e) {
-                listener.onFailure(request.exceptionProcessingRequest(e, authenticationToken));
-            }
         }
 
         /**

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

@@ -93,6 +93,7 @@ public class Realms extends AbstractComponent implements Iterable<Realm> {
 
         this.standardRealmsOnly = Collections.unmodifiableList(standardRealms);
         this.nativeRealmsOnly = Collections.unmodifiableList(nativeRealms);
+        realms.forEach(r -> r.initialize(this, licenseState));
     }
 
     @Override

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

@@ -13,6 +13,7 @@ import org.elasticsearch.common.cache.Cache;
 import org.elasticsearch.common.cache.CacheBuilder;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationResult;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationToken;
@@ -21,6 +22,7 @@ import org.elasticsearch.xpack.core.security.authc.RealmConfig;
 import org.elasticsearch.xpack.core.security.authc.kerberos.KerberosRealmSettings;
 import org.elasticsearch.xpack.core.security.user.User;
 import org.elasticsearch.xpack.security.authc.support.CachingRealm;
+import org.elasticsearch.xpack.security.authc.support.DelegatedAuthorizationSupport;
 import org.elasticsearch.xpack.security.authc.support.UserRoleMapper;
 import org.elasticsearch.xpack.security.authc.support.mapper.NativeRoleMappingStore;
 import org.ietf.jgss.GSSException;
@@ -63,6 +65,7 @@ public final class KerberosRealm extends Realm implements CachingRealm {
     private final Path keytabPath;
     private final boolean enableKerberosDebug;
     private final boolean removeRealmName;
+    private DelegatedAuthorizationSupport delegatedRealms;
 
     public KerberosRealm(final RealmConfig config, final NativeRoleMappingStore nativeRoleMappingStore, final ThreadPool threadPool) {
         this(config, nativeRoleMappingStore, new KerberosTicketValidator(), threadPool, null);
@@ -100,6 +103,15 @@ public final class KerberosRealm extends Realm implements CachingRealm {
         }
         this.enableKerberosDebug = KerberosRealmSettings.SETTING_KRB_DEBUG_ENABLE.get(config.settings());
         this.removeRealmName = KerberosRealmSettings.SETTING_REMOVE_REALM_NAME.get(config.settings());
+        this.delegatedRealms = null;
+    }
+
+    @Override
+    public void initialize(Iterable<Realm> realms, XPackLicenseState licenseState) {
+        if (delegatedRealms != null) {
+            throw new IllegalStateException("Realm has already been initialized");
+        }
+        delegatedRealms = new DelegatedAuthorizationSupport(realms, config, licenseState);
     }
 
     @Override
@@ -133,13 +145,14 @@ public final class KerberosRealm extends Realm implements CachingRealm {
 
     @Override
     public void authenticate(final AuthenticationToken token, final ActionListener<AuthenticationResult> listener) {
+        assert delegatedRealms != null : "Realm has not been initialized correctly";
         assert token instanceof KerberosAuthenticationToken;
         final KerberosAuthenticationToken kerbAuthnToken = (KerberosAuthenticationToken) token;
         kerberosTicketValidator.validateTicket((byte[]) kerbAuthnToken.credentials(), keytabPath, enableKerberosDebug,
                 ActionListener.wrap(userPrincipalNameOutToken -> {
                     if (userPrincipalNameOutToken.v1() != null) {
                         final String username = maybeRemoveRealmName(userPrincipalNameOutToken.v1());
-                        buildUser(username, userPrincipalNameOutToken.v2(), listener);
+                        resolveUser(username, userPrincipalNameOutToken.v2(), listener);
                     } else {
                         /**
                          * This is when security context could not be established may be due to ongoing
@@ -192,35 +205,36 @@ public final class KerberosRealm extends Realm implements CachingRealm {
         }
     }
 
-    private void buildUser(final String username, final String outToken, final ActionListener<AuthenticationResult> listener) {
+    private void resolveUser(final String username, final String outToken, final ActionListener<AuthenticationResult> listener) {
         // if outToken is present then it needs to be communicated with peer, add it to
         // response header in thread context.
         if (Strings.hasText(outToken)) {
             threadPool.getThreadContext().addResponseHeader(WWW_AUTHENTICATE, NEGOTIATE_AUTH_HEADER_PREFIX + outToken);
         }
-        final User user = (userPrincipalNameToUserCache != null) ? userPrincipalNameToUserCache.get(username) : null;
+
-        if (user != null) {
+        if (delegatedRealms.hasDelegation()) {
-            /**
+            delegatedRealms.resolve(username, listener);
-             * TODO: bizybot If authorizing realms configured, resolve user from those
-             * realms and then return.
-             */
-            listener.onResponse(AuthenticationResult.success(user));
         } else {
-            /**
+            final User user = (userPrincipalNameToUserCache != null) ? userPrincipalNameToUserCache.get(username) : null;
-             * TODO: bizybot If authorizing realms configured, resolve user from those
+            if (user != null) {
-             * realms, cache it and then return.
+                listener.onResponse(AuthenticationResult.success(user));
-             */
+            } else {
-            final UserRoleMapper.UserData userData = new UserRoleMapper.UserData(username, null, Collections.emptySet(), null, this.config);
+                buildUser(username, listener);
-            userRoleMapper.resolveRoles(userData, ActionListener.wrap(roles -> {
+            }
-                final User computedUser = new User(username, roles.toArray(new String[roles.size()]), null, null, null, true);
-                if (userPrincipalNameToUserCache != null) {
-                    userPrincipalNameToUserCache.put(username, computedUser);
-                }
-                listener.onResponse(AuthenticationResult.success(computedUser));
-            }, listener::onFailure));
         }
     }
 
+    private void buildUser(final String username, final ActionListener<AuthenticationResult> listener) {
+        final UserRoleMapper.UserData userData = new UserRoleMapper.UserData(username, null, Collections.emptySet(), null, this.config);
+        userRoleMapper.resolveRoles(userData, ActionListener.wrap(roles -> {
+            final User computedUser = new User(username, roles.toArray(new String[roles.size()]), null, null, null, true);
+            if (userPrincipalNameToUserCache != null) {
+                userPrincipalNameToUserCache.put(username, computedUser);
+            }
+            listener.onResponse(AuthenticationResult.success(computedUser));
+        }, listener::onFailure));
+    }
+
     @Override
     public void lookupUser(final String username, final ActionListener<User> listener) {
         listener.onResponse(null);

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

@@ -8,7 +8,6 @@ package org.elasticsearch.xpack.security.authc.ldap;
 import com.unboundid.ldap.sdk.LDAPException;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
-import org.elasticsearch.core.internal.io.IOUtils;
 import org.elasticsearch.ElasticsearchTimeoutException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.ContextPreservingActionListener;
@@ -16,10 +15,13 @@ import org.elasticsearch.common.collect.MapBuilder;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.core.internal.io.IOUtils;
+import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.threadpool.ThreadPool.Names;
 import org.elasticsearch.watcher.ResourceWatcherService;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationResult;
+import org.elasticsearch.xpack.core.security.authc.Realm;
 import org.elasticsearch.xpack.core.security.authc.RealmConfig;
 import org.elasticsearch.xpack.core.security.authc.RealmSettings;
 import org.elasticsearch.xpack.core.security.authc.ldap.LdapRealmSettings;
@@ -31,6 +33,7 @@ import org.elasticsearch.xpack.security.authc.ldap.support.LdapLoadBalancing;
 import org.elasticsearch.xpack.security.authc.ldap.support.LdapSession;
 import org.elasticsearch.xpack.security.authc.ldap.support.SessionFactory;
 import org.elasticsearch.xpack.security.authc.support.CachingUsernamePasswordRealm;
+import org.elasticsearch.xpack.security.authc.support.DelegatedAuthorizationSupport;
 import org.elasticsearch.xpack.security.authc.support.UserRoleMapper;
 import org.elasticsearch.xpack.security.authc.support.UserRoleMapper.UserData;
 import org.elasticsearch.xpack.security.authc.support.mapper.CompositeRoleMapper;
@@ -53,7 +56,7 @@ public final class LdapRealm extends CachingUsernamePasswordRealm {
     private final UserRoleMapper roleMapper;
     private final ThreadPool threadPool;
     private final TimeValue executionTimeout;
-
+    private DelegatedAuthorizationSupport delegatedRealms;
 
     public LdapRealm(String type, RealmConfig config, SSLService sslService,
                      ResourceWatcherService watcherService,
@@ -118,6 +121,7 @@ public final class LdapRealm extends CachingUsernamePasswordRealm {
      */
     @Override
     protected void doAuthenticate(UsernamePasswordToken token, ActionListener<AuthenticationResult> listener) {
+        assert delegatedRealms != null : "Realm has not been initialized correctly";
         // we submit to the threadpool because authentication using LDAP will execute blocking I/O for a bind request and we don't want
         // network threads stuck waiting for a socket to connect. After the bind, then all interaction with LDAP should be async
         final CancellableLdapRunnable<AuthenticationResult> cancellableLdapRunnable = new CancellableLdapRunnable<>(listener,
@@ -159,6 +163,14 @@ public final class LdapRealm extends CachingUsernamePasswordRealm {
                 sessionListener);
     }
 
+    @Override
+    public void initialize(Iterable<Realm> realms, XPackLicenseState licenseState) {
+        if (delegatedRealms != null) {
+            throw new IllegalStateException("Realm has already been initialized");
+        }
+        delegatedRealms = new DelegatedAuthorizationSupport(realms, config, licenseState);
+    }
+
     @Override
     public void usageStats(ActionListener<Map<String, Object>> listener) {
         super.usageStats(ActionListener.wrap(usage -> {
@@ -171,39 +183,56 @@ public final class LdapRealm extends CachingUsernamePasswordRealm {
     }
 
     private static void buildUser(LdapSession session, String username, ActionListener<AuthenticationResult> listener,
-                                  UserRoleMapper roleMapper) {
+                                  UserRoleMapper roleMapper, DelegatedAuthorizationSupport delegatedAuthz) {
+        assert delegatedAuthz != null : "DelegatedAuthorizationSupport is null";
         if (session == null) {
             listener.onResponse(AuthenticationResult.notHandled());
+        } else if (delegatedAuthz.hasDelegation()) {
+            delegatedAuthz.resolve(username, listener);
         } else {
-            boolean loadingGroups = false;
+            lookupUserFromSession(username, session, roleMapper, listener);
-            try {
+        }
-                final Consumer<Exception> onFailure = e -> {
+    }
-                    IOUtils.closeWhileHandlingException(session);
+
-                    listener.onFailure(e);
+    @Override
-                };
+    protected void handleCachedAuthentication(User user, ActionListener<AuthenticationResult> listener) {
-                session.resolve(ActionListener.wrap((ldapData) -> {
+        if (delegatedRealms.hasDelegation()) {
-                    final Map<String, Object> metadata = MapBuilder.<String, Object>newMapBuilder()
+            delegatedRealms.resolve(user.principal(), listener);
-                            .put("ldap_dn", session.userDn())
+        } else {
-                            .put("ldap_groups", ldapData.groups)
+            super.handleCachedAuthentication(user, listener);
-                            .putAll(ldapData.metaData)
+        }
-                            .map();
+    }
-                    final UserData user = new UserData(username, session.userDn(), ldapData.groups,
+
-                            metadata, session.realm());
+    private static void lookupUserFromSession(String username, LdapSession session, UserRoleMapper roleMapper,
-                    roleMapper.resolveRoles(user, ActionListener.wrap(
+                                              ActionListener<AuthenticationResult> listener) {
-                            roles -> {
+        boolean loadingGroups = false;
-                                IOUtils.close(session);
+        try {
-                                String[] rolesArray = roles.toArray(new String[roles.size()]);
+            final Consumer<Exception> onFailure = e -> {
-                                listener.onResponse(AuthenticationResult.success(
+                IOUtils.closeWhileHandlingException(session);
-                                        new User(username, rolesArray, null, null, metadata, true))
+                listener.onFailure(e);
-                                );
+            };
-                            }, onFailure
+            session.resolve(ActionListener.wrap((ldapData) -> {
-                    ));
+                final Map<String, Object> metadata = MapBuilder.<String, Object>newMapBuilder()
-                }, onFailure));
+                    .put("ldap_dn", session.userDn())
-                loadingGroups = true;
+                    .put("ldap_groups", ldapData.groups)
-            } finally {
+                    .putAll(ldapData.metaData)
-                if (loadingGroups == false) {
+                    .map();
-                    session.close();
+                final UserData user = new UserData(username, session.userDn(), ldapData.groups,
-                }
+                    metadata, session.realm());
+                roleMapper.resolveRoles(user, ActionListener.wrap(
+                    roles -> {
+                        IOUtils.close(session);
+                        String[] rolesArray = roles.toArray(new String[roles.size()]);
+                        listener.onResponse(AuthenticationResult.success(
+                            new User(username, rolesArray, null, null, metadata, true))
+                        );
+                    }, onFailure
+                ));
+            }, onFailure));
+            loadingGroups = true;
+        } finally {
+            if (loadingGroups == false) {
+                session.close();
             }
         }
     }
@@ -233,7 +262,7 @@ public final class LdapRealm extends CachingUsernamePasswordRealm {
                 resultListener.onResponse(AuthenticationResult.notHandled());
             } else {
                 ldapSessionAtomicReference.set(session);
-                buildUser(session, username, resultListener, roleMapper);
+                buildUser(session, username, resultListener, roleMapper, delegatedRealms);
             }
         }
 

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

@@ -19,6 +19,7 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ReleasableLock;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.env.Environment;
+import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.watcher.ResourceWatcherService;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationResult;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationToken;
@@ -31,12 +32,12 @@ import org.elasticsearch.xpack.core.ssl.CertParsingUtils;
 import org.elasticsearch.xpack.core.ssl.SSLConfigurationSettings;
 import org.elasticsearch.xpack.security.authc.BytesKey;
 import org.elasticsearch.xpack.security.authc.support.CachingRealm;
+import org.elasticsearch.xpack.security.authc.support.DelegatedAuthorizationSupport;
 import org.elasticsearch.xpack.security.authc.support.UserRoleMapper;
 import org.elasticsearch.xpack.security.authc.support.mapper.CompositeRoleMapper;
 import org.elasticsearch.xpack.security.authc.support.mapper.NativeRoleMappingStore;
 
 import javax.net.ssl.X509TrustManager;
-
 import java.security.MessageDigest;
 import java.security.cert.Certificate;
 import java.security.cert.CertificateEncodingException;
@@ -75,6 +76,7 @@ public class PkiRealm extends Realm implements CachingRealm {
     private final Pattern principalPattern;
     private final UserRoleMapper roleMapper;
     private final Cache<BytesKey, User> cache;
+    private DelegatedAuthorizationSupport delegatedRealms;
 
     public PkiRealm(RealmConfig config, ResourceWatcherService watcherService, NativeRoleMappingStore nativeRoleMappingStore) {
         this(config, new CompositeRoleMapper(PkiRealmSettings.TYPE, config, watcherService, nativeRoleMappingStore));
@@ -91,6 +93,15 @@ public class PkiRealm extends Realm implements CachingRealm {
                 .setExpireAfterWrite(PkiRealmSettings.CACHE_TTL_SETTING.get(config.settings()))
                 .setMaximumWeight(PkiRealmSettings.CACHE_MAX_USERS_SETTING.get(config.settings()))
                 .build();
+        this.delegatedRealms = null;
+    }
+
+    @Override
+    public void initialize(Iterable<Realm> realms, XPackLicenseState licenseState) {
+        if (delegatedRealms != null) {
+            throw new IllegalStateException("Realm has already been initialized");
+        }
+        delegatedRealms = new DelegatedAuthorizationSupport(realms, config, licenseState);
     }
 
     @Override
@@ -105,32 +116,50 @@ public class PkiRealm extends Realm implements CachingRealm {
 
     @Override
     public void authenticate(AuthenticationToken authToken, ActionListener<AuthenticationResult> listener) {
+        assert delegatedRealms != null : "Realm has not been initialized correctly";
         X509AuthenticationToken token = (X509AuthenticationToken)authToken;
         try {
             final BytesKey fingerprint = computeFingerprint(token.credentials()[0]);
             User user = cache.get(fingerprint);
             if (user != null) {
-                listener.onResponse(AuthenticationResult.success(user));
+                if (delegatedRealms.hasDelegation()) {
+                    delegatedRealms.resolve(token.principal(), listener);
+                } else {
+                    listener.onResponse(AuthenticationResult.success(user));
+                }
             } else if (isCertificateChainTrusted(trustManager, token, logger) == false) {
                 listener.onResponse(AuthenticationResult.unsuccessful("Certificate for " + token.dn() + " is not trusted", null));
             } else {
-                final Map<String, Object> metadata = Collections.singletonMap("pki_dn", token.dn());
+                final ActionListener<AuthenticationResult> cachingListener = ActionListener.wrap(result -> {
-                final UserRoleMapper.UserData userData = new UserRoleMapper.UserData(token.principal(),
+                    if (result.isAuthenticated()) {
-                        token.dn(), Collections.emptySet(), metadata, this.config);
+                        try (ReleasableLock ignored = readLock.acquire()) {
-                roleMapper.resolveRoles(userData, ActionListener.wrap(roles -> {
+                            cache.put(fingerprint, result.getUser());
-                    final User computedUser =
+                        }
-                            new User(token.principal(), roles.toArray(new String[roles.size()]), null, null, metadata, true);
-                    try (ReleasableLock ignored = readLock.acquire()) {
-                        cache.put(fingerprint, computedUser);
                     }
-                    listener.onResponse(AuthenticationResult.success(computedUser));
+                    listener.onResponse(result);
-                }, listener::onFailure));
+                }, listener::onFailure);
+                if (delegatedRealms.hasDelegation()) {
+                    delegatedRealms.resolve(token.principal(), cachingListener);
+                } else {
+                    this.buildUser(token, cachingListener);
+                }
             }
         } catch (CertificateEncodingException e) {
             listener.onResponse(AuthenticationResult.unsuccessful("Certificate for " + token.dn() + " has encoding issues", e));
         }
     }
 
+    private void buildUser(X509AuthenticationToken token, ActionListener<AuthenticationResult> listener) {
+        final Map<String, Object> metadata = Collections.singletonMap("pki_dn", token.dn());
+        final UserRoleMapper.UserData userData = new UserRoleMapper.UserData(token.principal(),
+                token.dn(), Collections.emptySet(), metadata, this.config);
+        roleMapper.resolveRoles(userData, ActionListener.wrap(roles -> {
+            final User computedUser =
+                    new User(token.principal(), roles.toArray(new String[roles.size()]), null, null, metadata, true);
+            listener.onResponse(AuthenticationResult.success(computedUser));
+        }, listener::onFailure));
+    }
+
     @Override
     public void lookupUser(String username, ActionListener<User> listener) {
         listener.onResponse(null);

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

@@ -33,6 +33,7 @@ import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.CollectionUtils;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.util.set.Sets;
+import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.watcher.FileChangesListener;
 import org.elasticsearch.watcher.FileWatcher;
 import org.elasticsearch.watcher.ResourceWatcherService;
@@ -46,10 +47,12 @@ import org.elasticsearch.xpack.core.security.authc.saml.SamlRealmSettings;
 import org.elasticsearch.xpack.core.security.user.User;
 import org.elasticsearch.xpack.core.ssl.SSLConfiguration;
 import org.elasticsearch.xpack.core.ssl.CertParsingUtils;
+import org.elasticsearch.xpack.core.ssl.SSLConfiguration;
 import org.elasticsearch.xpack.core.ssl.SSLService;
 import org.elasticsearch.xpack.core.ssl.X509KeyPairSettings;
 import org.elasticsearch.xpack.security.authc.Realms;
 import org.elasticsearch.xpack.security.authc.TokenService;
+import org.elasticsearch.xpack.security.authc.support.DelegatedAuthorizationSupport;
 import org.elasticsearch.xpack.security.authc.support.UserRoleMapper;
 import org.opensaml.core.criterion.EntityIdCriterion;
 import org.opensaml.saml.common.xml.SAMLConstants;
@@ -117,6 +120,7 @@ import static org.elasticsearch.xpack.core.security.authc.saml.SamlRealmSettings
 import static org.elasticsearch.xpack.core.security.authc.saml.SamlRealmSettings.NAME_ATTRIBUTE;
 import static org.elasticsearch.xpack.core.security.authc.saml.SamlRealmSettings.POPULATE_USER_METADATA;
 import static org.elasticsearch.xpack.core.security.authc.saml.SamlRealmSettings.PRINCIPAL_ATTRIBUTE;
+import static org.elasticsearch.xpack.core.security.authc.saml.SamlRealmSettings.REQUESTED_AUTHN_CONTEXT_CLASS_REF;
 import static org.elasticsearch.xpack.core.security.authc.saml.SamlRealmSettings.SIGNING_KEY_ALIAS;
 import static org.elasticsearch.xpack.core.security.authc.saml.SamlRealmSettings.SIGNING_MESSAGE_TYPES;
 import static org.elasticsearch.xpack.core.security.authc.saml.SamlRealmSettings.SIGNING_SETTINGS;
@@ -124,7 +128,6 @@ import static org.elasticsearch.xpack.core.security.authc.saml.SamlRealmSettings
 import static org.elasticsearch.xpack.core.security.authc.saml.SamlRealmSettings.SP_ENTITY_ID;
 import static org.elasticsearch.xpack.core.security.authc.saml.SamlRealmSettings.SP_LOGOUT;
 import static org.elasticsearch.xpack.core.security.authc.saml.SamlRealmSettings.TYPE;
-import static org.elasticsearch.xpack.core.security.authc.saml.SamlRealmSettings.REQUESTED_AUTHN_CONTEXT_CLASS_REF;
 
 /**
  * This class is {@link Releasable} because it uses a library that thinks timers and timer tasks
@@ -166,6 +169,7 @@ public final class SamlRealm extends Realm implements Releasable {
     private final AttributeParser nameAttribute;
     private final AttributeParser mailAttribute;
 
+    private DelegatedAuthorizationSupport delegatedRealms;
 
     /**
      * Factory for SAML realm.
@@ -231,6 +235,14 @@ public final class SamlRealm extends Realm implements Releasable {
         this.releasables = new ArrayList<>();
     }
 
+    @Override
+    public void initialize(Iterable<Realm> realms, XPackLicenseState licenseState) {
+        if (delegatedRealms != null) {
+            throw new IllegalStateException("Realm has already been initialized");
+        }
+        delegatedRealms = new DelegatedAuthorizationSupport(realms, config, licenseState);
+    }
+
     static String require(RealmConfig config, Setting<String> setting) {
         final String value = setting.get(config.settings());
         if (value.isEmpty()) {
@@ -402,14 +414,27 @@ public final class SamlRealm extends Realm implements Releasable {
         }
     }
 
-    private void buildUser(SamlAttributes attributes, ActionListener<AuthenticationResult> listener) {
+    private void buildUser(SamlAttributes attributes, ActionListener<AuthenticationResult> baseListener) {
         final String principal = resolveSingleValueAttribute(attributes, principalAttribute, PRINCIPAL_ATTRIBUTE.name());
         if (Strings.isNullOrEmpty(principal)) {
-            listener.onResponse(AuthenticationResult.unsuccessful(
+            baseListener.onResponse(AuthenticationResult.unsuccessful(
                     principalAttribute + " not found in " + attributes.attributes(), null));
             return;
         }
 
+        final Map<String, Object> tokenMetadata = createTokenMetadata(attributes.name(), attributes.session());
+        ActionListener<AuthenticationResult> wrappedListener = ActionListener.wrap(auth -> {
+            if (auth.isAuthenticated()) {
+                config.threadContext().putTransient(CONTEXT_TOKEN_DATA, tokenMetadata);
+            }
+            baseListener.onResponse(auth);
+        }, baseListener::onFailure);
+
+        if (delegatedRealms.hasDelegation()) {
+            delegatedRealms.resolve(principal, wrappedListener);
+            return;
+        }
+
         final Map<String, Object> userMeta = new HashMap<>();
         if (populateUserMetadata) {
             for (SamlAttributes.SamlAttribute a : attributes.attributes()) {
@@ -424,7 +449,6 @@ public final class SamlRealm extends Realm implements Releasable {
             userMeta.put(USER_METADATA_NAMEID_FORMAT, attributes.name().format);
         }
 
-        final Map<String, Object> tokenMetadata = createTokenMetadata(attributes.name(), attributes.session());
 
         final List<String> groups = groupsAttribute.getAttribute(attributes);
         final String dn = resolveSingleValueAttribute(attributes, dnAttribute, DN_ATTRIBUTE.name());
@@ -433,9 +457,8 @@ public final class SamlRealm extends Realm implements Releasable {
         UserRoleMapper.UserData userData = new UserRoleMapper.UserData(principal, dn, groups, userMeta, config);
         roleMapper.resolveRoles(userData, ActionListener.wrap(roles -> {
             final User user = new User(principal, roles.toArray(new String[roles.size()]), name, mail, userMeta, true);
-            config.threadContext().putTransient(CONTEXT_TOKEN_DATA, tokenMetadata);
+            wrappedListener.onResponse(AuthenticationResult.success(user));
-            listener.onResponse(AuthenticationResult.success(user));
+        }, wrappedListener::onFailure));
-        }, listener::onFailure));
     }
 
     public Map<String, Object> createTokenMetadata(SamlNameId nameId, String session) {
@@ -745,10 +768,10 @@ public final class SamlRealm extends Realm implements Releasable {
                             attributes -> attributes.getAttributeValues(attributeName));
                 }
             } else if (required) {
-                throw new SettingsException("Setting" + RealmSettings.getFullSettingKey(realmConfig, setting.getAttribute())
+                throw new SettingsException("Setting " + RealmSettings.getFullSettingKey(realmConfig, setting.getAttribute())
                         + " is required");
             } else if (setting.getPattern().exists(settings)) {
-                throw new SettingsException("Setting" + RealmSettings.getFullSettingKey(realmConfig, setting.getPattern())
+                throw new SettingsException("Setting " + RealmSettings.getFullSettingKey(realmConfig, setting.getPattern())
                         + " cannot be set unless " + RealmSettings.getFullSettingKey(realmConfig, setting.getAttribute()) + " is also set");
             } else {
                 return new AttributeParser("No SAML attribute for [" + setting.name() + "]", attributes -> Collections.emptyList());

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

@@ -39,9 +39,9 @@ public abstract class CachingUsernamePasswordRealm extends UsernamePasswordRealm
         final TimeValue ttl = CachingUsernamePasswordRealmSettings.CACHE_TTL_SETTING.get(config.settings());
         if (ttl.getNanos() > 0) {
             cache = CacheBuilder.<String, ListenableFuture<UserWithHash>>builder()
-                    .setExpireAfterWrite(ttl)
+                .setExpireAfterWrite(ttl)
-                    .setMaximumWeight(CachingUsernamePasswordRealmSettings.CACHE_MAX_USERS_SETTING.get(config.settings()))
+                .setMaximumWeight(CachingUsernamePasswordRealmSettings.CACHE_MAX_USERS_SETTING.get(config.settings()))
-                    .build();
+                .build();
         } else {
             cache = null;
         }
@@ -108,10 +108,16 @@ public abstract class CachingUsernamePasswordRealm extends UsernamePasswordRealm
                 listenableCacheEntry.addListener(ActionListener.wrap(authenticatedUserWithHash -> {
                     if (authenticatedUserWithHash != null && authenticatedUserWithHash.verify(token.credentials())) {
                         // cached credential hash matches the credential hash for this forestalled request
-                        final User user = authenticatedUserWithHash.user;
+                        handleCachedAuthentication(authenticatedUserWithHash.user, ActionListener.wrap(cacheResult -> {
-                        logger.debug("realm [{}] authenticated user [{}], with roles [{}], from cache", name(), token.principal(),
+                            if (cacheResult.isAuthenticated()) {
-                                user.roles());
+                                logger.debug("realm [{}] authenticated user [{}], with roles [{}]",
-                        listener.onResponse(AuthenticationResult.success(user));
+                                    name(), token.principal(), cacheResult.getUser().roles());
+                            } else {
+                                logger.debug("realm [{}] authenticated user [{}] from cache, but then failed [{}]",
+                                    name(), token.principal(), cacheResult.getMessage());
+                            }
+                            listener.onResponse(cacheResult);
+                        }, listener::onFailure));
                     } else {
                         // The inflight request has failed or its credential hash does not match the
                         // hash of the credential for this forestalled request.
@@ -153,6 +159,16 @@ public abstract class CachingUsernamePasswordRealm extends UsernamePasswordRealm
         }
     }
 
+    /**
+     * {@code handleCachedAuthentication} is called when a {@link User} is retrieved from the cache.
+     * The first {@code user} parameter is the user object that was found in the cache.
+     * The default implementation returns a {@link AuthenticationResult#success(User) success result} with the
+     * provided user, but sub-classes can return a different {@code User} object, or an unsuccessful result.
+     */
+    protected void handleCachedAuthentication(User user, ActionListener<AuthenticationResult> listener) {
+        listener.onResponse(AuthenticationResult.success(user));
+    }
+
     @Override
     public void usageStats(ActionListener<Map<String, Object>> listener) {
         super.usageStats(ActionListener.wrap(stats -> {

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

@@ -0,0 +1,146 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.security.authc.support;
+
+import org.apache.logging.log4j.Logger;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.logging.Loggers;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.license.LicenseUtils;
+import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.xpack.core.security.authc.AuthenticationResult;
+import org.elasticsearch.xpack.core.security.authc.Realm;
+import org.elasticsearch.xpack.core.security.authc.RealmConfig;
+import org.elasticsearch.xpack.core.security.authc.RealmSettings;
+import org.elasticsearch.xpack.core.security.authc.support.DelegatedAuthorizationSettings;
+import org.elasticsearch.xpack.core.security.user.User;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.elasticsearch.common.Strings.collectionToDelimitedString;
+
+/**
+ * Utility class for supporting "delegated authorization" (aka "authorization_realms", aka "lookup realms").
+ * A {@link Realm} may support delegating authorization to another realm. It does this by registering a
+ * setting for {@link DelegatedAuthorizationSettings#AUTHZ_REALMS}, and constructing an instance of this
+ * class. Then, after the realm has performed any authentication steps, if {@link #hasDelegation()} is
+ * {@code true}, it delegates the construction of the {@link User} object and {@link AuthenticationResult}
+ * to {@link #resolve(String, ActionListener)}.
+ */
+public class DelegatedAuthorizationSupport {
+
+    private final RealmUserLookup lookup;
+    private final Logger logger;
+    private final XPackLicenseState licenseState;
+
+    /**
+     * Resolves the {@link DelegatedAuthorizationSettings#AUTHZ_REALMS} setting from {@code config} and calls
+     * {@link #DelegatedAuthorizationSupport(Iterable, List, Settings, ThreadContext, XPackLicenseState)}
+     */
+    public DelegatedAuthorizationSupport(Iterable<? extends Realm> allRealms, RealmConfig config, XPackLicenseState licenseState) {
+        this(allRealms, DelegatedAuthorizationSettings.AUTHZ_REALMS.get(config.settings()), config.globalSettings(), config.threadContext(),
+            licenseState);
+    }
+
+    /**
+     * Constructs a new object that delegates to the named realms ({@code lookupRealms}), which must exist within
+     * {@code allRealms}.
+     * @throws IllegalArgumentException if one of the specified realms does not exist
+     */
+    protected DelegatedAuthorizationSupport(Iterable<? extends Realm> allRealms, List<String> lookupRealms, Settings settings,
+                                            ThreadContext threadContext, XPackLicenseState licenseState) {
+        final List<Realm> resolvedLookupRealms = resolveRealms(allRealms, lookupRealms);
+        checkForRealmChains(resolvedLookupRealms, settings);
+        this.lookup = new RealmUserLookup(resolvedLookupRealms, threadContext);
+        this.logger = Loggers.getLogger(getClass());
+        this.licenseState = licenseState;
+    }
+
+    /**
+     * Are there any realms configured for delegated lookup
+     */
+    public boolean hasDelegation() {
+        return this.lookup.hasRealms();
+    }
+
+    /**
+     * Attempts to find the user specified by {@code username} in one of the delegated realms.
+     * The realms are searched in the order specified during construction.
+     * Returns a {@link AuthenticationResult#success(User) successful result} if a {@link User}
+     * was found, otherwise returns an
+     * {@link AuthenticationResult#unsuccessful(String, Exception) unsuccessful result}
+     * with a meaningful diagnostic message.
+     */
+    public void resolve(String username, ActionListener<AuthenticationResult> resultListener) {
+        if (licenseState.isAuthorizationRealmAllowed() == false) {
+            resultListener.onResponse(AuthenticationResult.unsuccessful(
+                DelegatedAuthorizationSettings.AUTHZ_REALMS.getKey() + " are not permitted",
+                LicenseUtils.newComplianceException(DelegatedAuthorizationSettings.AUTHZ_REALMS.getKey())
+            ));
+            return;
+        }
+        if (hasDelegation() == false) {
+            resultListener.onResponse(AuthenticationResult.unsuccessful(
+                "No [" + DelegatedAuthorizationSettings.AUTHZ_REALMS.getKey() + "] have been configured", null));
+            return;
+        }
+        ActionListener<Tuple<User, Realm>> userListener = ActionListener.wrap(tuple -> {
+            if (tuple != null) {
+                logger.trace("Found user " + tuple.v1() + " in realm " + tuple.v2());
+                resultListener.onResponse(AuthenticationResult.success(tuple.v1()));
+            } else {
+                resultListener.onResponse(AuthenticationResult.unsuccessful("the principal [" + username
+                    + "] was authenticated, but no user could be found in realms [" + collectionToDelimitedString(lookup.getRealms(), ",")
+                    + "]", null));
+            }
+        }, resultListener::onFailure);
+        lookup.lookup(username, userListener);
+    }
+
+    private List<Realm> resolveRealms(Iterable<? extends Realm> allRealms, List<String> lookupRealms) {
+        final List<Realm> result = new ArrayList<>(lookupRealms.size());
+        for (String name : lookupRealms) {
+            result.add(findRealm(name, allRealms));
+        }
+        assert result.size() == lookupRealms.size();
+        return result;
+    }
+
+    /**
+     * Checks for (and rejects) chains of delegation in the provided realms.
+     * A chain occurs when "realmA" delegates authorization to "realmB", and realmB also delegates authorization (to any realm).
+     * Since "realmB" does not handle its own authorization, it is not a valid target for delegated authorization.
+     * @param delegatedRealms The list of realms that are going to be used for authorization. If is an error if any of these realms are
+     *                        also configured to delegate their authorization.
+     * @throws IllegalArgumentException if a chain is detected
+     */
+    private void checkForRealmChains(Iterable<Realm> delegatedRealms, Settings globalSettings) {
+        final Map<String, Settings> settingsByRealm = RealmSettings.getRealmSettings(globalSettings);
+        for (Realm realm : delegatedRealms) {
+            final Settings realmSettings = settingsByRealm.get(realm.name());
+            if (realmSettings != null && DelegatedAuthorizationSettings.AUTHZ_REALMS.exists(realmSettings)) {
+                throw new IllegalArgumentException("cannot use realm [" + realm +
+                    "] as an authorization realm - it is already delegating authorization to [" +
+                    DelegatedAuthorizationSettings.AUTHZ_REALMS.get(realmSettings) + "]");
+            }
+        }
+    }
+
+    private Realm findRealm(String name, Iterable<? extends Realm> allRealms) {
+        for (Realm realm : allRealms) {
+            if (name.equals(realm.name())) {
+                return realm;
+            }
+        }
+        throw new IllegalArgumentException("configured authorization realm [" + name + "] does not exist (or is not enabled)");
+    }
+
+}

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

@@ -0,0 +1,63 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.security.authc.support;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.xpack.core.common.IteratingActionListener;
+import org.elasticsearch.xpack.core.security.authc.Realm;
+import org.elasticsearch.xpack.core.security.user.User;
+
+import java.util.Collections;
+import java.util.List;
+
+public class RealmUserLookup {
+
+    private final List<? extends Realm> realms;
+    private final ThreadContext threadContext;
+
+    public RealmUserLookup(List<? extends Realm> realms, ThreadContext threadContext) {
+        this.realms = realms;
+        this.threadContext = threadContext;
+    }
+
+    public List<Realm> getRealms() {
+        return Collections.unmodifiableList(realms);
+    }
+
+    public boolean hasRealms() {
+        return realms.isEmpty() == false;
+    }
+
+    /**
+     * Lookup the {@code principal} in the list of {@link #realms}.
+     * The realms are consulted in order. When one realm responds with a non-null {@link User}, this
+     * is returned with the matching realm, through the {@code listener}.
+     * If no user if found (including the case where the {@link #realms} list is empty), then
+     * {@link ActionListener#onResponse(Object)} is called with a {@code null} {@link Tuple}.
+     */
+    public void lookup(String principal, ActionListener<Tuple<User, Realm>> listener) {
+        final IteratingActionListener<Tuple<User, Realm>, ? extends Realm> userLookupListener =
+            new IteratingActionListener<>(listener,
+                (realm, lookupUserListener) -> realm.lookupUser(principal,
+                    ActionListener.wrap(foundUser -> {
+                            if (foundUser != null) {
+                                lookupUserListener.onResponse(new Tuple<>(foundUser, realm));
+                            } else {
+                                lookupUserListener.onResponse(null);
+                            }
+                        },
+                        lookupUserListener::onFailure)),
+                realms, threadContext);
+        try {
+            userLookupListener.run();
+        } catch (Exception e) {
+            listener.onFailure(e);
+        }
+    }
+}

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/kerberos/KerberosRealmAuthenticateFailedTests.java

@@ -11,13 +11,20 @@ import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.settings.SecureString;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.env.TestEnvironment;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationResult;
+import org.elasticsearch.xpack.core.security.authc.RealmConfig;
 import org.elasticsearch.xpack.core.security.authc.kerberos.KerberosRealmSettings;
 import org.elasticsearch.xpack.core.security.authc.support.UsernamePasswordToken;
+import org.elasticsearch.xpack.security.authc.support.MockLookupRealm;
 import org.elasticsearch.xpack.core.security.user.User;
 import org.ietf.jgss.GSSException;
 
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
+import java.util.Collections;
 import java.util.List;
 
 import javax.security.auth.login.LoginException;
@@ -29,7 +36,9 @@ import static org.hamcrest.Matchers.notNullValue;
 import static org.mockito.AdditionalMatchers.aryEq;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
 
 public class KerberosRealmAuthenticateFailedTests extends KerberosRealmTestCase {
 
@@ -105,4 +114,30 @@ public class KerberosRealmAuthenticateFailedTests extends KerberosRealmTestCase
                     any(ActionListener.class));
         }
     }
+
+    public void testDelegatedAuthorizationFailedToResolve() throws Exception {
+        final String username = randomPrincipalName();
+        final MockLookupRealm otherRealm = new MockLookupRealm(new RealmConfig("other_realm", Settings.EMPTY, globalSettings,
+                TestEnvironment.newEnvironment(globalSettings), new ThreadContext(globalSettings)));
+        final User lookupUser = new User(randomAlphaOfLength(5));
+        otherRealm.registerUser(lookupUser);
+
+        settings = Settings.builder().put(settings).putList("authorization_realms", "other_realm").build();
+        final KerberosRealm kerberosRealm = createKerberosRealm(Collections.singletonList(otherRealm), username);
+        final byte[] decodedTicket = "base64encodedticket".getBytes(StandardCharsets.UTF_8);
+        final Path keytabPath = config.env().configFile().resolve(KerberosRealmSettings.HTTP_SERVICE_KEYTAB_PATH.get(config.settings()));
+        final boolean krbDebug = KerberosRealmSettings.SETTING_KRB_DEBUG_ENABLE.get(config.settings());
+        mockKerberosTicketValidator(decodedTicket, keytabPath, krbDebug, new Tuple<>(username, "out-token"), null);
+        final KerberosAuthenticationToken kerberosAuthenticationToken = new KerberosAuthenticationToken(decodedTicket);
+
+        final PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
+        kerberosRealm.authenticate(kerberosAuthenticationToken, future);
+
+        AuthenticationResult result = future.actionGet();
+        assertThat(result.getStatus(), is(equalTo(AuthenticationResult.Status.CONTINUE)));
+        verify(mockKerberosTicketValidator, times(1)).validateTicket(aryEq(decodedTicket), eq(keytabPath), eq(krbDebug),
+                any(ActionListener.class));
+        verify(mockNativeRoleMappingStore).refreshRealmOnChange(kerberosRealm);
+        verifyNoMoreInteractions(mockKerberosTicketValidator, mockNativeRoleMappingStore);
+    }
 }

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/kerberos/KerberosRealmTestCase.java

@@ -13,11 +13,13 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.env.TestEnvironment;
+import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.watcher.ResourceWatcherService;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationResult;
+import org.elasticsearch.xpack.core.security.authc.Realm;
 import org.elasticsearch.xpack.core.security.authc.RealmConfig;
 import org.elasticsearch.xpack.core.security.authc.kerberos.KerberosRealmSettings;
 import org.elasticsearch.xpack.core.security.support.Exceptions;
@@ -30,6 +32,7 @@ import org.junit.Before;
 
 import java.nio.file.Path;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -58,6 +61,7 @@ public abstract class KerberosRealmTestCase extends ESTestCase {
 
     protected KerberosTicketValidator mockKerberosTicketValidator;
     protected NativeRoleMappingStore mockNativeRoleMappingStore;
+    protected XPackLicenseState licenseState;
 
     protected static final Set<String> roles = Sets.newHashSet("admin", "kibana_user");
 
@@ -69,6 +73,8 @@ public abstract class KerberosRealmTestCase extends ESTestCase {
         globalSettings = Settings.builder().put("path.home", dir).build();
         settings = KerberosTestCase.buildKerberosRealmSettings(KerberosTestCase.writeKeyTab(dir.resolve("key.keytab"), "asa").toString(),
                 100, "10m", true, randomBoolean());
+        licenseState = mock(XPackLicenseState.class);
+        when(licenseState.isAuthorizationRealmAllowed()).thenReturn(true);
     }
 
     @After
@@ -102,12 +108,18 @@ public abstract class KerberosRealmTestCase extends ESTestCase {
     }
 
     protected KerberosRealm createKerberosRealm(final String... userForRoleMapping) {
+        return createKerberosRealm(Collections.emptyList(), userForRoleMapping);
+    }
+
+    protected KerberosRealm createKerberosRealm(final List<Realm> delegatedRealms, final String... userForRoleMapping) {
         config = new RealmConfig("test-kerb-realm", settings, globalSettings, TestEnvironment.newEnvironment(globalSettings),
                 new ThreadContext(globalSettings));
         mockNativeRoleMappingStore = roleMappingStore(Arrays.asList(userForRoleMapping));
         mockKerberosTicketValidator = mock(KerberosTicketValidator.class);
         final KerberosRealm kerberosRealm =
                 new KerberosRealm(config, mockNativeRoleMappingStore, mockKerberosTicketValidator, threadPool, null);
+        Collections.shuffle(delegatedRealms, random());
+        kerberosRealm.initialize(delegatedRealms, licenseState);
         return kerberosRealm;
     }
 

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/kerberos/KerberosRealmTests.java

@@ -12,6 +12,7 @@ import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.settings.SecureString;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.env.TestEnvironment;
 import org.elasticsearch.xpack.core.security.user.User;
@@ -20,6 +21,7 @@ import org.elasticsearch.xpack.core.security.authc.AuthenticationResult;
 import org.elasticsearch.xpack.core.security.authc.RealmConfig;
 import org.elasticsearch.xpack.core.security.authc.kerberos.KerberosRealmSettings;
 import org.elasticsearch.xpack.core.security.authc.support.UsernamePasswordToken;
+import org.elasticsearch.xpack.security.authc.support.MockLookupRealm;
 import org.elasticsearch.xpack.security.authc.support.UserRoleMapper.UserData;
 import org.ietf.jgss.GSSException;
 
@@ -34,6 +36,7 @@ import java.nio.file.attribute.FileAttribute;
 import java.nio.file.attribute.PosixFilePermission;
 import java.nio.file.attribute.PosixFilePermissions;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.EnumSet;
 import java.util.Locale;
 import java.util.Set;
@@ -47,6 +50,7 @@ import static org.mockito.AdditionalMatchers.aryEq;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
@@ -160,4 +164,38 @@ public class KerberosRealmTests extends KerberosRealmTestCase {
                 () -> new KerberosRealm(config, mockNativeRoleMappingStore, mockKerberosTicketValidator, threadPool, null));
         assertThat(iae.getMessage(), is(equalTo(expectedErrorMessage)));
     }
+
+    public void testDelegatedAuthorization() throws Exception {
+        final String username = randomPrincipalName();
+        final String expectedUsername = maybeRemoveRealmName(username);
+        final MockLookupRealm otherRealm = spy(new MockLookupRealm(new RealmConfig("other_realm", Settings.EMPTY, globalSettings,
+                TestEnvironment.newEnvironment(globalSettings), new ThreadContext(globalSettings))));
+        final User lookupUser = new User(expectedUsername, new String[] { "admin-role" }, expectedUsername,
+                expectedUsername + "@example.com", Collections.singletonMap("k1", "v1"), true);
+        otherRealm.registerUser(lookupUser);
+
+        settings = Settings.builder().put(settings).putList("authorization_realms", "other_realm").build();
+        final KerberosRealm kerberosRealm = createKerberosRealm(Collections.singletonList(otherRealm), username);
+        final User expectedUser = lookupUser;
+        final byte[] decodedTicket = "base64encodedticket".getBytes(StandardCharsets.UTF_8);
+        final Path keytabPath = config.env().configFile().resolve(KerberosRealmSettings.HTTP_SERVICE_KEYTAB_PATH.get(config.settings()));
+        final boolean krbDebug = KerberosRealmSettings.SETTING_KRB_DEBUG_ENABLE.get(config.settings());
+        mockKerberosTicketValidator(decodedTicket, keytabPath, krbDebug, new Tuple<>(username, "out-token"), null);
+        final KerberosAuthenticationToken kerberosAuthenticationToken = new KerberosAuthenticationToken(decodedTicket);
+
+        PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
+        kerberosRealm.authenticate(kerberosAuthenticationToken, future);
+        assertSuccessAuthenticationResult(expectedUser, "out-token", future.actionGet());
+
+        future = new PlainActionFuture<>();
+        kerberosRealm.authenticate(kerberosAuthenticationToken, future);
+        assertSuccessAuthenticationResult(expectedUser, "out-token", future.actionGet());
+
+        verify(mockKerberosTicketValidator, times(2)).validateTicket(aryEq(decodedTicket), eq(keytabPath), eq(krbDebug),
+                any(ActionListener.class));
+        verify(mockNativeRoleMappingStore).refreshRealmOnChange(kerberosRealm);
+        verifyNoMoreInteractions(mockKerberosTicketValidator, mockNativeRoleMappingStore);
+        verify(otherRealm, times(2)).lookupUser(eq(expectedUsername), any(ActionListener.class));
+    }
 }
+

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/ldap/ActiveDirectoryRealmTests.java

@@ -22,6 +22,8 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.TestEnvironment;
+import org.elasticsearch.license.TestUtils;
+import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -48,6 +50,7 @@ import org.junit.BeforeClass;
 import java.security.AccessController;
 import java.security.PrivilegedExceptionAction;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
@@ -91,6 +94,7 @@ public class ActiveDirectoryRealmTests extends ESTestCase {
     private ThreadPool threadPool;
     private Settings globalSettings;
     private SSLService sslService;
+    private XPackLicenseState licenseState;
 
     @BeforeClass
     public static void setNumberOfLdapServers() {
@@ -125,6 +129,7 @@ public class ActiveDirectoryRealmTests extends ESTestCase {
         resourceWatcherService = new ResourceWatcherService(Settings.EMPTY, threadPool);
         globalSettings = Settings.builder().put("path.home", createTempDir()).build();
         sslService = new SSLService(globalSettings, TestEnvironment.newEnvironment(globalSettings));
+        licenseState = new TestUtils.UpdatableLicenseState();
     }
 
     @After
@@ -163,6 +168,7 @@ public class ActiveDirectoryRealmTests extends ESTestCase {
         ActiveDirectorySessionFactory sessionFactory = new ActiveDirectorySessionFactory(config, sslService, threadPool);
         DnRoleMapper roleMapper = new DnRoleMapper(config, resourceWatcherService);
         LdapRealm realm = new LdapRealm(LdapRealmSettings.AD_TYPE, config, sessionFactory, roleMapper, threadPool);
+        realm.initialize(Collections.singleton(realm), licenseState);
 
         PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
         realm.authenticate(new UsernamePasswordToken("CN=ironman", new SecureString(PASSWORD)), future);
@@ -179,6 +185,7 @@ public class ActiveDirectoryRealmTests extends ESTestCase {
         ActiveDirectorySessionFactory sessionFactory = new ActiveDirectorySessionFactory(config, sslService, threadPool);
         DnRoleMapper roleMapper = new DnRoleMapper(config, resourceWatcherService);
         LdapRealm realm = new LdapRealm(LdapRealmSettings.AD_TYPE, config, sessionFactory, roleMapper, threadPool);
+        realm.initialize(Collections.singleton(realm), licenseState);
 
         // Thor does not have a UPN of form CN=Thor@ad.test.elasticsearch.com
         PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
@@ -203,6 +210,7 @@ public class ActiveDirectoryRealmTests extends ESTestCase {
         ActiveDirectorySessionFactory sessionFactory = spy(new ActiveDirectorySessionFactory(config, sslService, threadPool));
         DnRoleMapper roleMapper = new DnRoleMapper(config, resourceWatcherService);
         LdapRealm realm = new LdapRealm(LdapRealmSettings.AD_TYPE, config, sessionFactory, roleMapper, threadPool);
+        realm.initialize(Collections.singleton(realm), licenseState);
 
         int count = randomIntBetween(2, 10);
         for (int i = 0; i < count; i++) {
@@ -221,6 +229,7 @@ public class ActiveDirectoryRealmTests extends ESTestCase {
         ActiveDirectorySessionFactory sessionFactory = spy(new ActiveDirectorySessionFactory(config, sslService, threadPool));
         DnRoleMapper roleMapper = new DnRoleMapper(config, resourceWatcherService);
         LdapRealm realm = new LdapRealm(LdapRealmSettings.AD_TYPE, config, sessionFactory, roleMapper, threadPool);
+        realm.initialize(Collections.singleton(realm), licenseState);
 
         int count = randomIntBetween(2, 10);
         for (int i = 0; i < count; i++) {
@@ -239,6 +248,7 @@ public class ActiveDirectoryRealmTests extends ESTestCase {
         ActiveDirectorySessionFactory sessionFactory = spy(new ActiveDirectorySessionFactory(config, sslService, threadPool));
         DnRoleMapper roleMapper = new DnRoleMapper(config, resourceWatcherService);
         LdapRealm realm = new LdapRealm(LdapRealmSettings.AD_TYPE, config, sessionFactory, roleMapper, threadPool);
+        realm.initialize(Collections.singleton(realm), licenseState);
 
         int count = randomIntBetween(2, 10);
         for (int i = 0; i < count; i++) {
@@ -287,6 +297,7 @@ public class ActiveDirectoryRealmTests extends ESTestCase {
         try (ActiveDirectorySessionFactory sessionFactory = new ActiveDirectorySessionFactory(config, sslService, threadPool)) {
             DnRoleMapper roleMapper = new DnRoleMapper(config, resourceWatcherService);
             LdapRealm realm = new LdapRealm(LdapRealmSettings.AD_TYPE, config, sessionFactory, roleMapper, threadPool);
+            realm.initialize(Collections.singleton(realm), licenseState);
 
             PlainActionFuture<User> future = new PlainActionFuture<>();
             realm.lookupUser("CN=Thor", future);
@@ -304,6 +315,7 @@ public class ActiveDirectoryRealmTests extends ESTestCase {
         ActiveDirectorySessionFactory sessionFactory = new ActiveDirectorySessionFactory(config, sslService, threadPool);
         DnRoleMapper roleMapper = new DnRoleMapper(config, resourceWatcherService);
         LdapRealm realm = new LdapRealm(LdapRealmSettings.AD_TYPE, config, sessionFactory, roleMapper, threadPool);
+        realm.initialize(Collections.singleton(realm), licenseState);
 
         PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
         realm.authenticate(new UsernamePasswordToken("CN=ironman", new SecureString(PASSWORD)), future);
@@ -320,6 +332,7 @@ public class ActiveDirectoryRealmTests extends ESTestCase {
         ActiveDirectorySessionFactory sessionFactory = new ActiveDirectorySessionFactory(config, sslService, threadPool);
         DnRoleMapper roleMapper = new DnRoleMapper(config, resourceWatcherService);
         LdapRealm realm = new LdapRealm(LdapRealmSettings.AD_TYPE, config, sessionFactory, roleMapper, threadPool);
+        realm.initialize(Collections.singleton(realm), licenseState);
 
         PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
         realm.authenticate(new UsernamePasswordToken("CN=Thor", new SecureString(PASSWORD)), future);
@@ -338,6 +351,7 @@ public class ActiveDirectoryRealmTests extends ESTestCase {
         ActiveDirectorySessionFactory sessionFactory = new ActiveDirectorySessionFactory(config, sslService, threadPool);
         DnRoleMapper roleMapper = new DnRoleMapper(config, resourceWatcherService);
         LdapRealm realm = new LdapRealm(LdapRealmSettings.AD_TYPE, config, sessionFactory, roleMapper, threadPool);
+        realm.initialize(Collections.singleton(realm), licenseState);
 
         PlainActionFuture<Map<String, Object>> future = new PlainActionFuture<>();
         realm.usageStats(future);

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/ldap/LdapRealmTests.java

@@ -14,6 +14,7 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.TestEnvironment;
+import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.watcher.ResourceWatcherService;
@@ -25,6 +26,7 @@ import org.elasticsearch.xpack.core.security.authc.ldap.LdapRealmSettings;
 import org.elasticsearch.xpack.core.security.authc.ldap.LdapSessionFactorySettings;
 import org.elasticsearch.xpack.core.security.authc.ldap.support.LdapSearchScope;
 import org.elasticsearch.xpack.core.security.authc.support.CachingUsernamePasswordRealmSettings;
+import org.elasticsearch.xpack.core.security.authc.support.DelegatedAuthorizationSettings;
 import org.elasticsearch.xpack.core.security.authc.support.DnRoleMapperSettings;
 import org.elasticsearch.xpack.core.security.authc.support.UsernamePasswordToken;
 import org.elasticsearch.xpack.core.security.user.User;
@@ -33,10 +35,12 @@ import org.elasticsearch.xpack.core.ssl.VerificationMode;
 import org.elasticsearch.xpack.security.authc.ldap.support.LdapTestCase;
 import org.elasticsearch.xpack.security.authc.ldap.support.SessionFactory;
 import org.elasticsearch.xpack.security.authc.support.DnRoleMapper;
+import org.elasticsearch.xpack.security.authc.support.MockLookupRealm;
 import org.junit.After;
 import org.junit.Before;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
@@ -50,11 +54,14 @@ import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
+import static org.hamcrest.Matchers.sameInstance;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 public class LdapRealmTests extends LdapTestCase {
 
@@ -68,6 +75,7 @@ public class LdapRealmTests extends LdapTestCase {
     private ResourceWatcherService resourceWatcherService;
     private Settings defaultGlobalSettings;
     private SSLService sslService;
+    private XPackLicenseState licenseState;
 
     @Before
     public void init() throws Exception {
@@ -75,6 +83,8 @@ public class LdapRealmTests extends LdapTestCase {
         resourceWatcherService = new ResourceWatcherService(Settings.EMPTY, threadPool);
         defaultGlobalSettings = Settings.builder().put("path.home", createTempDir()).build();
         sslService = new SSLService(defaultGlobalSettings, TestEnvironment.newEnvironment(defaultGlobalSettings));
+        licenseState = mock(XPackLicenseState.class);
+        when(licenseState.isAuthorizationRealmAllowed()).thenReturn(true);
     }
 
     @After
@@ -87,10 +97,12 @@ public class LdapRealmTests extends LdapTestCase {
         String groupSearchBase = "o=sevenSeas";
         String userTemplate = VALID_USER_TEMPLATE;
         Settings settings = buildLdapSettings(ldapUrls(), userTemplate, groupSearchBase, LdapSearchScope.SUB_TREE);
-        RealmConfig config = new RealmConfig("test-ldap-realm", settings, defaultGlobalSettings, TestEnvironment.newEnvironment(defaultGlobalSettings), new ThreadContext(defaultGlobalSettings));
+        RealmConfig config = new RealmConfig("test-ldap-realm", settings, defaultGlobalSettings,
+            TestEnvironment.newEnvironment(defaultGlobalSettings), new ThreadContext(defaultGlobalSettings));
         LdapSessionFactory ldapFactory = new LdapSessionFactory(config, sslService, threadPool);
         LdapRealm ldap = new LdapRealm(LdapRealmSettings.LDAP_TYPE, config, ldapFactory, buildGroupAsRoleMapper(resourceWatcherService),
                 threadPool);
+        ldap.initialize(Collections.singleton(ldap), licenseState);
 
         PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
         ldap.authenticate(new UsernamePasswordToken(VALID_USERNAME, new SecureString(PASSWORD)), future);
@@ -111,11 +123,13 @@ public class LdapRealmTests extends LdapTestCase {
         Settings settings = Settings.builder()
                 .put(buildLdapSettings(ldapUrls(), userTemplate, groupSearchBase, LdapSearchScope.ONE_LEVEL))
                 .build();
-        RealmConfig config = new RealmConfig("test-ldap-realm", settings, defaultGlobalSettings, TestEnvironment.newEnvironment(defaultGlobalSettings), new ThreadContext(defaultGlobalSettings));
+        RealmConfig config = new RealmConfig("test-ldap-realm", settings, defaultGlobalSettings,
+            TestEnvironment.newEnvironment(defaultGlobalSettings), new ThreadContext(defaultGlobalSettings));
 
         LdapSessionFactory ldapFactory = new LdapSessionFactory(config, sslService, threadPool);
         LdapRealm ldap =
                 new LdapRealm(LdapRealmSettings.LDAP_TYPE, config, ldapFactory, buildGroupAsRoleMapper(resourceWatcherService), threadPool);
+        ldap.initialize(Collections.singleton(ldap), licenseState);
 
         PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
         ldap.authenticate(new UsernamePasswordToken(VALID_USERNAME, new SecureString(PASSWORD)), future);
@@ -136,12 +150,14 @@ public class LdapRealmTests extends LdapTestCase {
         Settings settings = Settings.builder()
                 .put(buildLdapSettings(ldapUrls(), userTemplate, groupSearchBase, LdapSearchScope.SUB_TREE))
                 .build();
-        RealmConfig config = new RealmConfig("test-ldap-realm", settings, defaultGlobalSettings, TestEnvironment.newEnvironment(defaultGlobalSettings), new ThreadContext(defaultGlobalSettings));
+        RealmConfig config = new RealmConfig("test-ldap-realm", settings, defaultGlobalSettings,
+            TestEnvironment.newEnvironment(defaultGlobalSettings), new ThreadContext(defaultGlobalSettings));
 
         LdapSessionFactory ldapFactory = new LdapSessionFactory(config, sslService, threadPool);
         ldapFactory = spy(ldapFactory);
         LdapRealm ldap =
                 new LdapRealm(LdapRealmSettings.LDAP_TYPE, config, ldapFactory, buildGroupAsRoleMapper(resourceWatcherService), threadPool);
+        ldap.initialize(Collections.singleton(ldap), licenseState);
 
         PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
         ldap.authenticate(new UsernamePasswordToken(VALID_USERNAME, new SecureString(PASSWORD)), future);
@@ -161,12 +177,15 @@ public class LdapRealmTests extends LdapTestCase {
         Settings settings = Settings.builder()
                 .put(buildLdapSettings(ldapUrls(), userTemplate, groupSearchBase, LdapSearchScope.SUB_TREE))
                 .build();
-        RealmConfig config = new RealmConfig("test-ldap-realm", settings, defaultGlobalSettings, TestEnvironment.newEnvironment(defaultGlobalSettings), new ThreadContext(defaultGlobalSettings));
+        RealmConfig config = new RealmConfig("test-ldap-realm", settings, defaultGlobalSettings,
+            TestEnvironment.newEnvironment(defaultGlobalSettings), new ThreadContext(defaultGlobalSettings));
 
         LdapSessionFactory ldapFactory = new LdapSessionFactory(config, sslService, threadPool);
         DnRoleMapper roleMapper = buildGroupAsRoleMapper(resourceWatcherService);
         ldapFactory = spy(ldapFactory);
         LdapRealm ldap = new LdapRealm(LdapRealmSettings.LDAP_TYPE, config, ldapFactory, roleMapper, threadPool);
+        ldap.initialize(Collections.singleton(ldap), licenseState);
+
         PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
         ldap.authenticate(new UsernamePasswordToken(VALID_USERNAME, new SecureString(PASSWORD)), future);
         future.actionGet();
@@ -194,12 +213,15 @@ public class LdapRealmTests extends LdapTestCase {
                 .put(buildLdapSettings(ldapUrls(), userTemplate, groupSearchBase, LdapSearchScope.SUB_TREE))
                 .put(CachingUsernamePasswordRealmSettings.CACHE_TTL_SETTING.getKey(), -1)
                 .build();
-        RealmConfig config = new RealmConfig("test-ldap-realm", settings, defaultGlobalSettings, TestEnvironment.newEnvironment(defaultGlobalSettings), new ThreadContext(defaultGlobalSettings));
+        RealmConfig config = new RealmConfig("test-ldap-realm", settings, defaultGlobalSettings,
+            TestEnvironment.newEnvironment(defaultGlobalSettings), new ThreadContext(defaultGlobalSettings));
 
         LdapSessionFactory ldapFactory = new LdapSessionFactory(config, sslService, threadPool);
         ldapFactory = spy(ldapFactory);
         LdapRealm ldap =
                 new LdapRealm(LdapRealmSettings.LDAP_TYPE, config, ldapFactory, buildGroupAsRoleMapper(resourceWatcherService), threadPool);
+        ldap.initialize(Collections.singleton(ldap), licenseState);
+
         PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
         ldap.authenticate(new UsernamePasswordToken(VALID_USERNAME, new SecureString(PASSWORD)), future);
         future.actionGet();
@@ -211,6 +233,48 @@ public class LdapRealmTests extends LdapTestCase {
         verify(ldapFactory, times(2)).session(anyString(), any(SecureString.class), any(ActionListener.class));
     }
 
+    public void testDelegatedAuthorization() throws Exception {
+        String groupSearchBase = "o=sevenSeas";
+        String userTemplate = VALID_USER_TEMPLATE;
+        final Settings.Builder builder = Settings.builder()
+            .put(buildLdapSettings(ldapUrls(), userTemplate, groupSearchBase, LdapSearchScope.SUB_TREE))
+            .putList(DelegatedAuthorizationSettings.AUTHZ_REALMS.getKey(), "mock_lookup");
+
+        if (randomBoolean()) {
+            // maybe disable caching
+            builder.put(CachingUsernamePasswordRealmSettings.CACHE_TTL_SETTING.getKey(), -1);
+        }
+
+        final Settings realmSettings = builder.build();
+        final Environment env = TestEnvironment.newEnvironment(defaultGlobalSettings);
+        RealmConfig config = new RealmConfig("test-ldap-realm", realmSettings, defaultGlobalSettings, env, threadPool.getThreadContext());
+
+        final LdapSessionFactory ldapFactory = new LdapSessionFactory(config, sslService, threadPool);
+        final DnRoleMapper roleMapper = buildGroupAsRoleMapper(resourceWatcherService);
+        final LdapRealm ldap = new LdapRealm(LdapRealmSettings.LDAP_TYPE, config, ldapFactory, roleMapper, threadPool);
+
+        final MockLookupRealm mockLookup = new MockLookupRealm(new RealmConfig("mock_lookup", Settings.EMPTY, defaultGlobalSettings, env,
+            threadPool.getThreadContext()));
+
+        ldap.initialize(Arrays.asList(ldap, mockLookup), licenseState);
+        mockLookup.initialize(Arrays.asList(ldap, mockLookup), licenseState);
+
+        PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
+        ldap.authenticate(new UsernamePasswordToken(VALID_USERNAME, new SecureString(PASSWORD)), future);
+        final AuthenticationResult result1 = future.actionGet();
+        assertThat(result1.getStatus(), equalTo(AuthenticationResult.Status.CONTINUE));
+        assertThat(result1.getMessage(),
+            equalTo("the principal [" + VALID_USERNAME + "] was authenticated, but no user could be found in realms [mock/mock_lookup]"));
+
+        future = new PlainActionFuture<>();
+        final User fakeUser = new User(VALID_USERNAME, "fake_role");
+        mockLookup.registerUser(fakeUser);
+        ldap.authenticate(new UsernamePasswordToken(VALID_USERNAME, new SecureString(PASSWORD)), future);
+        final AuthenticationResult result2 = future.actionGet();
+        assertThat(result2.getStatus(), equalTo(AuthenticationResult.Status.SUCCESS));
+        assertThat(result2.getUser(), sameInstance(fakeUser));
+    }
+
     public void testLdapRealmSelectsLdapSessionFactory() throws Exception {
         String groupSearchBase = "o=sevenSeas";
         String userTemplate = VALID_USER_TEMPLATE;
@@ -279,7 +343,8 @@ public class LdapRealmTests extends LdapTestCase {
                 .put("group_search.scope", LdapSearchScope.SUB_TREE)
                 .put("ssl.verification_mode", VerificationMode.CERTIFICATE)
                 .build();
-        RealmConfig config = new RealmConfig("test-ldap-realm-user-search", settings, defaultGlobalSettings, TestEnvironment.newEnvironment(defaultGlobalSettings), new ThreadContext(defaultGlobalSettings));
+        RealmConfig config = new RealmConfig("test-ldap-realm-user-search", settings, defaultGlobalSettings,
+            TestEnvironment.newEnvironment(defaultGlobalSettings), new ThreadContext(defaultGlobalSettings));
         IllegalArgumentException e = expectThrows(IllegalArgumentException.class,
                 () -> LdapRealm.sessionFactory(config, null, threadPool, LdapRealmSettings.LDAP_TYPE));
         assertThat(e.getMessage(),
@@ -295,7 +360,8 @@ public class LdapRealmTests extends LdapTestCase {
                 .put("group_search.scope", LdapSearchScope.SUB_TREE)
                 .put("ssl.verification_mode", VerificationMode.CERTIFICATE)
                 .build();
-        RealmConfig config = new RealmConfig("test-ldap-realm-user-search", settings, defaultGlobalSettings, TestEnvironment.newEnvironment(defaultGlobalSettings), new ThreadContext(defaultGlobalSettings));
+        RealmConfig config = new RealmConfig("test-ldap-realm-user-search", settings, defaultGlobalSettings,
+            TestEnvironment.newEnvironment(defaultGlobalSettings), new ThreadContext(defaultGlobalSettings));
         IllegalArgumentException e = expectThrows(IllegalArgumentException.class,
                 () -> LdapRealm.sessionFactory(config, null, threadPool, LdapRealmSettings.LDAP_TYPE));
         assertThat(e.getMessage(),
@@ -312,11 +378,13 @@ public class LdapRealmTests extends LdapTestCase {
                 .put(DnRoleMapperSettings.ROLE_MAPPING_FILE_SETTING.getKey(),
                         getDataPath("/org/elasticsearch/xpack/security/authc/support/role_mapping.yml"))
                 .build();
-        RealmConfig config = new RealmConfig("test-ldap-realm-userdn", settings, defaultGlobalSettings, TestEnvironment.newEnvironment(defaultGlobalSettings), new ThreadContext(defaultGlobalSettings));
+        RealmConfig config = new RealmConfig("test-ldap-realm-userdn", settings, defaultGlobalSettings,
+            TestEnvironment.newEnvironment(defaultGlobalSettings), new ThreadContext(defaultGlobalSettings));
 
         LdapSessionFactory ldapFactory = new LdapSessionFactory(config, sslService, threadPool);
         LdapRealm ldap = new LdapRealm(LdapRealmSettings.LDAP_TYPE, config, ldapFactory,
                 new DnRoleMapper(config, resourceWatcherService), threadPool);
+        ldap.initialize(Collections.singleton(ldap), licenseState);
 
         PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
         ldap.authenticate(new UsernamePasswordToken("Horatio Hornblower", new SecureString(PASSWORD)), future);
@@ -339,10 +407,12 @@ public class LdapRealmTests extends LdapTestCase {
         String groupSearchBase = "o=sevenSeas";
         String userTemplate = VALID_USER_TEMPLATE;
         Settings settings = buildLdapSettings(new String[] { url.toString() }, userTemplate, groupSearchBase, LdapSearchScope.SUB_TREE);
-        RealmConfig config = new RealmConfig("test-ldap-realm", settings, defaultGlobalSettings, TestEnvironment.newEnvironment(defaultGlobalSettings), new ThreadContext(defaultGlobalSettings));
+        RealmConfig config = new RealmConfig("test-ldap-realm", settings, defaultGlobalSettings,
+            TestEnvironment.newEnvironment(defaultGlobalSettings), new ThreadContext(defaultGlobalSettings));
         LdapSessionFactory ldapFactory = new LdapSessionFactory(config, sslService, threadPool);
         LdapRealm ldap = new LdapRealm(LdapRealmSettings.LDAP_TYPE, config, ldapFactory, buildGroupAsRoleMapper(resourceWatcherService),
                 threadPool);
+        ldap.initialize(Collections.singleton(ldap), licenseState);
 
         PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
         ldap.authenticate(new UsernamePasswordToken(VALID_USERNAME, new SecureString(PASSWORD)), future);
@@ -386,6 +456,7 @@ public class LdapRealmTests extends LdapTestCase {
         LdapSessionFactory ldapFactory = new LdapSessionFactory(config, new SSLService(globalSettings, env), threadPool);
         LdapRealm realm = new LdapRealm(LdapRealmSettings.LDAP_TYPE, config, ldapFactory,
                 new DnRoleMapper(config, resourceWatcherService), threadPool);
+        realm.initialize(Collections.singleton(realm), licenseState);
 
         PlainActionFuture<Map<String, Object>> future = new PlainActionFuture<>();
         realm.usageStats(future);

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/pki/PkiRealmTests.java

@@ -12,10 +12,13 @@ import org.elasticsearch.common.settings.MockSecureSettings;
 import org.elasticsearch.common.settings.SecureString;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.CollectionUtils;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.env.TestEnvironment;
+import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationResult;
+import org.elasticsearch.xpack.core.security.authc.Realm;
 import org.elasticsearch.xpack.core.security.authc.RealmConfig;
 import org.elasticsearch.xpack.core.security.authc.RealmSettings;
 import org.elasticsearch.xpack.core.security.authc.pki.PkiRealmSettings;
@@ -23,6 +26,7 @@ import org.elasticsearch.xpack.core.security.authc.support.UsernamePasswordToken
 import org.elasticsearch.xpack.core.security.support.NoOpLogger;
 import org.elasticsearch.xpack.core.security.user.User;
 import org.elasticsearch.xpack.core.ssl.SSLConfigurationSettings;
+import org.elasticsearch.xpack.security.authc.support.MockLookupRealm;
 import org.elasticsearch.xpack.security.authc.support.UserRoleMapper;
 import org.junit.Before;
 import org.mockito.Mockito;
@@ -43,9 +47,11 @@ import java.util.regex.Pattern;
 
 import static org.hamcrest.Matchers.arrayContainingInAnyOrder;
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
+import static org.hamcrest.Matchers.sameInstance;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
@@ -56,12 +62,15 @@ import static org.mockito.Mockito.when;
 public class PkiRealmTests extends ESTestCase {
 
     private Settings globalSettings;
+    private XPackLicenseState licenseState;
 
     @Before
     public void setup() throws Exception {
         globalSettings = Settings.builder()
                 .put("path.home", createTempDir())
                 .build();
+        licenseState = mock(XPackLicenseState.class);
+        when(licenseState.isAuthorizationRealmAllowed()).thenReturn(true);
     }
 
     public void testTokenSupport() {
@@ -98,28 +107,14 @@ public class PkiRealmTests extends ESTestCase {
     }
 
     private void assertSuccessfulAuthentication(Set<String> roles) throws Exception {
-        String dn = "CN=Elasticsearch Test Node,";
+        X509AuthenticationToken token = buildToken();
-        final String expectedUsername = "Elasticsearch Test Node";
+        UserRoleMapper roleMapper = buildRoleMapper(roles, token.dn());
-        X509Certificate certificate = readCert(getDataPath("/org/elasticsearch/xpack/security/transport/ssl/certs/simple/testnode.crt"));
+        PkiRealm realm = buildRealm(roleMapper, Settings.EMPTY);
-        X509AuthenticationToken token = new X509AuthenticationToken(new X509Certificate[] { certificate }, "Elasticsearch Test Node", dn);
-        UserRoleMapper roleMapper = mock(UserRoleMapper.class);
-        PkiRealm realm = new PkiRealm(new RealmConfig("", Settings.EMPTY, globalSettings, TestEnvironment.newEnvironment(globalSettings),
-            new ThreadContext(globalSettings)), roleMapper);
         verify(roleMapper).refreshRealmOnChange(realm);
-        Mockito.doAnswer(invocation -> {
-            final UserRoleMapper.UserData userData = (UserRoleMapper.UserData) invocation.getArguments()[0];
-            final ActionListener<Set<String>> listener = (ActionListener<Set<String>>) invocation.getArguments()[1];
-            if (userData.getDn().equals(dn)) {
-                listener.onResponse(roles);
-            } else {
-                listener.onFailure(new IllegalArgumentException("Expected DN '" + dn + "' but was '" + userData + "'"));
-            }
-            return null;
-        }).when(roleMapper).resolveRoles(any(UserRoleMapper.UserData.class), any(ActionListener.class));
 
-        PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
+        final String expectedUsername = token.principal();
-        realm.authenticate(token, future);
+        final AuthenticationResult result = authenticate(token, realm);
-        final AuthenticationResult result = future.actionGet();
+        final PlainActionFuture<AuthenticationResult> future;
         assertThat(result.getStatus(), is(AuthenticationResult.Status.SUCCESS));
         User user = result.getUser();
         assertThat(user, is(notNullValue()));
@@ -149,17 +144,54 @@ public class PkiRealmTests extends ESTestCase {
         verifyNoMoreInteractions(roleMapper);
     }
 
+    private UserRoleMapper buildRoleMapper(Set<String> roles, String dn) {
+        UserRoleMapper roleMapper = mock(UserRoleMapper.class);
+        Mockito.doAnswer(invocation -> {
+            final UserRoleMapper.UserData userData = (UserRoleMapper.UserData) invocation.getArguments()[0];
+            final ActionListener<Set<String>> listener = (ActionListener<Set<String>>) invocation.getArguments()[1];
+            if (userData.getDn().equals(dn)) {
+                listener.onResponse(roles);
+            } else {
+                listener.onFailure(new IllegalArgumentException("Expected DN '" + dn + "' but was '" + userData + "'"));
+            }
+            return null;
+        }).when(roleMapper).resolveRoles(any(UserRoleMapper.UserData.class), any(ActionListener.class));
+        return roleMapper;
+    }
+
+    private PkiRealm buildRealm(UserRoleMapper roleMapper, Settings realmSettings, Realm... otherRealms) {
+        PkiRealm realm = new PkiRealm(new RealmConfig("", realmSettings, globalSettings, TestEnvironment.newEnvironment(globalSettings),
+            new ThreadContext(globalSettings)), roleMapper);
+        List<Realm> allRealms = CollectionUtils.arrayAsArrayList(otherRealms);
+        allRealms.add(realm);
+        Collections.shuffle(allRealms, random());
+        realm.initialize(allRealms, licenseState);
+        return realm;
+    }
+
+    private X509AuthenticationToken buildToken() throws Exception {
+        X509Certificate certificate = readCert(getDataPath("/org/elasticsearch/xpack/security/transport/ssl/certs/simple/testnode.crt"));
+        return new X509AuthenticationToken(new X509Certificate[]{certificate}, "Elasticsearch Test Node", "CN=Elasticsearch Test Node,");
+    }
+
+    private AuthenticationResult authenticate(X509AuthenticationToken token, PkiRealm realm) {
+        PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
+        realm.authenticate(token, future);
+        return future.actionGet();
+    }
+
     public void testCustomUsernamePattern() throws Exception {
+        ThreadContext threadContext = new ThreadContext(globalSettings);
         X509Certificate certificate = readCert(getDataPath("/org/elasticsearch/xpack/security/transport/ssl/certs/simple/testnode.crt"));
         UserRoleMapper roleMapper = mock(UserRoleMapper.class);
-        PkiRealm realm = new PkiRealm(new RealmConfig("", Settings.builder().put("username_pattern", "OU=(.*?),").build(), globalSettings, TestEnvironment.newEnvironment(globalSettings), new ThreadContext(globalSettings)),
+        PkiRealm realm = new PkiRealm(new RealmConfig("", Settings.builder().put("username_pattern", "OU=(.*?),").build(), globalSettings,
-                roleMapper);
+            TestEnvironment.newEnvironment(globalSettings), threadContext), roleMapper);
+        realm.initialize(Collections.emptyList(), licenseState);
         Mockito.doAnswer(invocation -> {
             ActionListener<Set<String>> listener = (ActionListener<Set<String>>) invocation.getArguments()[1];
             listener.onResponse(Collections.emptySet());
             return null;
         }).when(roleMapper).resolveRoles(any(UserRoleMapper.UserData.class), any(ActionListener.class));
-        ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
         threadContext.putTransient(PkiRealm.PKI_CERT_HEADER_NAME, new X509Certificate[] { certificate });
 
         X509AuthenticationToken token = realm.token(threadContext);
@@ -182,15 +214,16 @@ public class PkiRealmTests extends ESTestCase {
                 .put("truststore.path", getDataPath("/org/elasticsearch/xpack/security/transport/ssl/certs/simple/testnode.jks"))
                 .setSecureSettings(secureSettings)
                 .build();
+        ThreadContext threadContext = new ThreadContext(globalSettings);
         PkiRealm realm = new PkiRealm(new RealmConfig("", settings, globalSettings, TestEnvironment.newEnvironment(globalSettings),
-                new ThreadContext(globalSettings)), roleMapper);
+            threadContext), roleMapper);
+        realm.initialize(Collections.emptyList(), licenseState);
         Mockito.doAnswer(invocation -> {
             ActionListener<Set<String>> listener = (ActionListener<Set<String>>) invocation.getArguments()[1];
             listener.onResponse(Collections.emptySet());
             return null;
         }).when(roleMapper).resolveRoles(any(UserRoleMapper.UserData.class), any(ActionListener.class));
 
-        ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
         threadContext.putTransient(PkiRealm.PKI_CERT_HEADER_NAME, new X509Certificate[] { certificate });
 
         X509AuthenticationToken token = realm.token(threadContext);
@@ -213,15 +246,16 @@ public class PkiRealmTests extends ESTestCase {
                         getDataPath("/org/elasticsearch/xpack/security/transport/ssl/certs/simple/testnode-client-profile.jks"))
                 .setSecureSettings(secureSettings)
                 .build();
+        final ThreadContext threadContext = new ThreadContext(globalSettings);
         PkiRealm realm = new PkiRealm(new RealmConfig("", settings, globalSettings, TestEnvironment.newEnvironment(globalSettings),
-                new ThreadContext(globalSettings)), roleMapper);
+            threadContext), roleMapper);
+        realm.initialize(Collections.emptyList(), licenseState);
         Mockito.doAnswer(invocation -> {
             ActionListener<Set<String>> listener = (ActionListener<Set<String>>) invocation.getArguments()[1];
             listener.onResponse(Collections.emptySet());
             return null;
         }).when(roleMapper).resolveRoles(any(UserRoleMapper.UserData.class), any(ActionListener.class));
 
-        ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
         threadContext.putTransient(PkiRealm.PKI_CERT_HEADER_NAME, new X509Certificate[] { certificate });
 
         X509AuthenticationToken token = realm.token(threadContext);
@@ -307,6 +341,33 @@ public class PkiRealmTests extends ESTestCase {
         assertSettingDeprecationsAndWarnings(new Setting[] { SSLConfigurationSettings.withoutPrefix().legacyTruststorePassword });
     }
 
+    public void testDelegatedAuthorization() throws Exception {
+        final X509AuthenticationToken token = buildToken();
+
+        final MockLookupRealm otherRealm = new MockLookupRealm(new RealmConfig("other_realm", Settings.EMPTY, globalSettings,
+            TestEnvironment.newEnvironment(globalSettings), new ThreadContext(globalSettings)));
+        final User lookupUser = new User(token.principal());
+        otherRealm.registerUser(lookupUser);
+
+        final Settings realmSettings = Settings.builder()
+            .putList("authorization_realms", "other_realm")
+            .build();
+        final UserRoleMapper roleMapper = buildRoleMapper(Collections.emptySet(), token.dn());
+        final PkiRealm pkiRealm = buildRealm(roleMapper, realmSettings, otherRealm);
+
+        AuthenticationResult result = authenticate(token, pkiRealm);
+        assertThat(result.getStatus(), equalTo(AuthenticationResult.Status.SUCCESS));
+        assertThat(result.getUser(), sameInstance(lookupUser));
+
+        // check that the authorizing realm is consulted even for cached principals
+        final User lookupUser2 = new User(token.principal());
+        otherRealm.registerUser(lookupUser2);
+
+        result = authenticate(token, pkiRealm);
+        assertThat(result.getStatus(), equalTo(AuthenticationResult.Status.SUCCESS));
+        assertThat(result.getUser(), sameInstance(lookupUser2));
+    }
+
     static X509Certificate readCert(Path path) throws Exception {
         try (InputStream in = Files.newInputStream(path)) {
             CertificateFactory factory = CertificateFactory.getInstance("X.509");

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/saml/SamlRealmTests.java

@@ -14,18 +14,24 @@ import org.elasticsearch.common.settings.SettingsException;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.TestEnvironment;
+import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.test.http.MockResponse;
 import org.elasticsearch.test.http.MockWebServer;
 import org.elasticsearch.watcher.ResourceWatcherService;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationResult;
+import org.elasticsearch.xpack.core.security.authc.Realm;
 import org.elasticsearch.xpack.core.security.authc.RealmConfig;
 import org.elasticsearch.xpack.core.security.authc.RealmSettings;
 import org.elasticsearch.xpack.core.security.authc.saml.SamlRealmSettings;
+import org.elasticsearch.xpack.core.security.authc.support.DelegatedAuthorizationSettings;
+import org.elasticsearch.xpack.core.security.user.User;
 import org.elasticsearch.xpack.core.ssl.CertParsingUtils;
 import org.elasticsearch.xpack.core.ssl.PemUtils;
 import org.elasticsearch.xpack.core.ssl.SSLService;
 import org.elasticsearch.xpack.core.ssl.TestsSSLService;
+import org.elasticsearch.xpack.security.authc.support.MockLookupRealm;
 import org.elasticsearch.xpack.security.authc.support.UserRoleMapper;
+import org.hamcrest.Matchers;
 import org.junit.Before;
 import org.mockito.Mockito;
 import org.opensaml.saml.common.xml.SAMLConstants;
@@ -71,6 +77,7 @@ import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 /**
  * Basic unit tests for the SAMLRealm
@@ -83,9 +90,16 @@ public class SamlRealmTests extends SamlTestCase {
     private static final String REALM_NAME = "my-saml";
     private static final String REALM_SETTINGS_PREFIX = "xpack.security.authc.realms." + REALM_NAME;
 
+    private Settings globalSettings;
+    private Environment env;
+    private ThreadContext threadContext;
+
     @Before
-    public void initRealm() throws PrivilegedActionException {
+    public void setupEnv() throws PrivilegedActionException {
         SamlUtils.initialize(logger);
+        globalSettings = Settings.builder().put("path.home", createTempDir()).build();
+        env = TestEnvironment.newEnvironment(globalSettings);
+        threadContext = new ThreadContext(globalSettings);
     }
 
     public void testReadIdpMetadataFromFile() throws Exception {
@@ -140,15 +154,70 @@ public class SamlRealmTests extends SamlTestCase {
     }
 
     public void testAuthenticateWithRoleMapping() throws Exception {
+        final UserRoleMapper roleMapper = mock(UserRoleMapper.class);
+        AtomicReference<UserRoleMapper.UserData> userData = new AtomicReference<>();
+        Mockito.doAnswer(invocation -> {
+            assert invocation.getArguments().length == 2;
+            userData.set((UserRoleMapper.UserData) invocation.getArguments()[0]);
+            ActionListener<Set<String>> listener = (ActionListener<Set<String>>) invocation.getArguments()[1];
+            listener.onResponse(Collections.singleton("superuser"));
+            return null;
+        }).when(roleMapper).resolveRoles(any(UserRoleMapper.UserData.class), any(ActionListener.class));
+
         final boolean useNameId = randomBoolean();
         final boolean principalIsEmailAddress = randomBoolean();
         final Boolean populateUserMetadata = randomFrom(Boolean.TRUE, Boolean.FALSE, null);
+
+        AuthenticationResult result = performAuthentication(roleMapper, useNameId, principalIsEmailAddress, populateUserMetadata, false);
+        assertThat(result.getUser().roles(), arrayContainingInAnyOrder("superuser"));
+        if (populateUserMetadata == Boolean.FALSE) {
+            // TODO : "saml_nameid" should be null too, but the logout code requires it for now.
+            assertThat(result.getUser().metadata().get("saml_uid"), nullValue());
+        } else {
+            final String nameIdValue = principalIsEmailAddress ? "clint.barton@shield.gov" : "clint.barton";
+            final String uidValue = principalIsEmailAddress ? "cbarton@shield.gov" : "cbarton";
+            assertThat(result.getUser().metadata().get("saml_nameid"), equalTo(nameIdValue));
+            assertThat(result.getUser().metadata().get("saml_uid"), instanceOf(Iterable.class));
+            assertThat((Iterable<?>) result.getUser().metadata().get("saml_uid"), contains(uidValue));
+        }
+
+        assertThat(userData.get().getUsername(), equalTo(useNameId ? "clint.barton" : "cbarton"));
+        assertThat(userData.get().getGroups(), containsInAnyOrder("avengers", "shield"));
+    }
+
+    public void testAuthenticateWithAuthorizingRealm() throws Exception {
         final UserRoleMapper roleMapper = mock(UserRoleMapper.class);
+        Mockito.doAnswer(invocation -> {
+            assert invocation.getArguments().length == 2;
+            ActionListener<Set<String>> listener = (ActionListener<Set<String>>) invocation.getArguments()[1];
+            listener.onFailure(new RuntimeException("Role mapping should not be called"));
+            return null;
+        }).when(roleMapper).resolveRoles(any(UserRoleMapper.UserData.class), any(ActionListener.class));
+
+        final boolean useNameId = randomBoolean();
+        final boolean principalIsEmailAddress = randomBoolean();
+
+        AuthenticationResult result = performAuthentication(roleMapper, useNameId, principalIsEmailAddress, null, true);
+        assertThat(result.getUser().roles(), arrayContainingInAnyOrder("lookup_user_role"));
+        assertThat(result.getUser().fullName(), equalTo("Clinton Barton"));
+        assertThat(result.getUser().metadata().entrySet(), Matchers.iterableWithSize(1));
+        assertThat(result.getUser().metadata().get("is_lookup"), Matchers.equalTo(true));
+    }
+
+    private AuthenticationResult performAuthentication(UserRoleMapper roleMapper, boolean useNameId, boolean principalIsEmailAddress,
+                                                       Boolean populateUserMetadata, boolean useAuthorizingRealm) throws Exception {
         final EntityDescriptor idp = mockIdp();
         final SpConfiguration sp = new SpConfiguration("<sp>", "https://saml/", null, null, null, Collections.emptyList());
         final SamlAuthenticator authenticator = mock(SamlAuthenticator.class);
         final SamlLogoutRequestHandler logoutHandler = mock(SamlLogoutRequestHandler.class);
 
+        final String userPrincipal = useNameId ? "clint.barton" : "cbarton";
+        final String nameIdValue = principalIsEmailAddress ? "clint.barton@shield.gov" : "clint.barton";
+        final String uidValue = principalIsEmailAddress ? "cbarton@shield.gov" : "cbarton";
+
+        final MockLookupRealm lookupRealm = new MockLookupRealm(
+            new RealmConfig("mock_lookup", Settings.EMPTY,globalSettings, env, threadContext));
+
         final Settings.Builder settingsBuilder = Settings.builder()
                 .put(SamlRealmSettings.PRINCIPAL_ATTRIBUTE.name(), useNameId ? "nameid" : "uid")
                 .put(SamlRealmSettings.GROUPS_ATTRIBUTE.name(), "groups")
@@ -161,15 +230,20 @@ public class SamlRealmTests extends SamlTestCase {
         if (populateUserMetadata != null) {
             settingsBuilder.put(SamlRealmSettings.POPULATE_USER_METADATA.getKey(), populateUserMetadata.booleanValue());
         }
+        if (useAuthorizingRealm) {
+            settingsBuilder.putList(DelegatedAuthorizationSettings.AUTHZ_REALMS.getKey(), lookupRealm.name());
+            lookupRealm.registerUser(new User(userPrincipal, new String[]{ "lookup_user_role" }, "Clinton Barton", "cbarton@shield.gov",
+                Collections.singletonMap("is_lookup", true), true));
+        }
+
         final Settings realmSettings = settingsBuilder.build();
-
         final RealmConfig config = realmConfigFromRealmSettings(realmSettings);
-
         final SamlRealm realm = new SamlRealm(config, roleMapper, authenticator, logoutHandler, () -> idp, sp);
+
+        initializeRealms(realm, lookupRealm);
+
         final SamlToken token = new SamlToken(new byte[0], Collections.singletonList("<id>"));
 
-        final String nameIdValue = principalIsEmailAddress ? "clint.barton@shield.gov" : "clint.barton";
-        final String uidValue = principalIsEmailAddress ? "cbarton@shield.gov" : "cbarton";
         final SamlAttributes attributes = new SamlAttributes(
                 new SamlNameId(NameIDType.PERSISTENT, nameIdValue, idp.getEntityID(), sp.getEntityId(), null),
                 randomAlphaOfLength(16),
@@ -178,36 +252,27 @@ public class SamlRealmTests extends SamlTestCase {
                         new SamlAttributes.SamlAttribute("urn:oid:1.3.6.1.4.1.5923.1.5.1.1", "groups", Arrays.asList("avengers", "shield")),
                         new SamlAttributes.SamlAttribute("urn:oid:0.9.2342.19200300.100.1.3", "mail", Arrays.asList("cbarton@shield.gov"))
                 ));
-        Mockito.when(authenticator.authenticate(token)).thenReturn(attributes);
+        when(authenticator.authenticate(token)).thenReturn(attributes);
-
-        AtomicReference<UserRoleMapper.UserData> userData = new AtomicReference<>();
-        Mockito.doAnswer(invocation -> {
-            assert invocation.getArguments().length == 2;
-            userData.set((UserRoleMapper.UserData) invocation.getArguments()[0]);
-            ActionListener<Set<String>> listener = (ActionListener<Set<String>>) invocation.getArguments()[1];
-            listener.onResponse(Collections.singleton("superuser"));
-            return null;
-        }).when(roleMapper).resolveRoles(any(UserRoleMapper.UserData.class), any(ActionListener.class));
 
         final PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
         realm.authenticate(token, future);
         final AuthenticationResult result = future.get();
         assertThat(result, notNullValue());
         assertThat(result.getStatus(), equalTo(AuthenticationResult.Status.SUCCESS));
-        assertThat(result.getUser().principal(), equalTo(useNameId ? "clint.barton" : "cbarton"));
+        assertThat(result.getUser().principal(), equalTo(userPrincipal));
         assertThat(result.getUser().email(), equalTo("cbarton@shield.gov"));
-        assertThat(result.getUser().roles(), arrayContainingInAnyOrder("superuser"));
-        if (populateUserMetadata == Boolean.FALSE) {
-            // TODO : "saml_nameid" should be null too, but the logout code requires it for now.
-            assertThat(result.getUser().metadata().get("saml_uid"), nullValue());
-        } else {
-            assertThat(result.getUser().metadata().get("saml_nameid"), equalTo(nameIdValue));
-            assertThat(result.getUser().metadata().get("saml_uid"), instanceOf(Iterable.class));
-            assertThat((Iterable<?>) result.getUser().metadata().get("saml_uid"), contains(uidValue));
-        }
 
-        assertThat(userData.get().getUsername(), equalTo(useNameId ? "clint.barton" : "cbarton"));
+        return result;
-        assertThat(userData.get().getGroups(), containsInAnyOrder("avengers", "shield"));
+    }
+
+    private void initializeRealms(Realm... realms) {
+        XPackLicenseState licenseState = mock(XPackLicenseState.class);
+        when(licenseState.isAuthorizationRealmAllowed()).thenReturn(true);
+
+        final List<Realm> realmList = Arrays.asList(realms);
+        for (Realm realm : realms) {
+            realm.initialize(realmList, licenseState);
+        }
     }
 
     public void testAttributeSelectionWithRegex() throws Exception {
@@ -291,7 +356,7 @@ public class SamlRealmTests extends SamlTestCase {
                     Collections.singletonList(
                             new SamlAttributes.SamlAttribute("urn:oid:0.9.2342.19200300.100.1.3", "mail", Collections.singletonList(mail))
                     ));
-            Mockito.when(authenticator.authenticate(token)).thenReturn(attributes);
+            when(authenticator.authenticate(token)).thenReturn(attributes);
 
             final PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
             realm.authenticate(token, future);
@@ -515,8 +580,8 @@ public class SamlRealmTests extends SamlTestCase {
         final EntityDescriptor idp = mockIdp();
         final IDPSSODescriptor role = mock(IDPSSODescriptor.class);
         final SingleLogoutService slo = SamlUtils.buildObject(SingleLogoutService.class, SingleLogoutService.DEFAULT_ELEMENT_NAME);
-        Mockito.when(idp.getRoleDescriptors(IDPSSODescriptor.DEFAULT_ELEMENT_NAME)).thenReturn(Collections.singletonList(role));
+        when(idp.getRoleDescriptors(IDPSSODescriptor.DEFAULT_ELEMENT_NAME)).thenReturn(Collections.singletonList(role));
-        Mockito.when(role.getSingleLogoutServices()).thenReturn(Collections.singletonList(slo));
+        when(role.getSingleLogoutServices()).thenReturn(Collections.singletonList(slo));
         slo.setBinding(SAMLConstants.SAML2_REDIRECT_BINDING_URI);
         slo.setLocation("https://logout.saml/");
 
@@ -553,7 +618,7 @@ public class SamlRealmTests extends SamlTestCase {
 
     private EntityDescriptor mockIdp() {
         final EntityDescriptor descriptor = mock(EntityDescriptor.class);
-        Mockito.when(descriptor.getEntityID()).thenReturn("https://idp.saml/");
+        when(descriptor.getEntityID()).thenReturn("https://idp.saml/");
         return descriptor;
     }
 
@@ -585,9 +650,7 @@ public class SamlRealmTests extends SamlTestCase {
     }
 
     private RealmConfig realmConfigFromRealmSettings(Settings realmSettings) {
-        final Settings globalSettings = Settings.builder().put("path.home", createTempDir()).build();
+        return new RealmConfig(REALM_NAME, realmSettings, globalSettings, env, threadContext);
-        final Environment env = TestEnvironment.newEnvironment(globalSettings);
-        return new RealmConfig(REALM_NAME, realmSettings, globalSettings, env, new ThreadContext(globalSettings));
     }
 
     private RealmConfig realmConfigFromGlobalSettings(Settings globalSettings) {

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/support/CachingUsernamePasswordRealmTests.java

@@ -31,6 +31,7 @@ import java.util.List;
 import java.util.Locale;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static java.util.Collections.emptyMap;
 import static org.hamcrest.Matchers.arrayContaining;
@@ -39,6 +40,7 @@ import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
 import static org.hamcrest.Matchers.sameInstance;
 
@@ -341,6 +343,33 @@ public class CachingUsernamePasswordRealmTests extends ESTestCase {
         assertThat(e.getMessage(), containsString("lookup exception"));
     }
 
+    public void testReturnDifferentObjectFromCache() throws Exception {
+        final AtomicReference<User> userArg = new AtomicReference<>();
+        final AtomicReference<AuthenticationResult> result = new AtomicReference<>();
+        Realm realm = new AlwaysAuthenticateCachingRealm(globalSettings, threadPool) {
+            @Override
+            protected void handleCachedAuthentication(User user, ActionListener<AuthenticationResult> listener) {
+                userArg.set(user);
+                listener.onResponse(result.get());
+            }
+        };
+        PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
+        realm.authenticate(new UsernamePasswordToken("user", new SecureString("pass")), future);
+        final AuthenticationResult result1 = future.actionGet();
+        assertThat(result1, notNullValue());
+        assertThat(result1.getUser(), notNullValue());
+        assertThat(result1.getUser().principal(), equalTo("user"));
+
+        final AuthenticationResult result2 = AuthenticationResult.success(new User("user"));
+        result.set(result2);
+
+        future = new PlainActionFuture<>();
+        realm.authenticate(new UsernamePasswordToken("user", new SecureString("pass")), future);
+        final AuthenticationResult result3 = future.actionGet();
+        assertThat(result3, sameInstance(result2));
+        assertThat(userArg.get(), sameInstance(result1.getUser()));
+    }
+
     public void testSingleAuthPerUserLimit() throws Exception {
         final String username = "username";
         final SecureString password = SecuritySettingsSourceField.TEST_PASSWORD_SECURE_STRING;

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/support/DelegatedAuthorizationSupportTests.java

@@ -0,0 +1,189 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.security.authc.support;
+
+import org.elasticsearch.ElasticsearchSecurityException;
+import org.elasticsearch.action.support.PlainActionFuture;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.env.Environment;
+import org.elasticsearch.env.TestEnvironment;
+import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.core.security.authc.AuthenticationResult;
+import org.elasticsearch.xpack.core.security.authc.Realm;
+import org.elasticsearch.xpack.core.security.authc.RealmConfig;
+import org.elasticsearch.xpack.core.security.user.User;
+import org.junit.Before;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+import static org.elasticsearch.common.Strings.collectionToDelimitedString;
+import static org.hamcrest.Matchers.arrayContaining;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.nullValue;
+import static org.hamcrest.Matchers.sameInstance;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class DelegatedAuthorizationSupportTests extends ESTestCase {
+
+    private List<MockLookupRealm> realms;
+    private Settings globalSettings;
+    private ThreadContext threadContext;
+    private Environment env;
+
+    @Before
+    public void setupRealms() {
+        globalSettings = Settings.builder()
+            .put("path.home", createTempDir())
+            .build();
+        env = TestEnvironment.newEnvironment(globalSettings);
+        threadContext = new ThreadContext(globalSettings);
+
+        final int realmCount = randomIntBetween(5, 9);
+        realms = new ArrayList<>(realmCount);
+        for (int i = 1; i <= realmCount; i++) {
+            realms.add(new MockLookupRealm(buildRealmConfig("lookup-" + i, Settings.EMPTY)));
+        }
+        shuffle(realms);
+    }
+
+    private <T> List<T> shuffle(List<T> list) {
+        Collections.shuffle(list, random());
+        return list;
+    }
+
+    private RealmConfig buildRealmConfig(String name, Settings settings) {
+        return new RealmConfig(name, settings, globalSettings, env, threadContext);
+    }
+
+    public void testEmptyDelegationList() throws ExecutionException, InterruptedException {
+        final XPackLicenseState license = getLicenseState(true);
+        final DelegatedAuthorizationSupport das = new DelegatedAuthorizationSupport(realms, buildRealmConfig("r", Settings.EMPTY), license);
+        assertThat(das.hasDelegation(), equalTo(false));
+        final PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
+        das.resolve("any", future);
+        final AuthenticationResult result = future.get();
+        assertThat(result.getStatus(), equalTo(AuthenticationResult.Status.CONTINUE));
+        assertThat(result.getUser(), nullValue());
+        assertThat(result.getMessage(), equalTo("No [authorization_realms] have been configured"));
+    }
+
+    public void testMissingRealmInDelegationList() {
+        final XPackLicenseState license = getLicenseState(true);
+        final Settings settings = Settings.builder()
+            .putList("authorization_realms", "no-such-realm")
+            .build();
+        final IllegalArgumentException ex = expectThrows(IllegalArgumentException.class, () ->
+            new DelegatedAuthorizationSupport(realms, buildRealmConfig("r", settings), license)
+        );
+        assertThat(ex.getMessage(), equalTo("configured authorization realm [no-such-realm] does not exist (or is not enabled)"));
+    }
+
+    public void testDelegationChainsAreRejected() {
+        final XPackLicenseState license = getLicenseState(true);
+        final Settings settings = Settings.builder()
+            .putList("authorization_realms", "lookup-1", "lookup-2", "lookup-3")
+            .build();
+        globalSettings = Settings.builder()
+            .put(globalSettings)
+            .putList("xpack.security.authc.realms.lookup-2.authorization_realms", "lookup-1")
+            .build();
+        final IllegalArgumentException ex = expectThrows(IllegalArgumentException.class, () ->
+            new DelegatedAuthorizationSupport(realms, buildRealmConfig("realm1", settings), license)
+        );
+        assertThat(ex.getMessage(),
+            equalTo("cannot use realm [mock/lookup-2] as an authorization realm - it is already delegating authorization to [[lookup-1]]"));
+    }
+
+    public void testMatchInDelegationList() throws Exception {
+        final XPackLicenseState license = getLicenseState(true);
+        final List<MockLookupRealm> useRealms = shuffle(randomSubsetOf(randomIntBetween(1, realms.size()), realms));
+        final Settings settings = Settings.builder()
+            .putList("authorization_realms", useRealms.stream().map(Realm::name).collect(Collectors.toList()))
+            .build();
+        final User user = new User("my_user");
+        randomFrom(useRealms).registerUser(user);
+        final DelegatedAuthorizationSupport das = new DelegatedAuthorizationSupport(realms, buildRealmConfig("r", settings), license);
+        assertThat(das.hasDelegation(), equalTo(true));
+        final PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
+        das.resolve("my_user", future);
+        final AuthenticationResult result = future.get();
+        assertThat(result.getStatus(), equalTo(AuthenticationResult.Status.SUCCESS));
+        assertThat(result.getUser(), sameInstance(user));
+    }
+
+    public void testRealmsAreOrdered() throws Exception {
+        final XPackLicenseState license = getLicenseState(true);
+        final List<MockLookupRealm> useRealms = shuffle(randomSubsetOf(randomIntBetween(3, realms.size()), realms));
+        final List<String> names = useRealms.stream().map(Realm::name).collect(Collectors.toList());
+        final Settings settings = Settings.builder()
+            .putList("authorization_realms", names)
+            .build();
+        final List<User> users = new ArrayList<>(names.size());
+        final String username = randomAlphaOfLength(8);
+        for (MockLookupRealm r : useRealms) {
+            final User user = new User(username, "role_" + r.name());
+            users.add(user);
+            r.registerUser(user);
+        }
+
+        final DelegatedAuthorizationSupport das = new DelegatedAuthorizationSupport(realms, buildRealmConfig("r", settings), license);
+        assertThat(das.hasDelegation(), equalTo(true));
+        final PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
+        das.resolve(username, future);
+        final AuthenticationResult result = future.get();
+        assertThat(result.getStatus(), equalTo(AuthenticationResult.Status.SUCCESS));
+        assertThat(result.getUser(), sameInstance(users.get(0)));
+        assertThat(result.getUser().roles(), arrayContaining("role_" + useRealms.get(0).name()));
+    }
+
+    public void testNoMatchInDelegationList() throws Exception {
+        final XPackLicenseState license = getLicenseState(true);
+        final List<MockLookupRealm> useRealms = shuffle(randomSubsetOf(randomIntBetween(1, realms.size()), realms));
+        final Settings settings = Settings.builder()
+            .putList("authorization_realms", useRealms.stream().map(Realm::name).collect(Collectors.toList()))
+            .build();
+        final DelegatedAuthorizationSupport das = new DelegatedAuthorizationSupport(realms, buildRealmConfig("r", settings), license);
+        assertThat(das.hasDelegation(), equalTo(true));
+        final PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
+        das.resolve("my_user", future);
+        final AuthenticationResult result = future.get();
+        assertThat(result.getStatus(), equalTo(AuthenticationResult.Status.CONTINUE));
+        assertThat(result.getUser(), nullValue());
+        assertThat(result.getMessage(), equalTo("the principal [my_user] was authenticated, but no user could be found in realms [" +
+            collectionToDelimitedString(useRealms.stream().map(Realm::toString).collect(Collectors.toList()), ",") + "]"));
+    }
+
+    public void testLicenseRejection() throws Exception {
+        final XPackLicenseState license = getLicenseState(false);
+        final Settings settings = Settings.builder()
+            .putList("authorization_realms", realms.get(0).name())
+            .build();
+        final DelegatedAuthorizationSupport das = new DelegatedAuthorizationSupport(realms, buildRealmConfig("r", settings), license);
+        assertThat(das.hasDelegation(), equalTo(true));
+        final PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
+        das.resolve("my_user", future);
+        final AuthenticationResult result = future.get();
+        assertThat(result.getStatus(), equalTo(AuthenticationResult.Status.CONTINUE));
+        assertThat(result.getUser(), nullValue());
+        assertThat(result.getMessage(), equalTo("authorization_realms are not permitted"));
+        assertThat(result.getException(), instanceOf(ElasticsearchSecurityException.class));
+        assertThat(result.getException().getMessage(), equalTo("current license is non-compliant for [authorization_realms]"));
+    }
+
+    private XPackLicenseState getLicenseState(boolean authzRealmsAllowed) {
+        final XPackLicenseState license = mock(XPackLicenseState.class);
+        when(license.isAuthorizationRealmAllowed()).thenReturn(authzRealmsAllowed);
+        return license;
+    }
+}

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/support/MockLookupRealm.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.security.authc.support;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.xpack.core.security.authc.AuthenticationResult;
+import org.elasticsearch.xpack.core.security.authc.AuthenticationToken;
+import org.elasticsearch.xpack.core.security.authc.Realm;
+import org.elasticsearch.xpack.core.security.authc.RealmConfig;
+import org.elasticsearch.xpack.core.security.user.User;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class MockLookupRealm extends Realm {
+
+    private final Map<String, User> lookup;
+
+    public MockLookupRealm(RealmConfig config) {
+        super("mock", config);
+        lookup = new HashMap<>();
+    }
+
+    public void registerUser(User user) {
+        this.lookup.put(user.principal(), user);
+    }
+
+    @Override
+    public boolean supports(AuthenticationToken token) {
+        return false;
+    }
+
+    @Override
+    public AuthenticationToken token(ThreadContext context) {
+        return null;
+    }
+
+    @Override
+    public void authenticate(AuthenticationToken token, ActionListener<AuthenticationResult> listener) {
+        listener.onResponse(AuthenticationResult.notHandled());
+    }
+
+    @Override
+    public void lookupUser(String username, ActionListener<User> listener) {
+        listener.onResponse(lookup.get(username));
+    }
+}

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/support/RealmUserLookupTests.java

@@ -0,0 +1,128 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.security.authc.support;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.PlainActionFuture;
+import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.env.Environment;
+import org.elasticsearch.env.TestEnvironment;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.core.security.authc.AuthenticationResult;
+import org.elasticsearch.xpack.core.security.authc.AuthenticationToken;
+import org.elasticsearch.xpack.core.security.authc.Realm;
+import org.elasticsearch.xpack.core.security.authc.RealmConfig;
+import org.elasticsearch.xpack.core.security.user.User;
+import org.junit.Before;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.nullValue;
+import static org.hamcrest.Matchers.sameInstance;
+
+public class RealmUserLookupTests extends ESTestCase {
+
+    private Settings globalSettings;
+    private ThreadContext threadContext;
+    private Environment env;
+
+    @Before
+    public void setup() {
+        globalSettings = Settings.builder()
+            .put("path.home", createTempDir())
+            .build();
+        env = TestEnvironment.newEnvironment(globalSettings);
+        threadContext = new ThreadContext(globalSettings);
+    }
+
+    public void testNoRealms() throws Exception {
+        final RealmUserLookup lookup = new RealmUserLookup(Collections.emptyList(), threadContext);
+        final PlainActionFuture<Tuple<User, Realm>> listener = new PlainActionFuture<>();
+        lookup.lookup(randomAlphaOfLengthBetween(3, 12), listener);
+        final Tuple<User, Realm> tuple = listener.get();
+        assertThat(tuple, nullValue());
+    }
+
+    public void testUserFound() throws Exception {
+        final List<MockLookupRealm> realms = buildRealms(randomIntBetween(5, 9));
+        final RealmUserLookup lookup = new RealmUserLookup(realms, threadContext);
+
+        final MockLookupRealm matchRealm = randomFrom(realms);
+        final User user = new User(randomAlphaOfLength(5));
+        matchRealm.registerUser(user);
+
+        final PlainActionFuture<Tuple<User, Realm>> listener = new PlainActionFuture<>();
+        lookup.lookup(user.principal(), listener);
+        final Tuple<User, Realm> tuple = listener.get();
+        assertThat(tuple, notNullValue());
+        assertThat(tuple.v1(), notNullValue());
+        assertThat(tuple.v1(), sameInstance(user));
+        assertThat(tuple.v2(), notNullValue());
+        assertThat(tuple.v2(), sameInstance(matchRealm));
+    }
+
+    public void testUserNotFound() throws Exception {
+        final List<MockLookupRealm> realms = buildRealms(randomIntBetween(5, 9));
+        final RealmUserLookup lookup = new RealmUserLookup(realms, threadContext);
+
+        final String username = randomAlphaOfLength(5);
+
+        final PlainActionFuture<Tuple<User, Realm>> listener = new PlainActionFuture<>();
+        lookup.lookup(username, listener);
+        final Tuple<User, Realm> tuple = listener.get();
+        assertThat(tuple, nullValue());
+    }
+
+    public void testRealmException() {
+        final Realm realm = new Realm("test", new RealmConfig("test", Settings.EMPTY, globalSettings, env, threadContext)) {
+            @Override
+            public boolean supports(AuthenticationToken token) {
+                return false;
+            }
+
+            @Override
+            public AuthenticationToken token(ThreadContext context) {
+                return null;
+            }
+
+            @Override
+            public void authenticate(AuthenticationToken token, ActionListener<AuthenticationResult> listener) {
+                listener.onResponse(AuthenticationResult.notHandled());
+            }
+
+            @Override
+            public void lookupUser(String username, ActionListener<User> listener) {
+                listener.onFailure(new RuntimeException("FAILURE"));
+            }
+        };
+        final RealmUserLookup lookup = new RealmUserLookup(Collections.singletonList(realm), threadContext);
+        final PlainActionFuture<Tuple<User, Realm>> listener = new PlainActionFuture<>();
+        lookup.lookup("anyone", listener);
+        final RuntimeException e = expectThrows(RuntimeException.class, listener::actionGet);
+        assertThat(e.getMessage(), equalTo("FAILURE"));
+    }
+
+    private List<MockLookupRealm> buildRealms(int realmCount) {
+        final List<MockLookupRealm> realms = new ArrayList<>(realmCount);
+        for (int i = 1; i <= realmCount; i++) {
+            final RealmConfig config = new RealmConfig("lookup-" + i, Settings.EMPTY, globalSettings, env, threadContext);
+            final MockLookupRealm realm = new MockLookupRealm(config);
+            for (int j = 0; j < 5; j++) {
+                realm.registerUser(new User(randomAlphaOfLengthBetween(6, 12)));
+            }
+            realms.add(realm);
+        }
+        Collections.shuffle(realms, random());
+        return realms;
+    }
+}

x-pack/qa/saml-idp-tests/build.gradle

@@ -37,17 +37,30 @@ integTestCluster {
   setting 'xpack.security.authc.token.enabled', 'true'
   setting 'xpack.security.authc.realms.file.type', 'file'
   setting 'xpack.security.authc.realms.file.order', '0'
+  // SAML realm 1 (no authorization_realms)
   setting 'xpack.security.authc.realms.shibboleth.type', 'saml'
   setting 'xpack.security.authc.realms.shibboleth.order', '1'
   setting 'xpack.security.authc.realms.shibboleth.idp.entity_id', 'https://test.shibboleth.elastic.local/'
   setting 'xpack.security.authc.realms.shibboleth.idp.metadata.path', 'idp-metadata.xml'
-  setting 'xpack.security.authc.realms.shibboleth.sp.entity_id', 'http://mock.http.elastic.local/'
+  setting 'xpack.security.authc.realms.shibboleth.sp.entity_id', 'http://mock1.http.elastic.local/'
   // The port in the ACS URL is fake - the test will bind the mock webserver
   // to a random port and then whenever it needs to connect to a URL on the
   // mock webserver it will replace 54321 with the real port
-  setting 'xpack.security.authc.realms.shibboleth.sp.acs', 'http://localhost:54321/saml/acs'
+  setting 'xpack.security.authc.realms.shibboleth.sp.acs', 'http://localhost:54321/saml/acs1'
   setting 'xpack.security.authc.realms.shibboleth.attributes.principal', 'uid'
   setting 'xpack.security.authc.realms.shibboleth.attributes.name', 'urn:oid:2.5.4.3'
+  // SAML realm 2 (uses authorization_realms)
+  setting 'xpack.security.authc.realms.shibboleth_native.type', 'saml'
+  setting 'xpack.security.authc.realms.shibboleth_native.order', '2'
+  setting 'xpack.security.authc.realms.shibboleth_native.idp.entity_id', 'https://test.shibboleth.elastic.local/'
+  setting 'xpack.security.authc.realms.shibboleth_native.idp.metadata.path', 'idp-metadata.xml'
+  setting 'xpack.security.authc.realms.shibboleth_native.sp.entity_id', 'http://mock2.http.elastic.local/'
+  setting 'xpack.security.authc.realms.shibboleth_native.sp.acs', 'http://localhost:54321/saml/acs2'
+  setting 'xpack.security.authc.realms.shibboleth_native.attributes.principal', 'uid'
+  setting 'xpack.security.authc.realms.shibboleth_native.authorization_realms', 'native'
+  setting 'xpack.security.authc.realms.native.type', 'native'
+  setting 'xpack.security.authc.realms.native.order', '3'
+
   setting 'xpack.ml.enabled', 'false'
 
   extraConfigFile 'idp-metadata.xml', idpFixtureProject.file("src/main/resources/provision/generated/idp-metadata.xml")

x-pack/qa/saml-idp-tests/src/test/java/org/elasticsearch/xpack/security/authc/saml/SamlAuthenticationIT.java

@@ -42,6 +42,8 @@ import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.common.CheckedFunction;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.collect.MapBuilder;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.io.Streams;
 import org.elasticsearch.common.settings.SecureString;
@@ -49,6 +51,7 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.mocksocket.MockHttpServer;
 import org.elasticsearch.test.rest.ESRestTestCase;
@@ -65,7 +68,6 @@ import javax.net.ssl.KeyManager;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.TrustManager;
 import javax.net.ssl.X509ExtendedTrustManager;
-
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
@@ -102,13 +104,16 @@ import static org.hamcrest.Matchers.startsWith;
 public class SamlAuthenticationIT extends ESRestTestCase {
 
     private static final String SP_LOGIN_PATH = "/saml/login";
-    private static final String SP_ACS_PATH = "/saml/acs";
+    private static final String SP_ACS_PATH_1 = "/saml/acs1";
+    private static final String SP_ACS_PATH_2 = "/saml/acs2";
     private static final String SAML_RESPONSE_FIELD = "SAMLResponse";
     private static final String REQUEST_ID_COOKIE = "saml-request-id";
 
     private static final String KIBANA_PASSWORD = "K1b@na K1b@na K1b@na";
     private static HttpServer httpServer;
 
+    private URI acs;
+
     @BeforeClass
     public static void setupHttpServer() throws IOException {
         InetSocketAddress address = new InetSocketAddress(InetAddress.getLoopbackAddress().getHostAddress(), 0);
@@ -133,7 +138,8 @@ public class SamlAuthenticationIT extends ESRestTestCase {
     @Before
     public void setupHttpContext() {
         httpServer.createContext(SP_LOGIN_PATH, wrapFailures(this::httpLogin));
-        httpServer.createContext(SP_ACS_PATH, wrapFailures(this::httpAcs));
+        httpServer.createContext(SP_ACS_PATH_1, wrapFailures(this::httpAcs));
+        httpServer.createContext(SP_ACS_PATH_2, wrapFailures(this::httpAcs));
     }
 
     /**
@@ -157,7 +163,8 @@ public class SamlAuthenticationIT extends ESRestTestCase {
     @After
     public void clearHttpContext() {
         httpServer.removeContext(SP_LOGIN_PATH);
-        httpServer.removeContext(SP_ACS_PATH);
+        httpServer.removeContext(SP_ACS_PATH_1);
+        httpServer.removeContext(SP_ACS_PATH_2);
     }
 
     @Override
@@ -202,6 +209,21 @@ public class SamlAuthenticationIT extends ESRestTestCase {
         adminClient().performRequest(request);
     }
 
+    /**
+     * Create a native user for "thor" that is used for user-lookup (authorizing realms)
+     */
+    @Before
+    public void setupNativeUser() throws IOException {
+        final Map<String, Object> body = MapBuilder.<String, Object>newMapBuilder()
+            .put("roles", Collections.singletonList("kibana_dashboard_only_user"))
+            .put("full_name", "Thor Son of Odin")
+            .put("password", randomAlphaOfLengthBetween(8, 16))
+            .put("metadata", Collections.singletonMap("is_native", true))
+            .map();
+        final Response response = adminClient().performRequest(buildRequest("PUT", "/_xpack/security/user/thor", body));
+        assertOK(response);
+    }
+
     /**
      * Tests that a user can login via a SAML idp:
      * It uses:
@@ -218,7 +240,24 @@ public class SamlAuthenticationIT extends ESRestTestCase {
      * <li>Uses that token to verify the user details</li>
      * </ol>
      */
-    public void testLoginUser() throws Exception {
+    public void testLoginUserWithSamlRoleMapping() throws Exception {
+        // this ACS comes from the config in build.gradle
+        final Tuple<String, String> authTokens = loginViaSaml("http://localhost:54321" + SP_ACS_PATH_1);
+        verifyElasticsearchAccessTokenForRoleMapping(authTokens.v1());
+        final String accessToken = verifyElasticsearchRefreshToken(authTokens.v2());
+        verifyElasticsearchAccessTokenForRoleMapping(accessToken);
+    }
+
+    public void testLoginUserWithAuthorizingRealm() throws Exception {
+        // this ACS comes from the config in build.gradle
+        final Tuple<String, String> authTokens = loginViaSaml("http://localhost:54321" + SP_ACS_PATH_2);
+        verifyElasticsearchAccessTokenForAuthorizingRealms(authTokens.v1());
+        final String accessToken = verifyElasticsearchRefreshToken(authTokens.v2());
+        verifyElasticsearchAccessTokenForAuthorizingRealms(accessToken);
+    }
+
+    private Tuple<String, String> loginViaSaml(String acs) throws Exception {
+        this.acs = new URI(acs);
         final BasicHttpContext context = new BasicHttpContext();
         try (CloseableHttpClient client = getHttpClient()) {
             final URI loginUri = goToLoginPage(client, context);
@@ -234,25 +273,21 @@ public class SamlAuthenticationIT extends ESRestTestCase {
             final Object accessToken = result.get("access_token");
             assertThat(accessToken, notNullValue());
             assertThat(accessToken, instanceOf(String.class));
-            verifyElasticsearchAccessToken((String) accessToken);
 
             final Object refreshToken = result.get("refresh_token");
             assertThat(refreshToken, notNullValue());
             assertThat(refreshToken, instanceOf(String.class));
-            verifyElasticsearchRefreshToken((String) refreshToken);
+
+            return new Tuple<>((String) accessToken, (String) refreshToken);
         }
     }
 
     /**
      * Verifies that the provided "Access Token" (see {@link org.elasticsearch.xpack.security.authc.TokenService})
-     * is for the expected user with the expected name and roles.
+     * is for the expected user with the expected name and roles if the user was created from Role-Mapping
      */
-    private void verifyElasticsearchAccessToken(String accessToken) throws IOException {
+    private void verifyElasticsearchAccessTokenForRoleMapping(String accessToken) throws IOException {
-        Request request = new Request("GET", "/_xpack/security/_authenticate");
+        final Map<String, Object> map = callAuthenticateApiUsingAccessToken(accessToken);
-        RequestOptions.Builder options = request.getOptions().toBuilder();
-        options.addHeader("Authorization", "Bearer " + accessToken);
-        request.setOptions(options);
-        final Map<String, Object> map = entityAsMap(client().performRequest(request));
         assertThat(map.get("username"), equalTo("thor"));
         assertThat(map.get("full_name"), equalTo("Thor Odinson"));
         assertSingletonList(map.get("roles"), "kibana_user");
@@ -266,15 +301,37 @@ public class SamlAuthenticationIT extends ESRestTestCase {
     }
 
     /**
-     * Verifies that the provided "Refresh Token" (see {@link org.elasticsearch.xpack.security.authc.TokenService})
+     * Verifies that the provided "Access Token" (see {@link org.elasticsearch.xpack.security.authc.TokenService})
-     * can be used to get a new valid access token and refresh token.
+     * is for the expected user with the expected name and roles if the user was retrieved from the native realm
      */
-    private void verifyElasticsearchRefreshToken(String refreshToken) throws IOException {
+    private void verifyElasticsearchAccessTokenForAuthorizingRealms(String accessToken) throws IOException {
-        Request request = new Request("POST", "/_xpack/security/oauth2/token");
+        final Map<String, Object> map = callAuthenticateApiUsingAccessToken(accessToken);
-        request.setJsonEntity("{ \"grant_type\":\"refresh_token\", \"refresh_token\":\"" + refreshToken + "\" }");
+        assertThat(map.get("username"), equalTo("thor"));
-        kibanaAuth(request);
+        assertThat(map.get("full_name"), equalTo("Thor Son of Odin"));
+        assertSingletonList(map.get("roles"), "kibana_dashboard_only_user");
 
-        final Map<String, Object> result = entityAsMap(client().performRequest(request));
+        assertThat(map.get("metadata"), instanceOf(Map.class));
+        final Map<?, ?> metadata = (Map<?, ?>) map.get("metadata");
+        assertThat(metadata.get("is_native"), equalTo(true));
+    }
+
+    private Map<String, Object> callAuthenticateApiUsingAccessToken(String accessToken) throws IOException {
+        Request request = new Request("GET", "/_xpack/security/_authenticate");
+        RequestOptions.Builder options = request.getOptions().toBuilder();
+        options.addHeader("Authorization", "Bearer " + accessToken);
+        request.setOptions(options);
+        return entityAsMap(client().performRequest(request));
+    }
+
+    private String verifyElasticsearchRefreshToken(String refreshToken) throws IOException {
+        final Map<String, ?> body = MapBuilder.<String, Object>newMapBuilder()
+            .put("grant_type", "refresh_token")
+            .put("refresh_token", refreshToken)
+            .map();
+        final Response response = client().performRequest(buildRequest("POST", "/_xpack/security/oauth2/token", body, kibanaAuth()));
+        assertOK(response);
+
+        final Map<String, Object> result = entityAsMap(response);
         final Object newRefreshToken = result.get("refresh_token");
         assertThat(newRefreshToken, notNullValue());
         assertThat(newRefreshToken, instanceOf(String.class));
@@ -282,7 +339,7 @@ public class SamlAuthenticationIT extends ESRestTestCase {
         final Object accessToken = result.get("access_token");
         assertThat(accessToken, notNullValue());
         assertThat(accessToken, instanceOf(String.class));
-        verifyElasticsearchAccessToken((String) accessToken);
+        return (String) accessToken;
     }
 
     /**
@@ -348,7 +405,7 @@ public class SamlAuthenticationIT extends ESRestTestCase {
         form.setEntity(new UrlEncodedFormEntity(params));
 
         return execute(client, form, context,
-                response -> parseSamlSubmissionForm(response.getEntity().getContent()));
+            response -> parseSamlSubmissionForm(response.getEntity().getContent()));
     }
 
     /**
@@ -358,14 +415,14 @@ public class SamlAuthenticationIT extends ESRestTestCase {
      * @param saml The (deflated + base64 encoded) {@code SAMLResponse} parameter to post the ACS
      */
     private Map<String, Object> submitSamlResponse(BasicHttpContext context, CloseableHttpClient client, URI acs, String saml)
-            throws IOException {
+        throws IOException {
         assertThat("SAML submission target", acs, notNullValue());
-        assertThat(acs.getPath(), equalTo(SP_ACS_PATH));
+        assertThat(acs, equalTo(this.acs));
         assertThat("SAML submission content", saml, notNullValue());
 
         // The ACS url provided from the SP is going to be wrong because the gradle
         // build doesn't know what the web server's port is, so it uses a fake one.
-        final HttpPost form = new HttpPost(getUrl(SP_ACS_PATH));
+        final HttpPost form = new HttpPost(getUrl(this.acs.getPath()));
         List<NameValuePair> params = new ArrayList<>();
         params.add(new BasicNameValuePair(SAML_RESPONSE_FIELD, saml));
         form.setEntity(new UrlEncodedFormEntity(params));
@@ -460,13 +517,14 @@ public class SamlAuthenticationIT extends ESRestTestCase {
      * sends a redirect to that page.
      */
     private void httpLogin(HttpExchange http) throws IOException {
-        Request request = new Request("POST", "/_xpack/security/saml/prepare");
+        final Map<String, String> body = Collections.singletonMap("acs", this.acs.toString());
-        request.setJsonEntity("{}");
+        Request request = buildRequest("POST", "/_xpack/security/saml/prepare", body, kibanaAuth());
-        kibanaAuth(request);
+        final Response prepare = client().performRequest(request);
-        final Map<String, Object> body = entityAsMap(client().performRequest(request));
+        assertOK(prepare);
-        logger.info("Created SAML authentication request {}", body);
+        final Map<String, Object> responseBody = parseResponseAsMap(prepare.getEntity());
-        http.getResponseHeaders().add("Set-Cookie", REQUEST_ID_COOKIE + "=" + body.get("id"));
+        logger.info("Created SAML authentication request {}", responseBody);
-        http.getResponseHeaders().add("Location", (String) body.get("redirect"));
+        http.getResponseHeaders().add("Set-Cookie", REQUEST_ID_COOKIE + "=" + responseBody.get("id"));
+        http.getResponseHeaders().add("Location", (String) responseBody.get("redirect"));
         http.sendResponseHeaders(302, 0);
         http.close();
     }
@@ -501,10 +559,11 @@ public class SamlAuthenticationIT extends ESRestTestCase {
         final String id = getCookie(REQUEST_ID_COOKIE, http);
         assertThat(id, notNullValue());
 
-        Request request = new Request("POST", "/_xpack/security/saml/authenticate");
+        final Map<String, ?> body = MapBuilder.<String, Object>newMapBuilder()
-        request.setJsonEntity("{ \"content\" : \"" + saml + "\", \"ids\": [\"" + id + "\"] }");
+            .put("content", saml)
-        kibanaAuth(request);
+            .put("ids", Collections.singletonList(id))
-        return client().performRequest(request);
+            .map();
+        return client().performRequest(buildRequest("POST", "/_xpack/security/saml/authenticate", body, kibanaAuth()));
     }
 
     private List<NameValuePair> parseRequestForm(HttpExchange http) throws IOException {
@@ -518,6 +577,7 @@ public class SamlAuthenticationIT extends ESRestTestCase {
         try {
             final String cookies = http.getRequestHeaders().getFirst("Cookie");
             if (cookies == null) {
+                logger.warn("No cookies in: {}", http.getResponseHeaders());
                 return null;
             }
             Header header = new BasicHeader("Cookie", cookies);
@@ -540,11 +600,23 @@ public class SamlAuthenticationIT extends ESRestTestCase {
         assertThat(((List<?>) value), contains(expectedElement));
     }
 
-    private static void kibanaAuth(Request request) {
+    private Request buildRequest(String method, String endpoint, Map<String, ?> body, Header... headers) throws IOException {
-        RequestOptions.Builder options = request.getOptions().toBuilder();
+        Request request = new Request(method, endpoint);
-        options.addHeader("Authorization",
+        XContentBuilder builder = XContentFactory.jsonBuilder().map(body);
-                UsernamePasswordToken.basicAuthHeaderValue("kibana", new SecureString(KIBANA_PASSWORD.toCharArray())));
+        if (body != null) {
+            request.setJsonEntity(BytesReference.bytes(builder).utf8ToString());
+        }
+        final RequestOptions.Builder options = request.getOptions().toBuilder();
+        for (Header header : headers) {
+            options.addHeader(header.getName(), header.getValue());
+        }
         request.setOptions(options);
+        return request;
+    }
+
+    private static BasicHeader kibanaAuth() {
+        final String auth = UsernamePasswordToken.basicAuthHeaderValue("kibana", new SecureString(KIBANA_PASSWORD.toCharArray()));
+        return new BasicHeader(UsernamePasswordToken.BASIC_AUTH_HEADER, auth);
     }
 
     private CloseableHttpClient getHttpClient() throws Exception {