Allow User/Password realms to disable authc

The "lookupUser" method on a realm facilitates the "run-as" and
"authorization_realms" features.
This commit allows a realm to be used for "lookup only", in which
case the "authenticate" method (and associated token methods) are
disabled.
It does this through the introduction of a new
"authentication.enabled" setting, which defaults to true.

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

@@ -19,6 +19,6 @@ private NativeRealmSettings() {}
      * @return The {@link Setting setting configuration} for this realm type
      */
     public static Set<Setting<?>> getSettings() {
-        return CachingUsernamePasswordRealmSettings.getCachingSettings();
+        return CachingUsernamePasswordRealmSettings.getSettings();
     }
 }

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

@@ -19,6 +19,6 @@ private FileRealmSettings() {}
      * @return The {@link Setting setting configuration} for this realm type
      */
     public static Set<Setting<?>> getSettings() {
-        return CachingUsernamePasswordRealmSettings.getCachingSettings();
+        return CachingUsernamePasswordRealmSettings.getSettings();
     }
 }

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

@@ -29,7 +29,7 @@ private LdapRealmSettings() {}
      */
     public static Set<Setting<?>> getSettings(String type) {
         Set<Setting<?>> settings = new HashSet<>();
-        settings.addAll(CachingUsernamePasswordRealmSettings.getCachingSettings());
+        settings.addAll(CachingUsernamePasswordRealmSettings.getSettings());
         settings.addAll(CompositeRoleMapperSettings.getSettings());
         settings.add(LdapRealmSettings.EXECUTION_TIMEOUT);
         if (AD_TYPE.equals(type)) {

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

@@ -21,12 +21,15 @@ public final class CachingUsernamePasswordRealmSettings {
     public static final Setting<Integer> CACHE_MAX_USERS_SETTING = Setting.intSetting("cache.max_users", DEFAULT_MAX_USERS,
             Setting.Property.NodeScope);
 
+    public static final Setting<Boolean> AUTHC_ENABLED_SETTING = Setting.boolSetting("authentication.enabled", true,
+        Setting.Property.NodeScope);
+
     private CachingUsernamePasswordRealmSettings() {}
 
     /**
      * Returns the {@link Setting setting configuration} that is common for all caching realms
      */
-    public static Set<Setting<?>> getCachingSettings() {
-        return new HashSet<>(Arrays.asList(CACHE_HASH_ALGO_SETTING, CACHE_TTL_SETTING, CACHE_MAX_USERS_SETTING));
+    public static Set<Setting<?>> getSettings() {
+        return new HashSet<>(Arrays.asList(CACHE_HASH_ALGO_SETTING, CACHE_TTL_SETTING, CACHE_MAX_USERS_SETTING, AUTHC_ENABLED_SETTING));
     }
 }

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

@@ -11,6 +11,7 @@
 import org.elasticsearch.common.settings.SecureString;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.ListenableFuture;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationResult;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationToken;
@@ -30,6 +31,7 @@ public abstract class CachingUsernamePasswordRealm extends UsernamePasswordRealm
 
     private final Cache<String, ListenableFuture<UserWithHash>> cache;
     private final ThreadPool threadPool;
+    private final boolean authenticationEnabled;
     final Hasher cacheHasher;
 
     protected CachingUsernamePasswordRealm(String type, RealmConfig config, ThreadPool threadPool) {
@@ -45,6 +47,7 @@ protected CachingUsernamePasswordRealm(String type, RealmConfig config, ThreadPo
         } else {
             cache = null;
         }
+        this.authenticationEnabled = CachingUsernamePasswordRealmSettings.AUTHC_ENABLED_SETTING.get(config.settings());
     }
 
     @Override
@@ -63,15 +66,34 @@ public final void expireAll() {
         }
     }
 
+    @Override
+    public UsernamePasswordToken token(ThreadContext threadContext) {
+        if (authenticationEnabled == false) {
+            return null;
+        }
+        return super.token(threadContext);
+    }
+
+    @Override
+    public boolean supports(AuthenticationToken token) {
+        return authenticationEnabled && super.supports(token);
+    }
+
     /**
      * If the user exists in the cache (keyed by the principle name), then the password is validated
      * against a hash also stored in the cache.  Otherwise the subclass authenticates the user via
-     * doAuthenticate
+     * doAuthenticate.
+     * This method will respond with {@link AuthenticationResult#notHandled()} if
+     * {@link CachingUsernamePasswordRealmSettings#AUTHC_ENABLED_SETTING authentication is not enabled}.
      * @param authToken The authentication token
      * @param listener to be called at completion
      */
     @Override
     public final void authenticate(AuthenticationToken authToken, ActionListener<AuthenticationResult> listener) {
+        if (authenticationEnabled == false) {
+            listener.onResponse(AuthenticationResult.notHandled());
+            return;
+        }
         final UsernamePasswordToken token = (UsernamePasswordToken) authToken;
         try {
             if (cache == null) {

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

@@ -11,6 +11,7 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 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.test.SecuritySettingsSourceField;
@@ -62,7 +63,7 @@ public void stop() throws InterruptedException {
         }
     }
 
-    public void testSettings() throws Exception {
+    public void testCacheSettings() throws Exception {
         String cachingHashAlgo = Hasher.values()[randomIntBetween(0, Hasher.values().length - 1)].name().toLowerCase(Locale.ROOT);
         int maxUsers = randomIntBetween(10, 100);
         TimeValue ttl = TimeValue.timeValueMinutes(randomIntBetween(10, 20));
@@ -560,6 +561,33 @@ protected void doLookupUser(String username, ActionListener<User> listener) {
         assertEquals(1, lookupCounter.get());
     }
 
+    public void testAuthenticateDisabled() throws Exception {
+        final Settings settings = Settings.builder()
+            .put(CachingUsernamePasswordRealmSettings.AUTHC_ENABLED_SETTING.getKey(), false)
+            .build();
+        final Environment env = TestEnvironment.newEnvironment(globalSettings);
+        final ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
+        final RealmConfig config = new RealmConfig("test_authentication_disabled", settings, globalSettings, env, threadContext);
+        final AlwaysAuthenticateCachingRealm realm = new AlwaysAuthenticateCachingRealm(config, threadPool);
+
+        final UsernamePasswordToken token = new UsernamePasswordToken("phil", new SecureString("tahiti"));
+        UsernamePasswordToken.putTokenHeader(threadContext, token);
+        assertThat(realm.token(threadContext), nullValue());
+        assertThat(realm.supports(token), equalTo(false));
+
+        PlainActionFuture<AuthenticationResult> authFuture = new PlainActionFuture<>();
+        realm.authenticate(token, authFuture);
+        final AuthenticationResult authResult = authFuture.get();
+        assertThat(authResult.isAuthenticated(), equalTo(false));
+        assertThat(authResult.getStatus(), equalTo(AuthenticationResult.Status.CONTINUE));
+
+        PlainActionFuture<User> lookupFuture = new PlainActionFuture<>();
+        realm.lookupUser(token.principal(), lookupFuture);
+        final User user = lookupFuture.get();
+        assertThat(user, notNullValue());
+        assertThat(user.principal(), equalTo(token.principal()));
+    }
+
     static class FailingAuthenticationRealm extends CachingUsernamePasswordRealm {
 
         FailingAuthenticationRealm(Settings settings, Settings global, ThreadPool threadPool) {