Add authorizing_realms support to PKI realm

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

@@ -8,6 +8,7 @@
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.xpack.core.security.authc.support.DelegatedAuthorizationSettings;
 import org.elasticsearch.xpack.core.security.user.User;
 
 import java.util.HashMap;
@@ -131,6 +132,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) {
+    }
+
     /**
      * A factory interface to construct a security realm.
      */

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/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("authorizing_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;
 
@@ -379,33 +381,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

@@ -92,6 +92,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));
     }
 
     @Override

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

@@ -31,12 +31,12 @@
 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 +75,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 +92,15 @@ public PkiRealm(RealmConfig config, ResourceWatcherService watcherService, Nativ
                 .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) {
+        if (delegatedRealms != null) {
+            throw new IllegalStateException("Realm has already been initialized");
+        }
+        delegatedRealms = new DelegatedAuthorizationSupport(realms, config);
     }
 
     @Override
@@ -105,32 +115,50 @@ public X509AuthenticationToken token(ThreadContext context) {
 
     @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 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);
-                    try (ReleasableLock ignored = readLock.acquire()) {
-                        cache.put(fingerprint, computedUser);
+                final ActionListener<AuthenticationResult> cachingListener = ActionListener.wrap(result -> {
+                    if (result.isAuthenticated()) {
+                        try (ReleasableLock ignored = readLock.acquire()) {
+                            cache.put(fingerprint, result.getUser());
+                        }
                     }
-                    listener.onResponse(AuthenticationResult.success(computedUser));
-                }, listener::onFailure));
+                    listener.onResponse(result);
+                }, 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/support/DelegatedAuthorizationSupport.java

@@ -0,0 +1,108 @@
+/*
+ * 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.util.concurrent.ThreadContext;
+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.support.DelegatedAuthorizationSettings;
+import org.elasticsearch.xpack.core.security.user.User;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.elasticsearch.common.Strings.collectionToDelimitedString;
+
+/**
+ * Utility class for supporting "delegated authorization" (aka "authorizing_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;
+
+    /**
+     * Resolves the {@link DelegatedAuthorizationSettings#AUTHZ_REALMS} setting from {@code config} and calls
+     * {@link #DelegatedAuthorizationSupport(Iterable, List, ThreadContext)}
+     */
+    public DelegatedAuthorizationSupport(Iterable<? extends Realm> allRealms, RealmConfig config) {
+        this(allRealms, DelegatedAuthorizationSettings.AUTHZ_REALMS.get(config.settings()), config.threadContext());
+    }
+
+    /**
+     * 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, ThreadContext threadContext) {
+       this.lookup = new RealmUserLookup(resolveRealms(allRealms, lookupRealms), threadContext);
+       this.logger = Loggers.getLogger(getClass());
+    }
+
+    /**
+     * 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 (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;
+    }
+
+    private Realm findRealm(String name, Iterable<? extends Realm> allRealms) {
+        for (Realm realm : allRealms) {
+            if (name.equals(realm.name())) {
+                return realm;
+            }
+        }
+        throw new IllegalArgumentException("configured authorizing realm [" + name + "] does not exist (or is not enabled)");
+    }
+
+}