Kerberos Spnego Authentication Router Issue (#5706)

* Adding decoration method to proxy servlet

Change-Id: I872f9282fb60bfa20524271535980a36a87b9621

* moving the proxy request decoration to authenticators

Change-Id: I7f94b9ff5ecf08e8abf7169b58bc410f33148448

* added docs

Change-Id: I901543e52f0faf4666bfea6256a7c05593b1ae70

* use the authentication result to decorate request

Change-Id: I052650de9cd02b4faefdbcdaf2332dd3b2966af5

* adding authenticated by name

Change-Id: I074d2933460165feeddb19352eac9bd0f96f42ca

* ensure that authenticator is not null

Change-Id: Idb58e308f90db88224a06f3759114872165b24f5

* fix types and minor bug

Change-Id: I6801d49a05d5d8324406fc0280286954eb66db10

* fix typo

Change-Id: I390b12af74f44d760d0812a519125fbf0df4e97b

* use actual type names

Change-Id: I62c3ee763363781e52809ec912aafd50b8486b8e

* set authenitcatedBy to null for AutheticationResults created by
Escalator.

Change-Id: I4a675c372f59ebd8a8d19c61b85a1e4bf227a8ba

extensions-core/druid-basic-security/src/main/java/io/druid/security/basic/authentication/BasicHTTPAuthenticator.java

@@ -105,7 +105,7 @@ public class BasicHTTPAuthenticator implements Authenticator
     }
 
     if (checkCredentials(user, password.toCharArray())) {
-      return new AuthenticationResult(user, name, null);
+      return new AuthenticationResult(user, authorizerName, name, null);
     } else {
       return null;
     }
@@ -173,7 +173,7 @@ public class BasicHTTPAuthenticator implements Authenticator
       char[] password = splits[1].toCharArray();
 
       if (checkCredentials(user, password)) {
-        AuthenticationResult authenticationResult = new AuthenticationResult(user, authorizerName, null);
+        AuthenticationResult authenticationResult = new AuthenticationResult(user, authorizerName, name, null);
         servletRequest.setAttribute(AuthConfig.DRUID_AUTHENTICATION_RESULT, authenticationResult);
       }
 

extensions-core/druid-basic-security/src/main/java/io/druid/security/basic/authentication/BasicHTTPEscalator.java

@@ -59,6 +59,8 @@ public class BasicHTTPEscalator implements Escalator
   @Override
   public AuthenticationResult createEscalatedAuthenticationResult()
   {
-    return new AuthenticationResult(internalClientUsername, authorizerName, null);
+    // if you found your self asking why the authenticatedBy field is set to null please read this:
+    // https://github.com/druid-io/druid/pull/5706#discussion_r185940889
+    return new AuthenticationResult(internalClientUsername, authorizerName, null, null);
   }
 }

extensions-core/druid-basic-security/src/test/java/io/druid/security/authorization/BasicRoleBasedAuthorizerTest.java

@@ -116,7 +116,7 @@ public class BasicRoleBasedAuthorizerTest
 
     updater.setPermissions(AUTHORIZER_NAME, "druidRole", permissions);
 
