[Authentication] Support chained authentication with same auth method name

pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/SaslAuthenticateTest.java

@@ -251,9 +251,11 @@ public void testSaslServerAndClientAuth() throws Exception {
 
         // prepare client and server side resource
         AuthenticationDataProvider dataProvider =  authSasl.getAuthData(hostName);
-        AuthenticationProviderSasl saslServer  = (AuthenticationProviderSasl)
+        AuthenticationProviderList providerList = (AuthenticationProviderList)
             (pulsar.getBrokerService().getAuthenticationService()
                 .getAuthenticationProvider(SaslConstants.AUTH_METHOD_NAME));
+        AuthenticationProviderSasl saslServer =
+            (AuthenticationProviderSasl) providerList.getProviders().get(0);
         AuthenticationState authState = saslServer.newAuthState(null, null, null);
 
         // auth between server and client.

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderList.java

@@ -0,0 +1,218 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.authentication;
+
+import java.io.IOException;
+import java.net.SocketAddress;
+import java.util.ArrayList;
+import java.util.List;
+import javax.naming.AuthenticationException;
+import javax.net.ssl.SSLSession;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.common.api.AuthData;
+
+/**
+ * An authentication provider wraps a list of auth providers.
+ */
+@Slf4j
+public class AuthenticationProviderList implements AuthenticationProvider {
+
+    private interface AuthProcessor<T, P> {
+
+        T apply(P process) throws AuthenticationException;
+
+    }
+
+    static <T, P> T applyAuthProcessor(List<P> processors, AuthProcessor<T, P> authFunc)
+        throws AuthenticationException {
+        AuthenticationException authenticationException = null;
+        for (P ap : processors) {
+            try {
+                return authFunc.apply(ap);
+            } catch (AuthenticationException ae) {
+                if (log.isDebugEnabled()) {
+                    log.debug("Authentication failed for auth provider " + ap.getClass() + ": ", ae);
+                }
+                // Store the exception so we can throw it later instead of a generic one
+                authenticationException = ae;
+            }
+        }
+
+        if (null == authenticationException) {
+            throw new AuthenticationException("Authentication required");
+        } else {
+            throw authenticationException;
+        }
+
+    }
+
+    private static class AuthenticationListState implements AuthenticationState {
+
+        private final List<AuthenticationState> states;
+        private AuthenticationState authState;
+
+        AuthenticationListState(List<AuthenticationState> states) {
+            this.states = states;
+            this.authState = states.get(0);
+        }
+
+        private AuthenticationState getAuthState() throws AuthenticationException {
+            if (authState != null) {
+                return authState;
+            } else {
+                throw new AuthenticationException("Authentication state is not initialized");
+            }
+        }
+
+        @Override
+        public String getAuthRole() throws AuthenticationException {
+            return getAuthState().getAuthRole();
+        }
+
+        @Override
+        public AuthData authenticate(AuthData authData) throws AuthenticationException {
+            return applyAuthProcessor(
+                states,
+                as -> {
+                    AuthData ad = as.authenticate(authData);
+                    AuthenticationListState.this.authState = as;
+                    return ad;
+                }
+            );
+        }
+
+        @Override
+        public AuthenticationDataSource getAuthDataSource() {
+            return authState.getAuthDataSource();
+        }
+
+        @Override
+        public boolean isComplete() {
+            return authState.isComplete();
+        }
+
+        @Override
+        public long getStateId() {
+            if (null != authState) {
+                return authState.getStateId();
+            } else {
+                return states.get(0).getStateId();
+            }
+        }
+
+        @Override
+        public boolean isExpired() {
+            return authState.isExpired();
+        }
+
+        @Override
+        public AuthData refreshAuthentication() throws AuthenticationException {
+            return getAuthState().refreshAuthentication();
+        }
+    }
+
+    private final List<AuthenticationProvider> providers;
+
+    public AuthenticationProviderList(List<AuthenticationProvider> providers) {
+        this.providers = providers;
+    }
+
+    public List<AuthenticationProvider> getProviders() {
+        return providers;
+    }
+
+    @Override
+    public void initialize(ServiceConfiguration config) throws IOException {
+        for (AuthenticationProvider ap : providers) {
+            ap.initialize(config);
+        }
+    }
+
+    @Override
+    public String getAuthMethodName() {
+        return providers.get(0).getAuthMethodName();
+    }
+
+    @Override
+    public String authenticate(AuthenticationDataSource authData) throws AuthenticationException {
+        return applyAuthProcessor(
+            providers,
+            provider -> provider.authenticate(authData)
+        );
+    }
+
+    @Override
+    public AuthenticationState newAuthState(AuthData authData, SocketAddress remoteAddress, SSLSession sslSession)
+        throws AuthenticationException {
+        final List<AuthenticationState> states = new ArrayList<>(providers.size());
+
+        AuthenticationException authenticationException = null;
+        try {
+            applyAuthProcessor(
+                providers,
+                provider -> {
+                    AuthenticationState state = provider.newAuthState(authData, remoteAddress, sslSession);
+                    states.add(state);
+                    return state;
+                }
+            );
+        } catch (AuthenticationException ae) {
+            authenticationException = ae;
+        }
+        if (states.isEmpty()) {
+            log.error("Failed to initialize a new auth state from {}", remoteAddress, authenticationException);
+            if (authenticationException != null) {
+                throw authenticationException;
+            } else {
+                throw new AuthenticationException("Failed to initialize a new auth state from " + remoteAddress);
+            }
+        } else {
+            return new AuthenticationListState(states);
+        }
+    }
+
+    @Override
+    public boolean authenticateHttpRequest(HttpServletRequest request, HttpServletResponse response) throws Exception {
+        Boolean authenticated = applyAuthProcessor(
+            providers,
+            provider -> {
+                try {
+                    return provider.authenticateHttpRequest(request, response);
+                } catch (Exception e) {
+                    if (e instanceof AuthenticationException) {
+                        throw (AuthenticationException) e;
+                    } else {
+                        throw new AuthenticationException("Failed to authentication http request");
+                    }
+                }
+            }
+        );
+        return authenticated.booleanValue();
+    }
+
+    @Override
+    public void close() throws IOException {
+        for (AuthenticationProvider provider : providers) {
+            provider.close();
+        }
+    }
+}

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderToken.java

@@ -45,6 +45,9 @@ public class AuthenticationProviderToken implements AuthenticationProvider {
     final static String HTTP_HEADER_NAME = "Authorization";
     final static String HTTP_HEADER_VALUE_PREFIX = "Bearer ";
 
+    // When symmetric key is configured
+    final static String CONF_TOKEN_SETTING_PREFIX = "";
+
     // When symmetric key is configured
     final static String CONF_TOKEN_SECRET_KEY = "tokenSecretKey";
 
@@ -71,13 +74,32 @@ public class AuthenticationProviderToken implements AuthenticationProvider {
     private String audienceClaim;
     private String audience;
 
+    // config keys
+    private String confTokenSecretKeySettingName;
+    private String confTokenPublicKeySettingName;
+    private String confTokenAuthClaimSettingName;
+    private String confTokenPublicAlgSettingName;
+    private String confTokenAudienceClaimSettingName;
+    private String confTokenAudienceSettingName;
+
     @Override
     public void close() throws IOException {
         // noop
     }
 
     @Override
     public void initialize(ServiceConfiguration config) throws IOException, IllegalArgumentException {
+        String prefix = (String) config.getProperty(CONF_TOKEN_SETTING_PREFIX);
+        if (null == prefix) {
+            prefix = "";
+        }
+        this.confTokenSecretKeySettingName = prefix + CONF_TOKEN_SECRET_KEY;
+        this.confTokenPublicKeySettingName = prefix + CONF_TOKEN_PUBLIC_KEY;
+        this.confTokenAuthClaimSettingName = prefix + CONF_TOKEN_AUTH_CLAIM;
+        this.confTokenPublicAlgSettingName = prefix + CONF_TOKEN_PUBLIC_ALG;
+        this.confTokenAudienceClaimSettingName = prefix + CONF_TOKEN_AUDIENCE_CLAIM;
+        this.confTokenAudienceSettingName = prefix + CONF_TOKEN_AUDIENCE;
+
         // we need to fetch the algorithm before we fetch the key
         this.publicKeyAlg = getPublicKeyAlgType(config);
         this.validationKey = getValidationKey(config);
@@ -184,14 +206,14 @@ private String getPrincipal(Jwt<?, Claims> jwt) {
      * Try to get the validation key for tokens from several possible config options.
      */
     private Key getValidationKey(ServiceConfiguration conf) throws IOException {
-        if (conf.getProperty(CONF_TOKEN_SECRET_KEY) != null
-                && StringUtils.isNotBlank((String) conf.getProperty(CONF_TOKEN_SECRET_KEY))) {
-            final String validationKeyConfig = (String) conf.getProperty(CONF_TOKEN_SECRET_KEY);
+        if (conf.getProperty(confTokenSecretKeySettingName) != null
+                && StringUtils.isNotBlank((String) conf.getProperty(confTokenSecretKeySettingName))) {
+            final String validationKeyConfig = (String) conf.getProperty(confTokenSecretKeySettingName);
             final byte[] validationKey = AuthTokenUtils.readKeyFromUrl(validationKeyConfig);
             return AuthTokenUtils.decodeSecretKey(validationKey);
-        } else if (conf.getProperty(CONF_TOKEN_PUBLIC_KEY) != null
-                && StringUtils.isNotBlank((String) conf.getProperty(CONF_TOKEN_PUBLIC_KEY))) {
-            final String validationKeyConfig = (String) conf.getProperty(CONF_TOKEN_PUBLIC_KEY);
+        } else if (conf.getProperty(confTokenPublicKeySettingName) != null
+                && StringUtils.isNotBlank((String) conf.getProperty(confTokenPublicKeySettingName))) {
+            final String validationKeyConfig = (String) conf.getProperty(confTokenPublicKeySettingName);
             final byte[] validationKey = AuthTokenUtils.readKeyFromUrl(validationKeyConfig);
             return AuthTokenUtils.decodePublicKey(validationKey, publicKeyAlg);
         } else {
@@ -200,18 +222,18 @@ private Key getValidationKey(ServiceConfiguration conf) throws IOException {
     }
 
     private String getTokenRoleClaim(ServiceConfiguration conf) throws IOException {
-        if (conf.getProperty(CONF_TOKEN_AUTH_CLAIM) != null
-                && StringUtils.isNotBlank((String) conf.getProperty(CONF_TOKEN_AUTH_CLAIM))) {
-            return (String) conf.getProperty(CONF_TOKEN_AUTH_CLAIM);
+        if (conf.getProperty(confTokenAuthClaimSettingName) != null
+                && StringUtils.isNotBlank((String) conf.getProperty(confTokenAuthClaimSettingName))) {
+            return (String) conf.getProperty(confTokenAuthClaimSettingName);
         } else {
             return Claims.SUBJECT;
         }
     }
 
     private SignatureAlgorithm getPublicKeyAlgType(ServiceConfiguration conf) throws IllegalArgumentException {
-        if (conf.getProperty(CONF_TOKEN_PUBLIC_ALG) != null
-                && StringUtils.isNotBlank((String) conf.getProperty(CONF_TOKEN_PUBLIC_ALG))) {
-            String alg = (String) conf.getProperty(CONF_TOKEN_PUBLIC_ALG);
+        if (conf.getProperty(confTokenPublicAlgSettingName) != null
+                && StringUtils.isNotBlank((String) conf.getProperty(confTokenPublicAlgSettingName))) {
+            String alg = (String) conf.getProperty(confTokenPublicAlgSettingName);
             try {
                 return SignatureAlgorithm.forName(alg);
             } catch (SignatureException ex) {
@@ -224,19 +246,19 @@ private SignatureAlgorithm getPublicKeyAlgType(ServiceConfiguration conf) throws
 
     // get Token Audience Claim from configuration, if not configured return null.
     private String getTokenAudienceClaim(ServiceConfiguration conf) throws IllegalArgumentException {
-        if (conf.getProperty(CONF_TOKEN_AUDIENCE_CLAIM) != null
-            && StringUtils.isNotBlank((String) conf.getProperty(CONF_TOKEN_AUDIENCE_CLAIM))) {
-            return (String) conf.getProperty(CONF_TOKEN_AUDIENCE_CLAIM);
+        if (conf.getProperty(confTokenAudienceClaimSettingName) != null
+            && StringUtils.isNotBlank((String) conf.getProperty(confTokenAudienceClaimSettingName))) {
+            return (String) conf.getProperty(confTokenAudienceClaimSettingName);
         } else {
             return null;
         }
     }
 
     // get Token Audience that stands for this broker from configuration, if not configured return null.
     private String getTokenAudience(ServiceConfiguration conf) throws IllegalArgumentException {
-        if (conf.getProperty(CONF_TOKEN_AUDIENCE) != null
-            && StringUtils.isNotBlank((String) conf.getProperty(CONF_TOKEN_AUDIENCE))) {
-            return (String) conf.getProperty(CONF_TOKEN_AUDIENCE);
+        if (conf.getProperty(confTokenAudienceSettingName) != null
+            && StringUtils.isNotBlank((String) conf.getProperty(confTokenAudienceSettingName))) {
+            return (String) conf.getProperty(confTokenAudienceSettingName);
         } else {
             return null;
         }