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 
@@ -211,4 +216,4 @@ xpack:
           type: ldap
           metadata: cn
 --------------------------------------------------
---
+--

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}
-roles. This mapping is performed through the
-{ref}/security-api-put-role-mapping.html[add role mapping API].
+Your SAML users cannot do anything until they are assigned {security}
+roles. This is done through either the
+{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 AllowedRealmType allowedRealmType() {
      */
     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 @@
 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 String toString() {
         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.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 @@ private 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,
-                SETTING_REMOVE_REALM_NAME);
+        final Set<Setting<?>> settings = Sets.newHashSet(HTTP_SERVICE_KEYTAB_PATH, CACHE_TTL_SETTING, CACHE_MAX_USERS_SETTING,
+                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.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 static Set<Setting<?>> getSettings(String type) {
             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 @@
 
 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 static Set<Setting<?>> getSettings() {
         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 @@
 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 static Set<Setting<?>> getSettings() {
         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.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 @@ private void consumeUser(User user, Map<Realm, Tuple<String, Exception>> message
          * names of users that exist using a timing attack
          */
         private void lookupRunAsUser(final User user, String runAsUsername, Consumer<User> userConsumer) {
-            final List<Realm> realmsList = realms.asList();
-            final BiConsumer<Realm, ActionListener<User>> realmLookupConsumer = (realm, lookupUserListener) ->
-                    realm.lookupUser(runAsUsername, ActionListener.wrap((lookedupUser) -> {
-                        if (lookedupUser != null) {
-                            lookedupBy = new RealmRef(realm.name(), realm.type(), nodeName);
-                            lookupUserListener.onResponse(lookedupUser);
-                        } else {
-                            lookupUserListener.onResponse(null);
-                        }
-                    }, lookupUserListener::onFailure));
-
-            final IteratingActionListener<User, Realm> userLookupListener =
-                    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));
-            }
+            final RealmUserLookup lookup = new RealmUserLookup(realms.asList(), threadContext);
+            lookup.lookup(runAsUsername, ActionListener.wrap(tuple -> {
+                if (tuple == null) {
+                    // the user does not exist, but we still create a User object, which will later be rejected by authz
+                    userConsumer.accept(new User(runAsUsername, null, user));
+                } else {
+                    User foundUser = Objects.requireNonNull(tuple.v1());
+                    Realm realm = Objects.requireNonNull(tuple.v2());
+                    lookedupBy = new RealmRef(realm.name(), realm.type(), nodeName);
+                    userConsumer.accept(new User(foundUser, user));
+                }
+            }, exception -> listener.onFailure(request.exceptionProcessingRequest(exception, authenticationToken))));
         }
 
         /**

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

@@ -93,6 +93,7 @@ public Realms(Settings settings, Environment env, Map<String, Realm.Factory> fac
 
         this.standardRealmsOnly = Collections.unmodifiableList(standardRealms);
         this.nativeRealmsOnly = Collections.unmodifiableList(nativeRealms);
+        realms.forEach(r -> r.initialize(this, licenseState));
     }
 
     @Override