-    AuthenticationResult authenticationResult = new AuthenticationResult("druid", "druid", null);
+    AuthenticationResult authenticationResult = new AuthenticationResult("druid", "druid", null, null);
 
     Access access = authorizer.authorize(
         authenticationResult,

extensions-core/druid-kerberos/src/main/java/io/druid/security/kerberos/KerberosAuthenticator.java

@@ -23,6 +23,8 @@ import com.fasterxml.jackson.annotation.JacksonInject;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonTypeName;
+import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
 import com.google.common.base.Throwables;
 import io.druid.guice.annotations.Self;
 import io.druid.java.util.common.StringUtils;
@@ -41,6 +43,8 @@ import org.apache.hadoop.security.authentication.util.KerberosUtil;
 import org.apache.hadoop.security.authentication.util.Signer;
 import org.apache.hadoop.security.authentication.util.SignerException;
 import org.apache.hadoop.security.authentication.util.SignerSecretProvider;
+import org.eclipse.jetty.client.api.Request;
+import org.eclipse.jetty.http.HttpHeader;
 import sun.security.krb5.EncryptedData;
 import sun.security.krb5.EncryptionKey;
 import sun.security.krb5.internal.APReq;
@@ -72,6 +76,7 @@ import javax.servlet.http.HttpServletRequestWrapper;
 import javax.servlet.http.HttpServletResponse;
 import java.io.File;
 import java.io.IOException;
+import java.net.HttpCookie;
 import java.security.Principal;
 import java.text.SimpleDateFormat;
 import java.util.Collections;
@@ -88,6 +93,8 @@ import java.util.Set;
 import java.util.TimeZone;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
 
 @JsonTypeName("kerberos")
 public class KerberosAuthenticator implements Authenticator
@@ -95,6 +102,7 @@ public class KerberosAuthenticator implements Authenticator
   private static final Logger log = new Logger(KerberosAuthenticator.class);
   private static final Pattern HADOOP_AUTH_COOKIE_REGEX = Pattern.compile(".*p=(\\S+)&t=.*");
   public static final List<String> DEFAULT_EXCLUDED_PATHS = Collections.emptyList();
+  public static final String SIGNED_TOKEN_ATTRIBUTE = "signedToken";
 
   private final DruidNode node;
   private final String serverPrincipal;
@@ -103,6 +111,7 @@ public class KerberosAuthenticator implements Authenticator
   private final List<String> excludedPaths;
   private final String cookieSignatureSecret;
   private final String authorizerName;
+  private final String name;
   private LoginContext loginContext;
 
   @JsonCreator
@@ -113,6 +122,7 @@ public class KerberosAuthenticator implements Authenticator
       @JsonProperty("excludedPaths") List<String> excludedPaths,
       @JsonProperty("cookieSignatureSecret") String cookieSignatureSecret,
       @JsonProperty("authorizerName") String authorizerName,
+      @JsonProperty("name") String name,
       @JacksonInject @Self DruidNode node
   )
   {
@@ -123,6 +133,7 @@ public class KerberosAuthenticator implements Authenticator
     this.excludedPaths = excludedPaths == null ? DEFAULT_EXCLUDED_PATHS : excludedPaths;
     this.cookieSignatureSecret = cookieSignatureSecret;
     this.authorizerName = authorizerName;
+    this.name = Preconditions.checkNotNull(name);
   }
 
   @Override
@@ -257,7 +268,7 @@ public class KerberosAuthenticator implements Authenticator
           if (clientPrincipal != null) {
             request.setAttribute(
                 AuthConfig.DRUID_AUTHENTICATION_RESULT,
-                new AuthenticationResult(clientPrincipal, authorizerName, null)
+                new AuthenticationResult(clientPrincipal, authorizerName, name, null)
             );
           }
         }
@@ -337,11 +348,19 @@ public class KerberosAuthenticator implements Authenticator
                     !token.isExpired() && token.getExpires() > 0,
                     isHttps
                 );
+                request.setAttribute(SIGNED_TOKEN_ATTRIBUTE, tokenToCookieString(
+                    signedToken,
+                    getCookieDomain(),
+                    getCookiePath(),
+                    token.getExpires(),
+                    !token.isExpired() && token.getExpires() > 0,
+                    isHttps
+                ));
               }
               // Since this request is validated also set DRUID_AUTHENTICATION_RESULT
               request.setAttribute(
                   AuthConfig.DRUID_AUTHENTICATION_RESULT,
-                  new AuthenticationResult(token.getName(), authorizerName, null)
+                  new AuthenticationResult(token.getName(), authorizerName, name, null)
               );
               doFilter(filterChain, httpRequest, httpResponse);
             }
@@ -354,7 +373,7 @@ public class KerberosAuthenticator implements Authenticator
           errCode = HttpServletResponse.SC_FORBIDDEN;
           authenticationEx = ex;
           if (log.isDebugEnabled()) {
-            log.debug("Authentication exception: " + ex.getMessage(), ex);
+            log.debug(ex, "Authentication exception: " + ex.getMessage());
           } else {
             log.warn("Authentication exception: " + ex.getMessage());
           }
@@ -455,6 +474,22 @@ public class KerberosAuthenticator implements Authenticator
     return false;
   }
 
