Add support for secondary authentication (#52093)

This change makes it possible to send secondary authentication
credentials to select endpoints that need to perform a single action
in the context of two users.

Typically this need arises when a server process needs to call an
endpoint that users should not (or might not) have direct access to,
but some part of that action must be performed using the logged-in
user's identity.

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

@@ -15,13 +15,15 @@
 import org.elasticsearch.xpack.core.security.authc.Authentication;
 import org.elasticsearch.xpack.core.security.authc.Authentication.AuthenticationType;
 import org.elasticsearch.xpack.core.security.authc.support.AuthenticationContextSerializer;
+import org.elasticsearch.xpack.core.security.authc.support.SecondaryAuthentication;
 import org.elasticsearch.xpack.core.security.user.User;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.util.Collections;
 import java.util.Objects;
 import java.util.function.Consumer;
+import java.util.function.Function;
 
 /**
  * A lightweight utility that can find the current user and authentication information for the local thread.
@@ -55,6 +57,19 @@ public Authentication getAuthentication() {
         }
     }
 
+    /**
+     * Returns the "secondary authentication" (see {@link SecondaryAuthentication}) information,
+     * or {@code null} if the current request does not have a secondary authentication context
+     */
+    public SecondaryAuthentication getSecondaryAuthentication() {
+        try {
+            return SecondaryAuthentication.readFromContext(this);
+        } catch (IOException e) {
+            logger.error("failed to read secondary authentication", e);
+            throw new UncheckedIOException(e);
+        }
+    }
+
     public ThreadContext getThreadContext() {
         return threadContext;
     }
@@ -97,6 +112,18 @@ public void executeAsUser(User user, Consumer<StoredContext> consumer, Version v
         }
     }
 
+    /**
+     * Runs the consumer in a new context as the provided user. The original context is provided to the consumer. When this method
+     * returns, the original context is restored.
+     */
+    public <T> T executeWithAuthentication(Authentication authentication, Function<StoredContext, T> consumer) {
+        final StoredContext original = threadContext.newStoredContext(true);
+        try (ThreadContext.StoredContext ignore = threadContext.stashContext()) {
+            setAuthentication(authentication);
+            return consumer.apply(original);
+        }
+    }
+
     /**
      * Runs the consumer in a new context after setting a new version of the authentication that is compatible with the version provided.
      * The original context is provided to the consumer. When this method returns, the original context is restored.

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

@@ -143,6 +143,14 @@ public int hashCode() {
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
+        toXContentFragment(builder);
+        return builder.endObject();
+    }
+
+    /**
+     * Generates XContent without the start/end object.
+     */
+    public void toXContentFragment(XContentBuilder builder) throws IOException {
         builder.field(User.Fields.USERNAME.getPreferredName(), user.principal());
         builder.array(User.Fields.ROLES.getPreferredName(), user.roles());
         builder.field(User.Fields.FULL_NAME.getPreferredName(), user.fullName());
@@ -162,7 +170,6 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             builder.field(User.Fields.REALM_TYPE.getPreferredName(), getAuthenticatedBy().getType());
         }
         builder.endObject();
-        return builder.endObject();
     }
 
     @Override

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

