Allow AD realm to perform 'run-as' lookups (elastic/x-pack-elasticsearch#2531)

- Marks the AD Session factory as supporting "lookup" (Refer: elastic/x-pack-elasticsearch@40b07b3)
- Adds "pool.enabled" as a registered setting on AD realm (Refer: elastic/x-pack-elasticsearch@40b07b3)
- Fixes LDAP user lookup that has been broken since 6.x (Refer: elastic/x-pack-elasticsearch@f796949)

Original commit: elastic/x-pack-elasticsearch@62ff6129a1

plugin/src/main/java/org/elasticsearch/xpack/security/authc/ldap/ActiveDirectorySessionFactory.java

@@ -65,7 +65,7 @@ class ActiveDirectorySessionFactory extends PoolingSessionFactory {
     static final String AD_DOWN_LEVEL_USER_SEARCH_FILTER_SETTING = "user_search.down_level_filter";
     static final String AD_USER_SEARCH_SCOPE_SETTING = "user_search.scope";
     private static final String NETBIOS_NAME_FILTER_TEMPLATE = "(netbiosname={0})";
-    private static final Setting<Boolean> POOL_ENABLED = Setting.boolSetting("user_search.pool.enabled",
+    static final Setting<Boolean> POOL_ENABLED = Setting.boolSetting("user_search.pool.enabled",
             settings -> Boolean.toString(PoolingSessionFactory.BIND_DN.exists(settings)), Setting.Property.NodeScope);
 
     final DefaultADAuthenticator defaultADAuthenticator;
@@ -108,6 +108,13 @@ class ActiveDirectorySessionFactory extends PoolingSessionFactory {
         return new String[] {"ldap://" + settings.get(AD_DOMAIN_NAME_SETTING) + ":389"};
     }
 
+    @Override
+    public boolean supportsUnauthenticatedSession() {
+        // Strictly, we only support unauthenticated sessions if there is a bind_dn or a connection pool, but the
+        // getUnauthenticatedSession... methods handle the situations correctly, so it's OK to always return true here.
+        return true;
+    }
+
     @Override
     void getSessionWithPool(LDAPConnectionPool connectionPool, String user, SecureString password, ActionListener<LdapSession> listener) {
         getADAuthenticator(user).authenticate(connectionPool, user, password, listener);
@@ -207,6 +214,7 @@ class ActiveDirectorySessionFactory extends PoolingSessionFactory {
         settings.add(Setting.simpleString(AD_UPN_USER_SEARCH_FILTER_SETTING, Setting.Property.NodeScope));
         settings.add(Setting.simpleString(AD_DOWN_LEVEL_USER_SEARCH_FILTER_SETTING, Setting.Property.NodeScope));
         settings.add(Setting.simpleString(AD_USER_SEARCH_SCOPE_SETTING, Setting.Property.NodeScope));
+        settings.add(POOL_ENABLED);
         settings.addAll(PoolingSessionFactory.getSettings());
         return settings;
     }

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

@@ -161,7 +161,8 @@ public final class LdapRealm extends CachingUsernamePasswordRealm {
         if (sessionFactory.supportsUnauthenticatedSession()) {
             // 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 ActionListener<AuthenticationResult> sessionListener = ActionListener.wrap(AuthenticationResult::getUser,
+            final ActionListener<AuthenticationResult> sessionListener = ActionListener.wrap(
+                    result -> userActionListener.onResponse(result.getUser()),
                     userActionListener::onFailure);
             final CancellableLdapRunnable<User> cancellableLdapRunnable = new CancellableLdapRunnable<>(userActionListener, e -> null,
                     () -> sessionFactory.unauthenticatedSession(username,

plugin/src/test/java/org/elasticsearch/integration/ldap/ActiveDirectoryRunAsTests.java

@@ -0,0 +1,75 @@
+/*
+ * 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.integration.ldap;
+
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import org.elasticsearch.action.ActionFuture;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.common.collect.MapBuilder;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.test.junit.annotations.Network;
+import org.elasticsearch.xpack.security.action.user.AuthenticateAction;
+import org.elasticsearch.xpack.security.action.user.AuthenticateRequest;
+import org.elasticsearch.xpack.security.action.user.AuthenticateResponse;
+import org.elasticsearch.xpack.security.authc.AuthenticationService;
+import org.elasticsearch.xpack.security.authc.Realm;
+import org.elasticsearch.xpack.security.authc.Realms;
+import org.elasticsearch.xpack.security.authc.ldap.ActiveDirectorySessionFactoryTests;
+import org.elasticsearch.xpack.security.authc.ldap.LdapRealm;
+import org.elasticsearch.xpack.security.authc.support.UsernamePasswordToken;
+import org.elasticsearch.xpack.security.user.ElasticUser;
+import org.hamcrest.Matchers;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import static org.elasticsearch.xpack.security.authc.support.UsernamePasswordToken.BASIC_AUTH_HEADER;
+
+/**
+ * This tests that "run-as" works on LDAP/AD realms
+ */
+@Network
+public class ActiveDirectoryRunAsTests extends AbstractAdLdapRealmTestCase {
+
+    @BeforeClass
+    public static void selectRealmConfig() {
+        realmConfig = randomFrom(RealmConfig.AD, RealmConfig.AD_SSL);
+    }
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        final Settings.Builder builder = Settings.builder().put(super.nodeSettings(nodeOrdinal));
+        switch (realmConfig) {
+            case AD:
+            case AD_SSL:
+                builder.put(XPACK_SECURITY_AUTHC_REALMS_EXTERNAL + ".bind_dn", "ironman@ad.test.elasticsearch.com")
+                        .put(XPACK_SECURITY_AUTHC_REALMS_EXTERNAL + ".bind_password", ActiveDirectorySessionFactoryTests.PASSWORD)
+                        .put(XPACK_SECURITY_AUTHC_REALMS_EXTERNAL + ".user_search.pool.enabled", false);
+                break;
+            default:
+                throw new IllegalStateException("Unknown realm config " + realmConfig);
+        }
+        return builder.build();
+    }
+
+    public void testRunAs() throws Exception {
+        String avenger = realmConfig.loginWithCommonName ? "Natasha Romanoff" : "blackwidow";
+        final AuthenticateRequest request = new AuthenticateRequest(avenger);
+        final ActionFuture<AuthenticateResponse> future = runAsClient(avenger).execute(AuthenticateAction.INSTANCE, request);
+        final AuthenticateResponse response = future.get(30, TimeUnit.SECONDS);
+        assertThat(response.user().principal(), Matchers.equalTo(avenger));
+    }
+
+    protected Client runAsClient(String user) {
+        final Map<String, String> headers = MapBuilder.<String, String>newMapBuilder()
+                .put(BASIC_AUTH_HEADER, UsernamePasswordToken.basicAuthHeaderValue(ElasticUser.NAME, BOOTSTRAP_PASSWORD))
+                .put(AuthenticationService.RUN_AS_USER_HEADER, user)
+                .map();
+        return client().filterWithHeader(headers);
+    }
+
+}

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

@@ -54,7 +54,6 @@ import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
 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;
@@ -237,6 +236,34 @@ public class ActiveDirectoryRealmTests extends ESTestCase {
         verify(sessionFactory, times(2)).session(eq("CN=ironman"), any(SecureString.class), any(ActionListener.class));
     }
 
+    public void testUnauthenticatedLookupWithConnectionPool() throws Exception {
+        doUnauthenticatedLookup(true);
+    }
+
+    public void testUnauthenticatedLookupWithoutConnectionPool() throws Exception {
+        doUnauthenticatedLookup(false);
+    }
+
+    private void doUnauthenticatedLookup(boolean pooled) throws Exception {
+        Settings settings = settings(Settings.builder()
+                .put(ActiveDirectorySessionFactory.POOL_ENABLED.getKey(), pooled)
+                .put(ActiveDirectorySessionFactory.BIND_DN.getKey(), "CN=ironman@ad.test.elasticsearch.com")
+                .put(ActiveDirectorySessionFactory.BIND_PASSWORD.getKey(), PASSWORD)
+                .build());
+        RealmConfig config = new RealmConfig("testUnauthenticatedLookupWithConnectionPool", settings, globalSettings,
+                new Environment(globalSettings), new ThreadContext(globalSettings));
+        try(ActiveDirectorySessionFactory sessionFactory = new ActiveDirectorySessionFactory(config, sslService)) {
+            DnRoleMapper roleMapper = new DnRoleMapper(config, resourceWatcherService);
+            LdapRealm realm = new LdapRealm(LdapRealm.AD_TYPE, config, sessionFactory, roleMapper, threadPool);
+
+            PlainActionFuture<User> future = new PlainActionFuture<>();
+            realm.lookupUser("CN=Thor", future);
+            final User user = future.actionGet();
+            assertThat(user, notNullValue());
+            assertThat(user.principal(), equalTo("CN=Thor"));
+        }
+    }
+
     public void testRealmMapsGroupsToRoles() throws Exception {
         Settings settings = settings(Settings.builder()
                 .put(ROLE_MAPPING_FILE_SETTING, getDataPath("role_mapping.yml"))