+  @Override
+  public void decorateProxyRequest(
+      HttpServletRequest clientRequest, HttpServletResponse proxyResponse, Request proxyRequest
+  )
+  {
+    Object cookieToken = clientRequest.getAttribute(SIGNED_TOKEN_ATTRIBUTE);
+    if (cookieToken != null && cookieToken instanceof String) {
+      log.debug("Found cookie token will attache it to proxyRequest as cookie");
+      String authResult = (String) cookieToken;
+      String existingCookies = proxyRequest.getCookies()
+                                           .stream()
+                                           .map(HttpCookie::toString)
+                                           .collect(Collectors.joining(";"));
+      proxyRequest.header(HttpHeader.COOKIE, Joiner.on(";").join(authResult, existingCookies));
+    }
+  }
 
   /**
    * Kerberos context configuration for the JDK GSS library. Copied from hadoop-auth's KerberosAuthenticationHandler.
@@ -648,6 +683,16 @@ public class KerberosAuthenticator implements Authenticator
       boolean isCookiePersistent,
       boolean isSecure
   )
+  {
+    resp.addHeader("Set-Cookie", tokenToCookieString(token, domain, path, expires, isCookiePersistent, isSecure));
+  }
+
+  private static String tokenToCookieString(
+      String token,
+      String domain, String path, long expires,
+      boolean isCookiePersistent,
+      boolean isSecure
+  )
   {
     StringBuilder sb = new StringBuilder(AuthenticatedURL.AUTH_COOKIE)
         .append("=");
@@ -675,6 +720,6 @@ public class KerberosAuthenticator implements Authenticator
     }
 
     sb.append("; HttpOnly");
-    resp.addHeader("Set-Cookie", sb.toString());
+    return sb.toString();
   }
 }

extensions-core/druid-kerberos/src/main/java/io/druid/security/kerberos/KerberosEscalator.java

@@ -22,8 +22,8 @@ package io.druid.security.kerberos;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonTypeName;
-import io.druid.java.util.http.client.HttpClient;
 import io.druid.java.util.common.logger.Logger;
+import io.druid.java.util.http.client.HttpClient;
 import io.druid.server.security.AuthenticationResult;
 import io.druid.server.security.Escalator;
 
@@ -57,6 +57,9 @@ public class KerberosEscalator implements Escalator
   @Override
   public AuthenticationResult createEscalatedAuthenticationResult()
   {
-    return new AuthenticationResult(internalClientPrincipal, authorizerName, null);
+    // if you found your self asking why the authenticatedBy field is set to null please read this:
+    // https://github.com/druid-io/druid/pull/5706#discussion_r185940889
+    return new AuthenticationResult(internalClientPrincipal, authorizerName, null, null);
   }
+
 }

indexing-service/src/test/java/io/druid/indexing/overlord/http/OverlordResourceTest.java

@@ -125,7 +125,7 @@ public class OverlordResourceTest
 
   public void expectAuthorizationTokenCheck()
   {
-    AuthenticationResult authenticationResult = new AuthenticationResult("druid", "druid", null);
+    AuthenticationResult authenticationResult = new AuthenticationResult("druid", "druid", null, null);
     EasyMock.expect(req.getAttribute(AuthConfig.DRUID_ALLOW_UNSECURED_PATH)).andReturn(null).anyTimes();
     EasyMock.expect(req.getAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED)).andReturn(null).anyTimes();
     EasyMock.expect(req.getAttribute(AuthConfig.DRUID_AUTHENTICATION_RESULT))
@@ -245,7 +245,7 @@ public class OverlordResourceTest
     Assert.assertTrue(taskRunner.getRunningTasks().size() == 3);
     List<TaskStatusPlus> responseObjects = (List) overlordResource
           .getCompleteTasks(null, req).getEntity();
-    
+
     Assert.assertEquals(2, responseObjects.size());
     Assert.assertEquals(tasksIds.get(1), responseObjects.get(0).getId());
     Assert.assertEquals(tasksIds.get(2), responseObjects.get(1).getId());
@@ -367,7 +367,7 @@ public class OverlordResourceTest
     );
     Assert.assertEquals(new TaskStatusResponse("othertask", null), taskStatusResponse2);
   }
-  
+
   @Test
   public void testGetRunningTasksByDataSource()
   {
@@ -412,7 +412,7 @@ public class OverlordResourceTest
     Assert.assertTrue(taskStorageQueryAdapter.getTask("id_2").isPresent());
     List<TaskRunnerWorkItem> responseObjects = (List) overlordResource.getRunningTasksByDataSource("ds_NA", req)
         .getEntity();
-    
+
     Assert.assertEquals(0, responseObjects.size());
   }
 

indexing-service/src/test/java/io/druid/indexing/overlord/http/OverlordTest.java

@@ -131,7 +131,7 @@ public class OverlordTest
     EasyMock.expect(req.getAttribute(AuthConfig.DRUID_ALLOW_UNSECURED_PATH)).andReturn(null).anyTimes();
     EasyMock.expect(req.getAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED)).andReturn(null).anyTimes();
     EasyMock.expect(req.getAttribute(AuthConfig.DRUID_AUTHENTICATION_RESULT)).andReturn(
-        new AuthenticationResult("druid", "druid", null)
+        new AuthenticationResult("druid", "druid", null, null)
     ).anyTimes();
     req.setAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED, true);
     EasyMock.expectLastCall().anyTimes();

indexing-service/src/test/java/io/druid/indexing/overlord/supervisor/SupervisorResourceTest.java

@@ -116,7 +116,7 @@ public class SupervisorResourceTest extends EasyMockSupport
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_ALLOW_UNSECURED_PATH)).andReturn(null).atLeastOnce();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED)).andReturn(null).atLeastOnce();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHENTICATION_RESULT)).andReturn(
-        new AuthenticationResult("druid", "druid", null)
+        new AuthenticationResult("druid", "druid", null, null)
     ).atLeastOnce();
     request.setAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED, true);
     EasyMock.expectLastCall().anyTimes();
@@ -166,7 +166,7 @@ public class SupervisorResourceTest extends EasyMockSupport
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_ALLOW_UNSECURED_PATH)).andReturn(null).atLeastOnce();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED)).andReturn(null).atLeastOnce();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHENTICATION_RESULT)).andReturn(
-        new AuthenticationResult("druid", "druid", null)
+        new AuthenticationResult("druid", "druid", null, null)
     ).atLeastOnce();
     request.setAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED, true);
     EasyMock.expectLastCall().anyTimes();
@@ -350,7 +350,7 @@ public class SupervisorResourceTest extends EasyMockSupport
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_ALLOW_UNSECURED_PATH)).andReturn(null).atLeastOnce();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED)).andReturn(null).atLeastOnce();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHENTICATION_RESULT)).andReturn(
-        new AuthenticationResult("druid", "druid", null)
+        new AuthenticationResult("druid", "druid", null, null)
     ).atLeastOnce();
     request.setAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED, true);
     EasyMock.expectLastCall().anyTimes();
@@ -463,7 +463,7 @@ public class SupervisorResourceTest extends EasyMockSupport
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_ALLOW_UNSECURED_PATH)).andReturn(null).atLeastOnce();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED)).andReturn(null).atLeastOnce();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHENTICATION_RESULT)).andReturn(
-        new AuthenticationResult("wronguser", "druid", null)
+        new AuthenticationResult("wronguser", "druid", null, null)
     ).atLeastOnce();
     request.setAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED, true);
     EasyMock.expectLastCall().anyTimes();
@@ -554,7 +554,7 @@ public class SupervisorResourceTest extends EasyMockSupport
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_ALLOW_UNSECURED_PATH)).andReturn(null).atLeastOnce();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED)).andReturn(null).atLeastOnce();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHENTICATION_RESULT)).andReturn(
-        new AuthenticationResult("druid", "druid", null)
+        new AuthenticationResult("druid", "druid", null, null)
     ).atLeastOnce();
     request.setAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED, true);
     EasyMock.expectLastCall().anyTimes();
@@ -652,7 +652,7 @@ public class SupervisorResourceTest extends EasyMockSupport
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_ALLOW_UNSECURED_PATH)).andReturn(null).atLeastOnce();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED)).andReturn(null).atLeastOnce();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHENTICATION_RESULT)).andReturn(
-        new AuthenticationResult("notdruid", "druid", null)
+        new AuthenticationResult("notdruid", "druid", null, null)
     ).atLeastOnce();
     request.setAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED, true);
     EasyMock.expectLastCall().anyTimes();

server/src/main/java/io/druid/server/AsyncQueryForwardingServlet.java

@@ -45,6 +45,9 @@ import io.druid.server.metrics.QueryCountStatsProvider;
 import io.druid.server.router.QueryHostFinder;
 import io.druid.server.router.Router;
 import io.druid.server.security.AuthConfig;
+import io.druid.server.security.AuthenticationResult;
+import io.druid.server.security.Authenticator;
+import io.druid.server.security.AuthenticatorMapper;
 import org.apache.http.client.utils.URIBuilder;
 import org.eclipse.jetty.client.HttpClient;
 import org.eclipse.jetty.client.api.Request;
@@ -113,6 +116,7 @@ public class AsyncQueryForwardingServlet extends AsyncProxyServlet implements Qu
   private final ServiceEmitter emitter;
   private final RequestLogger requestLogger;
   private final GenericQueryMetricsFactory queryMetricsFactory;
+  private final AuthenticatorMapper authenticatorMapper;
 
   private HttpClient broadcastClient;
 
@@ -126,7 +130,8 @@ public class AsyncQueryForwardingServlet extends AsyncProxyServlet implements Qu
       @Router DruidHttpClientConfig httpClientConfig,
       ServiceEmitter emitter,
       RequestLogger requestLogger,
-      GenericQueryMetricsFactory queryMetricsFactory
+      GenericQueryMetricsFactory queryMetricsFactory,
+      AuthenticatorMapper authenticatorMapper
   )
   {
     this.warehouse = warehouse;
@@ -138,6 +143,7 @@ public class AsyncQueryForwardingServlet extends AsyncProxyServlet implements Qu
     this.emitter = emitter;
     this.requestLogger = requestLogger;
     this.queryMetricsFactory = queryMetricsFactory;
+    this.authenticatorMapper = authenticatorMapper;
   }
 
   @Override
@@ -313,6 +319,22 @@ public class AsyncQueryForwardingServlet extends AsyncProxyServlet implements Qu
     // will log that on the remote node.
     clientRequest.setAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED, true);
 
+    // Check if there is an authentication result and use it to decorate the proxy request if needed.
+    AuthenticationResult authenticationResult = (AuthenticationResult) clientRequest.getAttribute(
+        AuthConfig.DRUID_AUTHENTICATION_RESULT);
+    if (authenticationResult != null && authenticationResult.getAuthenticatedBy() != null) {
+      Authenticator authenticator = authenticatorMapper.getAuthenticatorMap()
+                                                       .get(authenticationResult.getAuthenticatedBy());
+      if (authenticator != null) {
+        authenticator.decorateProxyRequest(
+            clientRequest,
+            proxyResponse,
+            proxyRequest
+        );
+      } else {
+        log.error("Can not find Authenticator with Name [%s]", authenticationResult.getAuthenticatedBy());
+      }
+    }
     super.sendProxyRequest(
         clientRequest,
         proxyResponse,

server/src/main/java/io/druid/server/security/AllowAllAuthenticator.java

@@ -38,7 +38,7 @@ public class AllowAllAuthenticator implements Authenticator
   public static final AuthenticationResult ALLOW_ALL_RESULT = new AuthenticationResult(
       AuthConfig.ALLOW_ALL_NAME,
       AuthConfig.ALLOW_ALL_NAME,
-      null
+      AuthConfig.ALLOW_ALL_NAME, null
   );
 
   @Override

server/src/main/java/io/druid/server/security/AllowOptionsResourceFilter.java

@@ -63,7 +63,7 @@ public class AllowOptionsResourceFilter implements Filter
         if (allowUnauthenticatedHttpOptions) {
           httpReq.setAttribute(
               AuthConfig.DRUID_AUTHENTICATION_RESULT,
-              new AuthenticationResult(AuthConfig.ALLOW_ALL_NAME, AuthConfig.ALLOW_ALL_NAME, null)
+              new AuthenticationResult(AuthConfig.ALLOW_ALL_NAME, AuthConfig.ALLOW_ALL_NAME, null, null)
           );
         } else {
           ((HttpServletResponse) response).sendError(HttpServletResponse.SC_UNAUTHORIZED);

server/src/main/java/io/druid/server/security/AuthenticationResult.java

@@ -37,6 +37,15 @@ public class AuthenticationResult
    */
   private final String authorizerName;
 
