[broker][authentication]Support pass http auth status

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

@@ -104,6 +104,14 @@ default AuthenticationState newAuthState(AuthData authData,
         return new OneStageAuthenticationState(authData, remoteAddress, sslSession, this);
     }
 
+    /**
+     * Create an http authentication data State use passed in AuthenticationDataSource.
+     */
+    default AuthenticationState newHttpAuthState(HttpServletRequest request)
+            throws AuthenticationException {
+        return new OneStageAuthenticationState(request, this);
+    }
+
     /**
      * Validate the authentication for the given credentials with the specified authentication data.
      *

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

@@ -190,6 +190,37 @@ public AuthenticationState newAuthState(AuthData authData, SocketAddress remoteA
         }
     }
 
+    @Override
+    public AuthenticationState newHttpAuthState(HttpServletRequest request) throws AuthenticationException {
+        final List<AuthenticationState> states = new ArrayList<>(providers.size());
+
+        AuthenticationException authenticationException = null;
+        try {
+            applyAuthProcessor(
+                    providers,
+                    provider -> {
+                        AuthenticationState state = provider.newHttpAuthState(request);
+                        states.add(state);
+                        return state;
+                    }
+            );
+        } catch (AuthenticationException ae) {
+            authenticationException = ae;
+        }
+        if (states.isEmpty()) {
+            log.debug("Failed to initialize a new http auth state from {}",
+                    request.getRemoteHost(), authenticationException);
+            if (authenticationException != null) {
+                throw authenticationException;
+            } else {
+                throw new AuthenticationException(
+                        "Failed to initialize a new http auth state from " + request.getRemoteHost());
+            }
+        } else {
+            return new AuthenticationListState(states);
+        }
+    }
+
     @Override
     public boolean authenticateHttpRequest(HttpServletRequest request, HttpServletResponse response) throws Exception {
         Boolean authenticated = applyAuthProcessor(

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

@@ -87,7 +87,6 @@ public AuthenticationService(ServiceConfiguration conf) throws PulsarServerExcep
 
     public String authenticateHttpRequest(HttpServletRequest request) throws AuthenticationException {
         AuthenticationException authenticationException = null;
-        AuthenticationDataSource authData = new AuthenticationDataHttps(request);
         String authMethodName = request.getHeader("X-Pulsar-Auth-Method-Name");
 
         if (authMethodName != null) {
@@ -96,6 +95,8 @@ public String authenticateHttpRequest(HttpServletRequest request) throws Authent
                 throw new AuthenticationException(
                         String.format("Unsupported authentication method: [%s].", authMethodName));
             }
+            AuthenticationState authenticationState = providerToUse.newHttpAuthState(request);
+            AuthenticationDataSource authData = authenticationState.getAuthDataSource();
             try {
                 return providerToUse.authenticate(authData);
             } catch (AuthenticationException e) {
@@ -109,6 +110,8 @@ public String authenticateHttpRequest(HttpServletRequest request) throws Authent
         } else {
             for (AuthenticationProvider provider : providers.values()) {
                 try {
+                    AuthenticationState authenticationState = provider.newHttpAuthState(request);
+                    AuthenticationDataSource authData = authenticationState.getAuthDataSource();
                     return provider.authenticate(authData);
                 } catch (AuthenticationException e) {
                     if (LOG.isDebugEnabled()) {

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

@@ -22,6 +22,7 @@
 import java.net.SocketAddress;
 import javax.naming.AuthenticationException;
 import javax.net.ssl.SSLSession;
+import javax.servlet.http.HttpServletRequest;
 import org.apache.pulsar.common.api.AuthData;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
@@ -46,6 +47,12 @@ public OneStageAuthenticationState(AuthData authData,
         this.authRole = provider.authenticate(authenticationDataSource);
     }
 
+    public OneStageAuthenticationState(HttpServletRequest request, AuthenticationProvider provider)
+            throws AuthenticationException {
+        this.authenticationDataSource = new AuthenticationDataHttps(request);
+        this.authRole = provider.authenticate(authenticationDataSource);
+    }
+
     @Override
     public String getAuthRole() {
         return authRole;

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/AuthenticationFilter.java

@@ -33,6 +33,7 @@
 
 import org.apache.pulsar.broker.authentication.AuthenticationDataHttps;
 import org.apache.pulsar.broker.authentication.AuthenticationService;
+import org.apache.pulsar.broker.authentication.AuthenticationState;
 import org.apache.pulsar.common.sasl.SaslConstants;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -76,8 +77,15 @@ public void doFilter(ServletRequest request, ServletResponse response, FilterCha
                 // not sasl type, return role directly.
                 String role = authenticationService.authenticateHttpRequest((HttpServletRequest) request);
                 request.setAttribute(AuthenticatedRoleAttributeName, role);
-                request.setAttribute(AuthenticatedDataAttributeName,
-                    new AuthenticationDataHttps((HttpServletRequest) request));
+                String authMethodName = httpRequest.getHeader("X-Pulsar-Auth-Method-Name");
+                if (authMethodName != null && authenticationService.getAuthenticationProvider(authMethodName) != null) {
+                    AuthenticationState authenticationState = authenticationService
+                            .getAuthenticationProvider(authMethodName).newHttpAuthState(httpRequest);
+                    request.setAttribute(AuthenticatedDataAttributeName, authenticationState.getAuthDataSource());
+                } else {
+                    request.setAttribute(AuthenticatedDataAttributeName,
+                            new AuthenticationDataHttps((HttpServletRequest) request));
+                }
                 if (LOG.isDebugEnabled()) {
                     LOG.debug("[{}] Authenticated HTTP request with role {}", request.getRemoteAddr(), role);
                 }

pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authentication/AuthenticationProviderListTest.java

@@ -19,6 +19,8 @@
 package org.apache.pulsar.broker.authentication;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+import javax.servlet.http.HttpServletRequest;
+import static org.mockito.Mockito.mock;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;
@@ -165,6 +167,14 @@ private AuthenticationState newAuthState(String token, String expectedSubject) t
         return authState;
     }
 
+    private AuthenticationState newHttpAuthState(HttpServletRequest request, String expectedSubject) throws Exception {
+        AuthenticationState authState = authProvider.newHttpAuthState(request);
+        assertEquals(authState.getAuthRole(), expectedSubject);
+        assertTrue(authState.isComplete());
+        assertFalse(authState.isExpired());
+        return authState;
+    }
+
     private void verifyAuthStateExpired(AuthenticationState authState, String expectedSubject)
         throws Exception {
         assertEquals(authState.getAuthRole(), expectedSubject);
@@ -188,4 +198,20 @@ public void testNewAuthState() throws Exception {
 
     }
 
+    @Test
+    public void testNewHttpAuthState() throws Exception {
+        HttpServletRequest request = mock(HttpServletRequest.class);
+        AuthenticationState authStateAA = newHttpAuthState(request, SUBJECT_A);
+        AuthenticationState authStateAB = newHttpAuthState(request, SUBJECT_B);
+        AuthenticationState authStateBA = newHttpAuthState(request, SUBJECT_A);
+        AuthenticationState authStateBB = newHttpAuthState(request, SUBJECT_B);
+
+        Thread.sleep(TimeUnit.SECONDS.toMillis(6));
+
+        verifyAuthStateExpired(authStateAA, SUBJECT_A);
+        verifyAuthStateExpired(authStateAB, SUBJECT_B);
+        verifyAuthStateExpired(authStateBA, SUBJECT_A);
+        verifyAuthStateExpired(authStateBB, SUBJECT_B);
+    }
+
 }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataBasic.java

@@ -29,12 +29,13 @@
 
 public class AuthenticationDataBasic implements AuthenticationDataProvider {
     private static final String HTTP_HEADER_NAME = "Authorization";
+    private static final String PULSAR_AUTH_METHOD_NAME = "X-Pulsar-Auth-Method-Name";
     private String httpAuthToken;
     private String commandAuthToken;
 
     public AuthenticationDataBasic(String userId, String password) {
         httpAuthToken = "Basic " + Base64.getEncoder().encodeToString((userId + ":" + password).getBytes());
-        commandAuthToken = userId+":"+password;
+        commandAuthToken = userId + ":" + password;
     }
 
     @Override
@@ -46,6 +47,7 @@ public boolean hasDataForHttp() {
     public Set<Map.Entry<String, String>> getHttpHeaders() {
         Map<String, String> headers = new HashMap<>();
         headers.put(HTTP_HEADER_NAME, httpAuthToken);
+        headers.put(PULSAR_AUTH_METHOD_NAME, "basic");
         return headers.entrySet();
     }
 

pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataTls.java

@@ -24,6 +24,9 @@
 import java.security.PrivateKey;
 import java.security.cert.Certificate;
 import java.security.cert.X509Certificate;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
 import java.util.function.Supplier;
 
 import org.apache.pulsar.client.api.AuthenticationDataProvider;
@@ -35,6 +38,8 @@
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 
 public class AuthenticationDataTls implements AuthenticationDataProvider {
+    private static final String PULSAR_AUTH_METHOD_NAME = "X-Pulsar-Auth-Method-Name";
+
     private static final long serialVersionUID = 1L;
     protected X509Certificate[] tlsCertificates;
     protected PrivateKey tlsPrivateKey;
@@ -88,6 +93,11 @@ public boolean hasDataForTls() {
         return true;
     }
 
+    @Override
+    public Set<Map.Entry<String, String>> getHttpHeaders() {
+        return Collections.singletonMap(PULSAR_AUTH_METHOD_NAME, "tls").entrySet();
+    }
+
     @Override
     public Certificate[] getTlsCertificates() {
         if (certFile != null && certFile.checkAndRefresh()) {

pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataToken.java

@@ -19,7 +19,7 @@
 
 package org.apache.pulsar.client.impl.auth;
 
-import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
 import java.util.function.Supplier;
@@ -29,6 +29,8 @@
 public class AuthenticationDataToken implements AuthenticationDataProvider {
     public static final String HTTP_HEADER_NAME = "Authorization";
 
+    private static final String PULSAR_AUTH_METHOD_NAME = "X-Pulsar-Auth-Method-Name";
+
     private final Supplier<String> tokenSupplier;
 
     public AuthenticationDataToken(Supplier<String> tokenSupplier) {
@@ -42,7 +44,10 @@ public boolean hasDataForHttp() {
 
     @Override
     public Set<Map.Entry<String, String>> getHttpHeaders() {
-        return Collections.singletonMap(HTTP_HEADER_NAME, "Bearer " + getToken()).entrySet();
+        Map<String, String> headers = new HashMap<>();
+        headers.put(PULSAR_AUTH_METHOD_NAME, "token");
+        headers.put(HTTP_HEADER_NAME, "Bearer " + getToken());
+        return headers.entrySet();
     }
 
     @Override

pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/oauth2/AuthenticationDataOAuth2.java

@@ -18,7 +18,7 @@
  */
 package org.apache.pulsar.client.impl.auth.oauth2;
 
-import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
 import org.apache.pulsar.client.api.AuthenticationDataProvider;
@@ -28,13 +28,12 @@
  */
 class AuthenticationDataOAuth2 implements AuthenticationDataProvider {
     public static final String HTTP_HEADER_NAME = "Authorization";
+    private static final String PULSAR_AUTH_METHOD_NAME = "X-Pulsar-Auth-Method-Name";
 
     private final String accessToken;
-    private final Set<Map.Entry<String, String>> headers;
 
     public AuthenticationDataOAuth2(String accessToken) {
         this.accessToken = accessToken;
-        this.headers = Collections.singletonMap(HTTP_HEADER_NAME, "Bearer " + accessToken).entrySet();
     }
 
     @Override
@@ -44,7 +43,10 @@ public boolean hasDataForHttp() {
 
     @Override
     public Set<Map.Entry<String, String>> getHttpHeaders() {
-        return this.headers;
+        Map<String, String> headers = new HashMap<>();
+        headers.put(HTTP_HEADER_NAME, "Bearer " + accessToken);
+        headers.put(PULSAR_AUTH_METHOD_NAME, "token");
+        return headers.entrySet();
     }
 
     @Override