@@ -75,6 +75,7 @@ public Authentication getAuthentication(ThreadContext context) {
     public void writeToContext(Authentication authentication, ThreadContext ctx) throws IOException {
         ensureContextDoesNotContainAuthentication(ctx);
         String header = authentication.encode();
+        assert header != null : "Authentication object encoded to null"; // this usually happens with mock objects in tests
         ctx.putTransient(contextKey, authentication);
         ctx.putHeader(contextKey, header);
     }

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

@@ -0,0 +1,86 @@
+/*
+ * 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.Nullable;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.xpack.core.security.SecurityContext;
+import org.elasticsearch.xpack.core.security.authc.Authentication;
+
+import java.io.IOException;
+import java.util.Objects;
+import java.util.function.Function;
+
+/**
+ * Some Elasticsearch APIs need to be provided with 2 sets of credentials.
+ * Typically this happens when a system user needs to perform an action while accessing data on behalf of, or user information regarding
+ * a logged in user.
+ * This class is a representation of that secondary user that can be activated in the security context while processing specific blocks
+ * of code or within a listener.
+ */
+public class SecondaryAuthentication {
+
+    private static final String THREAD_CTX_KEY = "_xpack_security_secondary_authc";
+
+    private final SecurityContext securityContext;
+    private final Authentication authentication;
+
+    public SecondaryAuthentication(SecurityContext securityContext, Authentication authentication) {
+        this.securityContext = Objects.requireNonNull(securityContext);
+        this.authentication = Objects.requireNonNull(authentication);
+    }
+
+    @Nullable
+    public static SecondaryAuthentication readFromContext(SecurityContext securityContext) throws IOException {
+        final Authentication authentication = serializer().readFromContext(securityContext.getThreadContext());
+        if (authentication == null) {
+            return null;
+        }
+        return new SecondaryAuthentication(securityContext, authentication);
+    }
+
+    public void writeToContext(ThreadContext threadContext) throws IOException {
+        serializer().writeToContext(this.authentication, threadContext);
+    }
+
+    private static AuthenticationContextSerializer serializer() {
+        return new AuthenticationContextSerializer(THREAD_CTX_KEY);
+    }
+
+    public Authentication getAuthentication() {
+        return authentication;
+    }
+
+    public <T> T execute(Function<ThreadContext.StoredContext, T> body) {
+        return this.securityContext.executeWithAuthentication(this.authentication, body);
+    }
+
+    public Runnable wrap(Runnable runnable) {
+        return () -> execute(ignore -> {
+            runnable.run();
+            return null;
+        });
+    }
+
+    @Override
+    public String toString() {
+        return getClass().getSimpleName() + "{" + authentication + "}";
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        final SecondaryAuthentication that = (SecondaryAuthentication) o;
+        return authentication.equals(that.authentication);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(authentication);
+    }
+}

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

@@ -0,0 +1,162 @@
+/*
+ * 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.Version;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.ContextPreservingActionListener;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.threadpool.TestThreadPool;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.xpack.core.security.SecurityContext;
+import org.elasticsearch.xpack.core.security.authc.Authentication;
+import org.elasticsearch.xpack.core.security.user.User;
+import org.elasticsearch.xpack.core.security.authc.support.SecondaryAuthentication;
+import org.junit.After;
+import org.junit.Before;
+
+import java.util.concurrent.Future;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.sameInstance;
+
+public class SecondaryAuthenticationTests extends ESTestCase {
+
+    private SecurityContext securityContext;
+    private ThreadPool threadPool;
+
+    @Before
+    public void setupObjects() {
+        this.threadPool = new TestThreadPool(getTestName());
+        this.securityContext = new SecurityContext(Settings.EMPTY, threadPool.getThreadContext());
+    }
+
+    @After
+    public void cleanup() {
+        this.threadPool.shutdownNow();
+    }
+
+    public void testCannotCreateObjectWithNullAuthentication() {
+        expectThrows(NullPointerException.class, () -> new SecondaryAuthentication(securityContext, null));
+    }
+
+    public void testSynchronousExecuteInSecondaryContext() {
+        final User user1 = new User("u1", "role1");
+        securityContext.setUser(user1, Version.CURRENT);
+        assertThat(securityContext.getUser().principal(), equalTo("u1"));
+
+        final Authentication authentication2 = new Authentication(new User("u2", "role2"), realm(), realm());
+        final SecondaryAuthentication secondaryAuth = new SecondaryAuthentication(securityContext, authentication2);
+
+        assertThat(securityContext.getUser().principal(), equalTo("u1"));
+        var result = secondaryAuth.execute(original -> {
+            assertThat(securityContext.getUser().principal(), equalTo("u2"));
+            assertThat(securityContext.getAuthentication(), sameInstance(authentication2));
+            return "xyzzy";
+        });
+        assertThat(securityContext.getUser().principal(), equalTo("u1"));
+        assertThat(result, equalTo("xyzzy"));
+    }
+
+    public void testSecondaryContextCanBeRestored() {
+        final User user1 = new User("u1", "role1");
+        securityContext.setUser(user1, Version.CURRENT);
+        assertThat(securityContext.getUser().principal(), equalTo("u1"));
+
+        final Authentication authentication2 = new Authentication(new User("u2", "role2"), realm(), realm());
+        final SecondaryAuthentication secondaryAuth = new SecondaryAuthentication(securityContext, authentication2);
+
+        assertThat(securityContext.getUser().principal(), equalTo("u1"));
+        final AtomicReference<ThreadContext.StoredContext> secondaryContext = new AtomicReference<>();
+        secondaryAuth.execute(storedContext -> {
+            assertThat(securityContext.getUser().principal(), equalTo("u2"));
+            assertThat(securityContext.getAuthentication(), sameInstance(authentication2));
+            secondaryContext.set(threadPool.getThreadContext().newStoredContext(false));
+            storedContext.restore();
+            assertThat(securityContext.getUser().principal(), equalTo("u1"));
+            return null;
+        });
+        assertThat(securityContext.getUser().principal(), equalTo("u1"));
+        secondaryContext.get().restore();
+        assertThat(securityContext.getUser().principal(), equalTo("u2"));
+    }
+
+    public void testWrapRunnable() throws Exception {
+        final User user1 = new User("u1", "role1");
+        securityContext.setUser(user1, Version.CURRENT);
+        assertThat(securityContext.getUser().principal(), equalTo("u1"));
+
+        final Authentication authentication2 = new Authentication(new User("u2", "role2"), realm(), realm());
+        final SecondaryAuthentication secondaryAuth = new SecondaryAuthentication(securityContext, authentication2);
+
+        assertThat(securityContext.getUser().principal(), equalTo("u1"));
+        final Semaphore semaphore = new Semaphore(0);
+        final Future<?> future = threadPool.generic().submit(secondaryAuth.wrap(() -> {
+            try {
+                assertThat(securityContext.getUser().principal(), equalTo("u2"));
+                semaphore.acquire();
+                assertThat(securityContext.getUser().principal(), equalTo("u2"));
+                semaphore.acquire();
+                assertThat(securityContext.getUser().principal(), equalTo("u2"));
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new RuntimeException(e);
+            }
+        }));
+        assertThat(securityContext.getUser().principal(), equalTo("u1"));
+        semaphore.release();
+        assertThat(securityContext.getUser().principal(), equalTo("u1"));
+        semaphore.release();
+        assertThat(securityContext.getUser().principal(), equalTo("u1"));
+
+        // ensure that the runnable didn't throw any exceptions / assertions
+        future.get(1, TimeUnit.SECONDS);
+    }
+
+    public void testPreserveSecondaryContextAcrossThreads() throws Exception {
+        final User user1 = new User("u1", "role1");
+        securityContext.setUser(user1, Version.CURRENT);
+        assertThat(securityContext.getUser().principal(), equalTo("u1"));
+
+        final Authentication authentication2 = new Authentication(new User("u2", "role2"), realm(), realm());
+        final SecondaryAuthentication secondaryAuth = new SecondaryAuthentication(securityContext, authentication2);
+
+        assertThat(securityContext.getUser().principal(), equalTo("u1"));
+
+        final AtomicReference<User> threadUser = new AtomicReference<>();
+        final AtomicReference<User> listenerUser = new AtomicReference<>();
+
+        final ThreadContext threadContext = threadPool.getThreadContext();
+        secondaryAuth.execute(originalContext -> {
+            assertThat(securityContext.getUser().principal(), equalTo("u2"));
+            ActionListener<Void> listener = new ContextPreservingActionListener<>(threadContext.newRestorableContext(false),
+                ActionListener.wrap(() -> listenerUser.set(securityContext.getUser())));
+            originalContext.restore();
+            threadPool.generic().execute(() -> {
+                threadUser.set(securityContext.getUser());
+                listener.onResponse(null);
+            });
+            return null;
+        });
+        assertThat(securityContext.getUser().principal(), equalTo("u1"));
+        assertBusy(() -> assertThat(listenerUser.get(), notNullValue()), 1, TimeUnit.SECONDS);
+        assertThat(threadUser.get(), notNullValue());
+        assertThat(threadUser.get().principal(), equalTo("u1"));
+        assertThat(listenerUser.get().principal(), equalTo("u2"));
+    }
+
+    private Authentication.RealmRef realm() {
+        return new Authentication.RealmRef(randomAlphaOfLengthBetween(4, 8), randomAlphaOfLengthBetween(2, 4), randomAlphaOfLength(12));
+    }
+
+}

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

@@ -184,6 +184,7 @@
 import org.elasticsearch.xpack.security.authc.TokenService;
 import org.elasticsearch.xpack.security.authc.esnative.NativeUsersStore;
 import org.elasticsearch.xpack.security.authc.esnative.ReservedRealm;
+import org.elasticsearch.xpack.security.authc.support.SecondaryAuthenticator;
 import org.elasticsearch.xpack.security.authc.support.mapper.NativeRoleMappingStore;
 import org.elasticsearch.xpack.security.authz.AuthorizationService;
 import org.elasticsearch.xpack.security.authz.SecuritySearchOperationListener;
@@ -287,6 +288,7 @@ public class Security extends Plugin implements SystemIndexPlugin, IngestPlugin,
     private final SetOnce<TransportInterceptor> securityInterceptor = new SetOnce<>();
     private final SetOnce<IPFilter> ipFilter = new SetOnce<>();
     private final SetOnce<AuthenticationService> authcService = new SetOnce<>();
+    private final SetOnce<SecondaryAuthenticator> secondayAuthc = new SetOnce<>();
     private final SetOnce<AuditTrailService> auditTrailService = new SetOnce<>();
     private final SetOnce<SecurityContext> securityContext = new SetOnce<>();
     private final SetOnce<ThreadContext> threadContext = new SetOnce<>();
@@ -470,6 +472,10 @@ auditTrailService, failureHandler, threadPool, anonymousUser, getAuthorizationEn
         components.add(allRolesStore); // for SecurityInfoTransportAction and clear roles cache
         components.add(authzService);
 
+        final SecondaryAuthenticator secondaryAuthenticator = new SecondaryAuthenticator(securityContext.get(), authcService.get());
+        this.secondayAuthc.set(secondaryAuthenticator);
+        components.add(secondaryAuthenticator);
+
         ipFilter.set(new IPFilter(settings, auditTrailService, clusterService.getClusterSettings(), getLicenseState()));
         components.add(ipFilter.get());
         DestructiveOperations destructiveOperations = new DestructiveOperations(settings, clusterService.getClusterSettings());
@@ -640,6 +646,7 @@ public static List<Setting<?>> getSettings(List<SecurityExtension> securityExten
     public Collection<RestHeaderDefinition> getRestHeaders() {
         Set<RestHeaderDefinition> headers = new HashSet<>();
         headers.add(new RestHeaderDefinition(UsernamePasswordToken.BASIC_AUTH_HEADER, false));
+        headers.add(new RestHeaderDefinition(SecondaryAuthenticator.SECONDARY_AUTH_HEADER_NAME, false));
         if (XPackSettings.AUDIT_ENABLED.get(settings)) {
             headers.add(new RestHeaderDefinition(AuditTrail.X_FORWARDED_FOR_HEADER, true));
         }
@@ -966,7 +973,8 @@ public UnaryOperator<RestHandler> getRestHandlerWrapper(ThreadContext threadCont
         final boolean ssl = HTTP_SSL_ENABLED.get(settings);
         final SSLConfiguration httpSSLConfig = getSslService().getHttpTransportSSLConfiguration();
         boolean extractClientCertificate = ssl && getSslService().isSSLClientAuthEnabled(httpSSLConfig);
-        return handler -> new SecurityRestFilter(getLicenseState(), threadContext, authcService.get(), handler, extractClientCertificate);
+        return handler -> new SecurityRestFilter(getLicenseState(), threadContext, authcService.get(), secondayAuthc.get(),
+            handler, extractClientCertificate);
     }
 
     @Override