+
+  /**
+   * Name of authenticator whom created the results
+   *
+   * If you found your self asking why the authenticatedBy field can be null please read this
+   * https://github.com/druid-io/druid/pull/5706#discussion_r185940889
+   */
+  @Nullable
+  private final String authenticatedBy;
   /**
    * parameter containing additional context information from an Authenticator
    */
@@ -46,11 +55,13 @@ public class AuthenticationResult
   public AuthenticationResult(
       final String identity,
       final String authorizerName,
+      final String authenticatedBy,
       final Map<String, Object> context
   )
   {
     this.identity = identity;
     this.authorizerName = authorizerName;
+    this.authenticatedBy = authenticatedBy;
     this.context = context;
   }
 
@@ -68,4 +79,9 @@ public class AuthenticationResult
   {
     return context;
   }
+
+  public String getAuthenticatedBy()
+  {
+    return authenticatedBy;
+  }
 }

server/src/main/java/io/druid/server/security/Authenticator.java

@@ -22,9 +22,12 @@ package io.druid.server.security;
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import io.druid.server.initialization.jetty.ServletFilterHolder;
+import org.eclipse.jetty.client.api.Request;
 
 import javax.annotation.Nullable;
 import javax.servlet.Filter;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
 import java.util.Map;
 
 @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
@@ -93,4 +96,23 @@ public interface Authenticator extends ServletFilterHolder
    */
   @Nullable
   AuthenticationResult authenticateJDBCContext(Map<String, Object> context);
+
+
+  /**
+   * This is used to add some Headers or Authentication token/results that can be used by down stream target host.
+   * Such token can be used to authenticate the user down stream, in cases where to original credenitals
+   * are not forwardable as is and therefore the need to attach some authentication tokens by the proxy.
+   *
+   * @param clientRequest original client request processed by the upstream chain of authenticator
+   * @param proxyResponse proxy Response
+   * @param proxyRequest actual proxy request targeted to a given broker
+   */
+  default void decorateProxyRequest(
+      final HttpServletRequest clientRequest,
+      final HttpServletResponse proxyResponse,
+      final Request proxyRequest
+  )
+  {
+    //noop
+  }
 }

server/src/main/java/io/druid/server/security/Escalator.java

@@ -50,4 +50,5 @@ public interface Escalator
    * @return an AuthenticationResult representing the identity of the internal system user.
    */
   AuthenticationResult createEscalatedAuthenticationResult();
+
 }

server/src/main/java/io/druid/server/security/UnsecuredResourceFilter.java

@@ -48,7 +48,7 @@ public class UnsecuredResourceFilter implements Filter
     // but the value doesn't matter since we skip authorization checks for requests that go through this filter
     servletRequest.setAttribute(
         AuthConfig.DRUID_AUTHENTICATION_RESULT,
-        new AuthenticationResult(AuthConfig.ALLOW_ALL_NAME, AuthConfig.ALLOW_ALL_NAME, null)
+        new AuthenticationResult(AuthConfig.ALLOW_ALL_NAME, AuthConfig.ALLOW_ALL_NAME, AuthConfig.ALLOW_ALL_NAME, null)
     );
 
     // This request will not go to an Authorizer, so we need to set this for PreResponseAuthorizationCheckFilter

server/src/test/java/io/druid/server/AsyncQueryForwardingServletTest.java

@@ -54,6 +54,7 @@ import io.druid.server.metrics.NoopServiceEmitter;
 import io.druid.server.router.QueryHostFinder;
 import io.druid.server.router.RendezvousHashAvaticaConnectionBalancer;
 import io.druid.server.security.AllowAllAuthorizer;
+import io.druid.server.security.AuthenticatorMapper;
 import io.druid.server.security.Authorizer;
 import io.druid.server.security.AuthorizerMapper;
 import org.easymock.EasyMock;
@@ -250,7 +251,8 @@ public class AsyncQueryForwardingServletTest extends BaseJettyTest
         null,
         new NoopServiceEmitter(),
         requestLogLine -> { /* noop */ },
-        new DefaultGenericQueryMetricsFactory(jsonMapper)
+        new DefaultGenericQueryMetricsFactory(jsonMapper),
+        new AuthenticatorMapper(ImmutableMap.of())
     )
     {
       @Override
@@ -341,7 +343,8 @@ public class AsyncQueryForwardingServletTest extends BaseJettyTest
               injector.getInstance(DruidHttpClientConfig.class),
               new NoopServiceEmitter(),
               requestLogLine -> { /* noop */ },
-              new DefaultGenericQueryMetricsFactory(jsonMapper)
+              new DefaultGenericQueryMetricsFactory(jsonMapper),
+              new AuthenticatorMapper(ImmutableMap.of())
           )
           {
             @Override

server/src/test/java/io/druid/server/QueryResourceTest.java

@@ -80,7 +80,7 @@ public class QueryResourceTest
 {
   private static final QueryToolChestWarehouse warehouse = new MapQueryToolChestWarehouse(ImmutableMap.<Class<? extends Query>, QueryToolChest>of());
   private static final ObjectMapper jsonMapper = new DefaultObjectMapper();
-  private static final AuthenticationResult authenticationResult = new AuthenticationResult("druid", "druid", null);
+  private static final AuthenticationResult authenticationResult = new AuthenticationResult("druid", "druid", null, null);
 
   private final HttpServletRequest testServletRequest = EasyMock.createMock(HttpServletRequest.class);
   public static final QuerySegmentWalker testSegmentWalker = new QuerySegmentWalker()

server/src/test/java/io/druid/server/http/DatasourcesResourceTest.java

@@ -131,7 +131,7 @@ public class DatasourcesResourceTest
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_ALLOW_UNSECURED_PATH)).andReturn(null).once();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED)).andReturn(null).once();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHENTICATION_RESULT)).andReturn(
-      new AuthenticationResult("druid", "druid", null)
+      new AuthenticationResult("druid", "druid", null, null)
     ).atLeastOnce();
     request.setAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED, true);
     EasyMock.expectLastCall().times(1);
@@ -146,7 +146,7 @@ public class DatasourcesResourceTest
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_ALLOW_UNSECURED_PATH)).andReturn(null).once();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED)).andReturn(null).once();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHENTICATION_RESULT)).andReturn(
-        new AuthenticationResult("druid", "druid", null)
+        new AuthenticationResult("druid", "druid", null, null)
     ).once();
     request.setAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED, true);
     EasyMock.expectLastCall().times(1);
@@ -180,7 +180,7 @@ public class DatasourcesResourceTest
   @Test
   public void testSecuredGetFullQueryableDataSources()
   {
-    AuthenticationResult authenticationResult = new AuthenticationResult("druid", "druid", null);
+    AuthenticationResult authenticationResult = new AuthenticationResult("druid", "druid", null, null);
     // first request
     EasyMock.expect(server.getDataSources()).andReturn(
       ImmutableList.of(listDataSources.get(0), listDataSources.get(1))
@@ -284,7 +284,7 @@ public class DatasourcesResourceTest
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_ALLOW_UNSECURED_PATH)).andReturn(null).once();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED)).andReturn(null).once();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHENTICATION_RESULT)).andReturn(
-        new AuthenticationResult("druid", "druid", null)
+        new AuthenticationResult("druid", "druid", null, null)
     ).atLeastOnce();
     request.setAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED, true);
     EasyMock.expectLastCall().times(1);

server/src/test/java/io/druid/server/http/IntervalsResourceTest.java

@@ -111,7 +111,7 @@ public class IntervalsResourceTest
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_ALLOW_UNSECURED_PATH)).andReturn(null).once();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED)).andReturn(null).once();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHENTICATION_RESULT)).andReturn(
-        new AuthenticationResult("druid", "druid", null)
+        new AuthenticationResult("druid", "druid", null, null)
     ).once();
     request.setAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED, true);
     EasyMock.expectLastCall().times(1);
@@ -149,7 +149,7 @@ public class IntervalsResourceTest
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_ALLOW_UNSECURED_PATH)).andReturn(null).once();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED)).andReturn(null).once();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHENTICATION_RESULT)).andReturn(
-        new AuthenticationResult("druid", "druid", null)
+        new AuthenticationResult("druid", "druid", null, null)
     ).once();
     request.setAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED, true);
     EasyMock.expectLastCall().times(1);
@@ -181,7 +181,7 @@ public class IntervalsResourceTest
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_ALLOW_UNSECURED_PATH)).andReturn(null).once();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED)).andReturn(null).once();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHENTICATION_RESULT)).andReturn(
-        new AuthenticationResult("druid", "druid", null)
+        new AuthenticationResult("druid", "druid", null, null)
     ).once();
     request.setAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED, true);
     EasyMock.expectLastCall().times(1);
@@ -215,7 +215,7 @@ public class IntervalsResourceTest
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_ALLOW_UNSECURED_PATH)).andReturn(null).once();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED)).andReturn(null).once();
     EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHENTICATION_RESULT)).andReturn(
-        new AuthenticationResult("druid", "druid", null)
+        new AuthenticationResult("druid", "druid", null, null)
     ).once();
     request.setAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED, true);
     EasyMock.expectLastCall().times(1);

server/src/test/java/io/druid/server/http/security/PreResponseAuthorizationCheckFilterTest.java

@@ -50,7 +50,7 @@ public class PreResponseAuthorizationCheckFilterTest
   @Test
   public void testValidRequest() throws Exception
   {
-    AuthenticationResult authenticationResult = new AuthenticationResult("so-very-valid", "so-very-valid", null);
+    AuthenticationResult authenticationResult = new AuthenticationResult("so-very-valid", "so-very-valid", null, null);
 
     HttpServletRequest req = EasyMock.createStrictMock(HttpServletRequest.class);
     HttpServletResponse resp = EasyMock.createStrictMock(HttpServletResponse.class);
@@ -103,7 +103,7 @@ public class PreResponseAuthorizationCheckFilterTest
     expectedException.expect(ISE.class);
     expectedException.expectMessage("Request did not have an authorization check performed.");
 
-    AuthenticationResult authenticationResult = new AuthenticationResult("so-very-valid", "so-very-valid", null);
+    AuthenticationResult authenticationResult = new AuthenticationResult("so-very-valid", "so-very-valid", null, null);
 
     HttpServletRequest req = EasyMock.createStrictMock(HttpServletRequest.class);
     HttpServletResponse resp = EasyMock.createStrictMock(HttpServletResponse.class);
@@ -138,7 +138,7 @@ public class PreResponseAuthorizationCheckFilterTest
   public void testMissingAuthorizationCheckWithError() throws Exception
   {
     EmittingLogger.registerEmitter(EasyMock.createNiceMock(ServiceEmitter.class));
-    AuthenticationResult authenticationResult = new AuthenticationResult("so-very-valid", "so-very-valid", null);
+    AuthenticationResult authenticationResult = new AuthenticationResult("so-very-valid", "so-very-valid", null, null);
 
     HttpServletRequest req = EasyMock.createStrictMock(HttpServletRequest.class);
     HttpServletResponse resp = EasyMock.createStrictMock(HttpServletResponse.class);

server/src/test/java/io/druid/server/http/security/ResourceFilterTestHelper.java

@@ -112,7 +112,7 @@ public class ResourceFilterTestHelper
     EasyMock.expect(request.getMethod()).andReturn(requestMethod).anyTimes();
     EasyMock.expect(req.getAttribute(AuthConfig.DRUID_ALLOW_UNSECURED_PATH)).andReturn(null).anyTimes();
     EasyMock.expect(req.getAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED)).andReturn(null).anyTimes();
-    AuthenticationResult authenticationResult = new AuthenticationResult("druid", "druid", null);
+    AuthenticationResult authenticationResult = new AuthenticationResult("druid", "druid", null, null);
     EasyMock.expect(req.getAttribute(AuthConfig.DRUID_AUTHENTICATION_RESULT))
             .andReturn(authenticationResult)
             .atLeastOnce();

server/src/test/java/io/druid/server/http/security/SecuritySanityCheckFilterTest.java

@@ -59,7 +59,9 @@ public class SecuritySanityCheckFilterTest
     FilterChain filterChain = EasyMock.createStrictMock(FilterChain.class);
     ServletOutputStream outputStream = EasyMock.createNiceMock(ServletOutputStream.class);
 
-    AuthenticationResult authenticationResult = new AuthenticationResult("does-not-belong", "does-not-belong", null);
+    AuthenticationResult authenticationResult = new AuthenticationResult("does-not-belong", "does-not-belong",
+                                                                         null,
+                                                                         null);
 
     EasyMock.expect(req.getAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED)).andReturn(true).once();
     EasyMock.expect(req.getAttribute(AuthConfig.DRUID_ALLOW_UNSECURED_PATH)).andReturn(null).anyTimes();

sql/src/test/java/io/druid/sql/calcite/util/CalciteTests.java

@@ -170,7 +170,7 @@ public class CalciteTests
           @Override
           public AuthenticationResult authenticateJDBCContext(Map<String, Object> context)
           {
-            return new AuthenticationResult((String) context.get("user"), AuthConfig.ALLOW_ALL_NAME, null);
+            return new AuthenticationResult((String) context.get("user"), AuthConfig.ALLOW_ALL_NAME, null, null);
           }
         }
     );
@@ -191,13 +191,13 @@ public class CalciteTests
   public static final AuthenticationResult REGULAR_USER_AUTH_RESULT = new AuthenticationResult(
       AuthConfig.ALLOW_ALL_NAME,
       AuthConfig.ALLOW_ALL_NAME,
-      null
+      null, null
   );
 
   public static final AuthenticationResult SUPER_USER_AUTH_RESULT = new AuthenticationResult(
       TEST_SUPERUSER_NAME,
       AuthConfig.ALLOW_ALL_NAME,
-      null
+      null, null
   );
 
   private static final String TIMESTAMP_COLUMN = "t";