diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProvider.java index 6a25cc57db2c5..840b5cb057db6 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProvider.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProvider.java @@ -101,6 +101,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. * diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderList.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderList.java index ecb2a078e52d6..c7a0387b66784 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderList.java +++ b/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 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( diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderToken.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderToken.java index 451c63fb807da..164d5ee672ce6 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderToken.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderToken.java @@ -38,6 +38,7 @@ import java.util.List; import javax.naming.AuthenticationException; import javax.net.ssl.SSLSession; +import javax.servlet.http.HttpServletRequest; import org.apache.commons.lang3.StringUtils; import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.authentication.metrics.AuthenticationMetrics; @@ -165,6 +166,11 @@ public AuthenticationState newAuthState(AuthData authData, SocketAddress remoteA return new TokenAuthenticationState(this, authData, remoteAddress, sslSession); } + @Override + public AuthenticationState newHttpAuthState(HttpServletRequest request) throws AuthenticationException { + return new TokenAuthenticationState(this, request); + } + public static String getToken(AuthenticationDataSource authData) throws AuthenticationException { if (authData.hasDataFromCommand()) { // Authenticate Pulsar binary connection @@ -310,8 +316,6 @@ private static final class TokenAuthenticationState implements AuthenticationSta private final AuthenticationProviderToken provider; private AuthenticationDataSource authenticationDataSource; private Jwt jwt; - private final SocketAddress remoteAddress; - private final SSLSession sslSession; private long expiration; TokenAuthenticationState( @@ -320,9 +324,24 @@ private static final class TokenAuthenticationState implements AuthenticationSta SocketAddress remoteAddress, SSLSession sslSession) throws AuthenticationException { this.provider = provider; - this.remoteAddress = remoteAddress; - this.sslSession = sslSession; - this.authenticate(authData); + String token = new String(authData.getBytes(), UTF_8); + this.authenticationDataSource = new AuthenticationDataCommand(token, remoteAddress, sslSession); + this.checkExpiration(token); + } + + TokenAuthenticationState( + AuthenticationProviderToken provider, + HttpServletRequest request) throws AuthenticationException { + this.provider = provider; + String httpHeaderValue = request.getHeader(HTTP_HEADER_NAME); + if (httpHeaderValue == null || !httpHeaderValue.startsWith(HTTP_HEADER_VALUE_PREFIX)) { + throw new AuthenticationException("Invalid HTTP Authorization header"); + } + + // Remove prefix + String token = httpHeaderValue.substring(HTTP_HEADER_VALUE_PREFIX.length()); + this.authenticationDataSource = new AuthenticationDataHttps(request); + this.checkExpiration(token); } @Override @@ -330,21 +349,25 @@ public String getAuthRole() throws AuthenticationException { return provider.getPrincipal(jwt); } + /** + * @param authData Authentication data. + * @return null. Explanation of returning null values, {@link AuthenticationState#authenticateAsync(AuthData)} + * @throws AuthenticationException + */ @Override public AuthData authenticate(AuthData authData) throws AuthenticationException { - String token = new String(authData.getBytes(), UTF_8); + // There's no additional auth stage required + return null; + } + private void checkExpiration(String token) throws AuthenticationException { this.jwt = provider.authenticateToken(token); - this.authenticationDataSource = new AuthenticationDataCommand(token, remoteAddress, sslSession); if (jwt.getBody().getExpiration() != null) { this.expiration = jwt.getBody().getExpiration().getTime(); } else { // Disable expiration this.expiration = Long.MAX_VALUE; } - - // There's no additional auth stage required - return null; } @Override diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationService.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationService.java index d3668204a7869..6fe6c5f8e7ca7 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationService.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationService.java @@ -31,6 +31,7 @@ import org.apache.commons.lang3.StringUtils; import org.apache.pulsar.broker.PulsarServerException; import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.web.AuthenticationFilter; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -84,10 +85,10 @@ public AuthenticationService(ServiceConfiguration conf) throws PulsarServerExcep } } - public String authenticateHttpRequest(HttpServletRequest request) throws AuthenticationException { + public String authenticateHttpRequest(HttpServletRequest request, AuthenticationDataSource authData) + throws AuthenticationException { AuthenticationException authenticationException = null; - AuthenticationDataSource authData = new AuthenticationDataHttps(request); - String authMethodName = request.getHeader("X-Pulsar-Auth-Method-Name"); + String authMethodName = request.getHeader(AuthenticationFilter.PULSAR_AUTH_METHOD_NAME); if (authMethodName != null) { AuthenticationProvider providerToUse = providers.get(authMethodName); @@ -96,6 +97,11 @@ public String authenticateHttpRequest(HttpServletRequest request) throws Authent String.format("Unsupported authentication method: [%s].", authMethodName)); } try { + if (authData == null) { + AuthenticationState authenticationState = providerToUse.newHttpAuthState(request); + authData = authenticationState.getAuthDataSource(); + } + // Backward compatible, the authData value was null in the previous implementation return providerToUse.authenticate(authData); } catch (AuthenticationException e) { if (LOG.isDebugEnabled()) { @@ -109,7 +115,8 @@ public String authenticateHttpRequest(HttpServletRequest request) throws Authent } else { for (AuthenticationProvider provider : providers.values()) { try { - return provider.authenticate(authData); + AuthenticationState authenticationState = provider.newHttpAuthState(request); + return provider.authenticate(authenticationState.getAuthDataSource()); } catch (AuthenticationException e) { if (LOG.isDebugEnabled()) { LOG.debug("Authentication failed for provider " + provider.getAuthMethodName() + ": " @@ -137,6 +144,15 @@ public String authenticateHttpRequest(HttpServletRequest request) throws Authent } } + /** + * Mark this function as deprecated, it is recommended to use a method with the AuthenticationDataSource + * signature to implement it. + */ + @Deprecated + public String authenticateHttpRequest(HttpServletRequest request) throws AuthenticationException { + return authenticateHttpRequest(request, null); + } + public AuthenticationProvider getAuthenticationProvider(String authMethodName) { return providers.get(authMethodName); } diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/OneStageAuthenticationState.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/OneStageAuthenticationState.java index f2cb251eaf82b..6e5b889ff3c9c 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/OneStageAuthenticationState.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/OneStageAuthenticationState.java @@ -23,6 +23,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; /** @@ -45,6 +46,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; diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/AuthenticationFilter.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/AuthenticationFilter.java index eacefbe49b59a..6c69bce42685e 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/AuthenticationFilter.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/AuthenticationFilter.java @@ -30,6 +30,7 @@ import javax.servlet.http.HttpServletResponse; 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; @@ -44,6 +45,8 @@ public class AuthenticationFilter implements Filter { public static final String AuthenticatedRoleAttributeName = AuthenticationFilter.class.getName() + "-role"; public static final String AuthenticatedDataAttributeName = AuthenticationFilter.class.getName() + "-data"; + public static final String PULSAR_AUTH_METHOD_NAME = "X-Pulsar-Auth-Method-Name"; + public AuthenticationFilter(AuthenticationService authenticationService) { this.authenticationService = authenticationService; @@ -71,10 +74,21 @@ public void doFilter(ServletRequest request, ServletResponse response, FilterCha if (!isSaslRequest(httpRequest)) { // not sasl type, return role directly. - String role = authenticationService.authenticateHttpRequest((HttpServletRequest) request); + String authMethodName = httpRequest.getHeader(PULSAR_AUTH_METHOD_NAME); + String role; + if (authMethodName != null && authenticationService.getAuthenticationProvider(authMethodName) != null) { + AuthenticationState authenticationState = authenticationService + .getAuthenticationProvider(authMethodName).newHttpAuthState(httpRequest); + request.setAttribute(AuthenticatedDataAttributeName, authenticationState.getAuthDataSource()); + role = authenticationService.authenticateHttpRequest( + (HttpServletRequest) request, authenticationState.getAuthDataSource()); + } else { + request.setAttribute(AuthenticatedDataAttributeName, + new AuthenticationDataHttps((HttpServletRequest) request)); + role = authenticationService.authenticateHttpRequest((HttpServletRequest) request); + } request.setAttribute(AuthenticatedRoleAttributeName, role); - request.setAttribute(AuthenticatedDataAttributeName, - new AuthenticationDataHttps((HttpServletRequest) request)); + if (LOG.isDebugEnabled()) { LOG.debug("[{}] Authenticated HTTP request with role {}", request.getRemoteAddr(), role); } diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authentication/AuthenticationProviderListTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authentication/AuthenticationProviderListTest.java index b4bf974ff4f54..7d7e0ca92f61a 100644 --- a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authentication/AuthenticationProviderListTest.java +++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authentication/AuthenticationProviderListTest.java @@ -19,6 +19,9 @@ 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.mockito.Mockito.when; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertTrue; @@ -165,6 +168,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 +199,38 @@ public void testNewAuthState() throws Exception { } + @Test + public void testNewHttpAuthState() throws Exception { + HttpServletRequest requestAA = mock(HttpServletRequest.class); + when(requestAA.getRemoteAddr()).thenReturn("127.0.0.1"); + when(requestAA.getRemotePort()).thenReturn(8080); + when(requestAA.getHeader("Authorization")).thenReturn("Bearer " + expiringTokenAA); + AuthenticationState authStateAA = newHttpAuthState(requestAA, SUBJECT_A); + + HttpServletRequest requestAB = mock(HttpServletRequest.class); + when(requestAB.getRemoteAddr()).thenReturn("127.0.0.1"); + when(requestAB.getRemotePort()).thenReturn(8080); + when(requestAB.getHeader("Authorization")).thenReturn("Bearer " + expiringTokenAB); + AuthenticationState authStateAB = newHttpAuthState(requestAB, SUBJECT_B); + + HttpServletRequest requestBA = mock(HttpServletRequest.class); + when(requestBA.getRemoteAddr()).thenReturn("127.0.0.1"); + when(requestBA.getRemotePort()).thenReturn(8080); + when(requestBA.getHeader("Authorization")).thenReturn("Bearer " + expiringTokenBA); + AuthenticationState authStateBA = newHttpAuthState(requestBA, SUBJECT_A); + + HttpServletRequest requestBB = mock(HttpServletRequest.class); + when(requestBB.getRemoteAddr()).thenReturn("127.0.0.1"); + when(requestBB.getRemotePort()).thenReturn(8080); + when(requestBB.getHeader("Authorization")).thenReturn("Bearer " + expiringTokenBB); + AuthenticationState authStateBB = newHttpAuthState(requestBB, SUBJECT_B); + + Thread.sleep(TimeUnit.SECONDS.toMillis(6)); + + verifyAuthStateExpired(authStateAA, SUBJECT_A); + verifyAuthStateExpired(authStateAB, SUBJECT_B); + verifyAuthStateExpired(authStateBA, SUBJECT_A); + verifyAuthStateExpired(authStateBB, SUBJECT_B); + } + } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java index 1e8d8dbec7c3f..2d6f44ce1c815 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java @@ -49,7 +49,6 @@ import org.apache.commons.lang3.tuple.Pair; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; -import org.apache.pulsar.broker.authentication.AuthenticationDataHttps; import org.apache.pulsar.broker.authentication.AuthenticationDataSource; import org.apache.pulsar.broker.authorization.AuthorizationService; import org.apache.pulsar.broker.namespace.LookupOptions; @@ -137,8 +136,8 @@ public String originalPrincipal() { return httpRequest.getHeader(ORIGINAL_PRINCIPAL_HEADER); } - public AuthenticationDataHttps clientAuthData() { - return (AuthenticationDataHttps) httpRequest.getAttribute(AuthenticationFilter.AuthenticatedDataAttributeName); + public AuthenticationDataSource clientAuthData() { + return (AuthenticationDataSource) httpRequest.getAttribute(AuthenticationFilter.AuthenticatedDataAttributeName); } public boolean isRequestHttps() { @@ -1175,7 +1174,8 @@ && pulsar().getBrokerService().isAuthorizationEnabled()) { return FutureUtil.failedFuture( new RestException(Status.UNAUTHORIZED, "Need to authenticate to perform the request")); } - AuthenticationDataHttps authData = clientAuthData(); + + AuthenticationDataSource authData = clientAuthData(); authData.setSubscription(subscription); return pulsar().getBrokerService().getAuthorizationService() .allowTopicOperationAsync(topicName, operation, originalPrincipal(), clientAppId(), authData) diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/AuthenticationDataProvider.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/AuthenticationDataProvider.java index 4624821f5e0c0..c0155fe66c05f 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/AuthenticationDataProvider.java +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/AuthenticationDataProvider.java @@ -36,6 +36,8 @@ @InterfaceAudience.LimitedPrivate @InterfaceStability.Stable public interface AuthenticationDataProvider extends Serializable { + + String PULSAR_AUTH_METHOD_NAME = "X-Pulsar-Auth-Method-Name"; /* * TLS */ diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationBasic.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationBasic.java index ad57212760a35..567771dc2aaea 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationBasic.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationBasic.java @@ -29,6 +29,7 @@ import org.apache.pulsar.client.api.PulsarClientException; public class AuthenticationBasic implements Authentication, EncodedAuthenticationParameterSupport { + static final String AUTH_METHOD_NAME = "basic"; private String userId; private String password; @@ -39,7 +40,7 @@ public void close() throws IOException { @Override public String getAuthMethodName() { - return "basic"; + return AUTH_METHOD_NAME; } @Override diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataBasic.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataBasic.java index 9e187510a3777..ba9e728a2d739 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataBasic.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataBasic.java @@ -20,6 +20,7 @@ package org.apache.pulsar.client.impl.auth; import java.util.Base64; +import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.Set; @@ -29,10 +30,14 @@ public class AuthenticationDataBasic implements AuthenticationDataProvider { private static final String HTTP_HEADER_NAME = "Authorization"; private String httpAuthToken; private String commandAuthToken; + private Map headers = new HashMap<>(); public AuthenticationDataBasic(String userId, String password) { httpAuthToken = "Basic " + Base64.getEncoder().encodeToString((userId + ":" + password).getBytes()); commandAuthToken = userId + ":" + password; + headers.put(HTTP_HEADER_NAME, httpAuthToken); + headers.put(PULSAR_AUTH_METHOD_NAME, AuthenticationBasic.AUTH_METHOD_NAME); + this.headers = Collections.unmodifiableMap(this.headers); } @Override @@ -42,9 +47,7 @@ public boolean hasDataForHttp() { @Override public Set> getHttpHeaders() { - Map headers = new HashMap<>(); - headers.put(HTTP_HEADER_NAME, httpAuthToken); - return headers.entrySet(); + return this.headers.entrySet(); } @Override diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataTls.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataTls.java index 9c3627a308649..14e67ba4ddf65 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataTls.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataTls.java @@ -25,6 +25,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; import org.apache.pulsar.common.util.FileModifiedTimeUpdater; @@ -42,6 +45,8 @@ public class AuthenticationDataTls implements AuthenticationDataProvider { @SuppressFBWarnings(value = "SE_TRANSIENT_FIELD_NOT_RESTORED", justification = "Using custom serializer which Findbugs can't detect") private transient Supplier certStreamProvider, keyStreamProvider, trustStoreStreamProvider; + private static final Map headers = Collections.singletonMap( + PULSAR_AUTH_METHOD_NAME, AuthenticationTls.AUTH_METHOD_NAME); public AuthenticationDataTls(String certFilePath, String keyFilePath) throws KeyManagementException { if (certFilePath == null) { @@ -87,6 +92,11 @@ public boolean hasDataForTls() { return true; } + @Override + public Set> getHttpHeaders() { + return headers.entrySet(); + } + @Override public Certificate[] getTlsCertificates() { if (certFile != null && certFile.checkAndRefresh()) { diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataToken.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataToken.java index a718f05988469..b69222a57c9b0 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataToken.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataToken.java @@ -20,6 +20,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; @@ -27,11 +28,14 @@ public class AuthenticationDataToken implements AuthenticationDataProvider { public static final String HTTP_HEADER_NAME = "Authorization"; - private final Supplier tokenSupplier; + private Map headers = new HashMap<>(); public AuthenticationDataToken(Supplier tokenSupplier) { this.tokenSupplier = tokenSupplier; + headers.put(PULSAR_AUTH_METHOD_NAME, AuthenticationToken.AUTH_METHOD_NAME); + headers.put(HTTP_HEADER_NAME, "Bearer " + getToken()); + this.headers = Collections.unmodifiableMap(this.headers); } @Override @@ -41,7 +45,7 @@ public boolean hasDataForHttp() { @Override public Set> getHttpHeaders() { - return Collections.singletonMap(HTTP_HEADER_NAME, "Bearer " + getToken()).entrySet(); + return this.headers.entrySet(); } @Override diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationTls.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationTls.java index 4e59e680736cd..37391d03cc2aa 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationTls.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationTls.java @@ -38,7 +38,7 @@ * */ public class AuthenticationTls implements Authentication, EncodedAuthenticationParameterSupport { - private static final String AUTH_NAME = "tls"; + static final String AUTH_METHOD_NAME = "tls"; private static final long serialVersionUID = 1L; private String certFilePath; @@ -73,7 +73,7 @@ public void close() throws IOException { @Override public String getAuthMethodName() { - return AUTH_NAME; + return AUTH_METHOD_NAME; } @Override diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationToken.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationToken.java index 64bbe08860c5e..8694352db0b9f 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationToken.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationToken.java @@ -39,6 +39,7 @@ * Token based authentication provider. */ public class AuthenticationToken implements Authentication, EncodedAuthenticationParameterSupport { + static final String AUTH_METHOD_NAME = "token"; private static final long serialVersionUID = 1L; private Supplier tokenSupplier = null; @@ -61,7 +62,7 @@ public void close() throws IOException { @Override public String getAuthMethodName() { - return "token"; + return AUTH_METHOD_NAME; } @Override diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/oauth2/AuthenticationDataOAuth2.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/oauth2/AuthenticationDataOAuth2.java index 59810f50a62db..788f2d5ba251f 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/oauth2/AuthenticationDataOAuth2.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/oauth2/AuthenticationDataOAuth2.java @@ -19,6 +19,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; @@ -30,11 +31,13 @@ class AuthenticationDataOAuth2 implements AuthenticationDataProvider { public static final String HTTP_HEADER_NAME = "Authorization"; private final String accessToken; - private final Set> headers; + private Map headers = new HashMap<>(); public AuthenticationDataOAuth2(String accessToken) { this.accessToken = accessToken; - this.headers = Collections.singletonMap(HTTP_HEADER_NAME, "Bearer " + accessToken).entrySet(); + headers.put(HTTP_HEADER_NAME, "Bearer " + accessToken); + headers.put(PULSAR_AUTH_METHOD_NAME, AuthenticationOAuth2.AUTH_METHOD_NAME); + this.headers = Collections.unmodifiableMap(this.headers); } @Override @@ -44,7 +47,7 @@ public boolean hasDataForHttp() { @Override public Set> getHttpHeaders() { - return this.headers; + return this.headers.entrySet(); } @Override diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/AuthenticationTokenTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/AuthenticationTokenTest.java index d5d42c9686b11..d5174a0880bf7 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/AuthenticationTokenTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/AuthenticationTokenTest.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.client.impl.auth; +import java.util.Map; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertNull; @@ -25,7 +26,6 @@ import java.io.*; import java.nio.charset.StandardCharsets; -import java.util.Collections; import java.util.function.Supplier; import org.apache.commons.io.FileUtils; @@ -34,6 +34,7 @@ import org.apache.pulsar.client.impl.PulsarClientImpl; import org.apache.pulsar.client.impl.conf.ClientConfigurationData; import org.testng.annotations.Test; +import org.testng.collections.Maps; public class AuthenticationTokenTest { @@ -51,8 +52,10 @@ public void testAuthToken() throws Exception { assertNull(authData.getTlsPrivateKey()); assertTrue(authData.hasDataForHttp()); - assertEquals(authData.getHttpHeaders(), - Collections.singletonMap("Authorization", "Bearer token-xyz").entrySet()); + Map headers = Maps.newHashMap(); + headers.put("Authorization", "Bearer token-xyz"); + headers.put("X-Pulsar-Auth-Method-Name", "token"); + assertEquals(authData.getHttpHeaders(), headers.entrySet()); authToken.close(); } @@ -78,8 +81,10 @@ public void testAuthTokenClientConfig() throws Exception { assertNull(authData.getTlsPrivateKey()); assertTrue(authData.hasDataForHttp()); - assertEquals(authData.getHttpHeaders(), - Collections.singletonMap("Authorization", "Bearer token-xyz").entrySet()); + Map headers = Maps.newHashMap(); + headers.put("Authorization", "Bearer token-xyz"); + headers.put("X-Pulsar-Auth-Method-Name", "token"); + assertEquals(authData.getHttpHeaders(), headers.entrySet()); authToken.close(); } diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/FunctionApiResource.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/FunctionApiResource.java index 03e0a74f1e83c..ce2fa77e836ea 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/FunctionApiResource.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/FunctionApiResource.java @@ -23,7 +23,7 @@ import javax.servlet.http.HttpServletRequest; import javax.ws.rs.core.Context; import javax.ws.rs.core.UriInfo; -import org.apache.pulsar.broker.authentication.AuthenticationDataHttps; +import org.apache.pulsar.broker.authentication.AuthenticationDataSource; import org.apache.pulsar.broker.web.AuthenticationFilter; import org.apache.pulsar.functions.worker.WorkerService; @@ -53,7 +53,7 @@ public String clientAppId() { : null; } - public AuthenticationDataHttps clientAuthData() { - return (AuthenticationDataHttps) httpRequest.getAttribute(AuthenticationFilter.AuthenticatedDataAttributeName); + public AuthenticationDataSource clientAuthData() { + return (AuthenticationDataSource) httpRequest.getAttribute(AuthenticationFilter.AuthenticatedDataAttributeName); } } diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java index 049c2b0caf21d..d0db047a02048 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java @@ -62,7 +62,6 @@ import org.apache.bookkeeper.clients.exceptions.NamespaceNotFoundException; import org.apache.bookkeeper.clients.exceptions.StreamNotFoundException; import org.apache.commons.io.IOUtils; -import org.apache.pulsar.broker.authentication.AuthenticationDataHttps; import org.apache.pulsar.broker.authentication.AuthenticationDataSource; import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.client.admin.internal.FunctionsImpl; @@ -393,7 +392,7 @@ public void deregisterFunction(final String tenant, final String namespace, final String componentName, final String clientRole, - AuthenticationDataHttps clientAuthenticationDataHttps) { + AuthenticationDataSource clientAuthenticationDataHttps) { if (!isWorkerServiceAvailable()) { throwUnavailableException(); @@ -1211,7 +1210,8 @@ public FunctionState getFunctionState(final String tenant, .getBytes(kv.value(), kv.value().readerIndex(), kv.value().readableBytes()), UTF_8), null, null, kv.version()); } catch (Exception e) { - value = new FunctionState(key, null, ByteBufUtil.getBytes(kv.value()), null, kv.version()); + value = new FunctionState( + key, null, ByteBufUtil.getBytes(kv.value()), null, kv.version()); } } } @@ -1345,7 +1345,7 @@ public void uploadFunction(final InputStream uploadedInputStream, final String p @Override public StreamingOutput downloadFunction(String tenant, String namespace, String componentName, - String clientRole, AuthenticationDataHttps clientAuthenticationDataHttps) { + String clientRole, AuthenticationDataSource clientAuthenticationDataHttps) { if (!isWorkerServiceAvailable()) { throwUnavailableException(); } @@ -1410,8 +1410,8 @@ private StreamingOutput getStreamingOutput(String pkgPath) { } @Override - public StreamingOutput downloadFunction(final String path, String clientRole, - AuthenticationDataHttps clientAuthenticationDataHttps) { + public StreamingOutput downloadFunction( + final String path, String clientRole, AuthenticationDataSource clientAuthenticationDataHttps) { if (!isWorkerServiceAvailable()) { throwUnavailableException(); diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java index 378eef0364ebb..5fbf98dca81f4 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java @@ -41,7 +41,6 @@ import javax.ws.rs.core.UriBuilder; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; -import org.apache.pulsar.broker.authentication.AuthenticationDataHttps; import org.apache.pulsar.broker.authentication.AuthenticationDataSource; import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.common.functions.FunctionConfig; @@ -80,7 +79,7 @@ public void registerFunction(final String tenant, final String functionPkgUrl, final FunctionConfig functionConfig, final String clientRole, - AuthenticationDataHttps clientAuthenticationDataHttps) { + AuthenticationDataSource clientAuthenticationDataHttps) { if (!isWorkerServiceAvailable()) { throwUnavailableException(); @@ -263,7 +262,7 @@ public void updateFunction(final String tenant, final String functionPkgUrl, final FunctionConfig functionConfig, final String clientRole, - AuthenticationDataHttps clientAuthenticationDataHttps, + AuthenticationDataSource clientAuthenticationDataHttps, UpdateOptionsImpl updateOptions) { if (!isWorkerServiceAvailable()) { diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SinksImpl.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SinksImpl.java index c794747ca4ec3..19a59396025d9 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SinksImpl.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SinksImpl.java @@ -38,7 +38,6 @@ import javax.ws.rs.core.Response; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; -import org.apache.pulsar.broker.authentication.AuthenticationDataHttps; import org.apache.pulsar.broker.authentication.AuthenticationDataSource; import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.common.functions.UpdateOptionsImpl; @@ -80,7 +79,7 @@ public void registerSink(final String tenant, final String sinkPkgUrl, final SinkConfig sinkConfig, final String clientRole, - AuthenticationDataHttps clientAuthenticationDataHttps) { + AuthenticationDataSource clientAuthenticationDataHttps) { if (!isWorkerServiceAvailable()) { throwUnavailableException(); @@ -263,7 +262,7 @@ public void updateSink(final String tenant, final String sinkPkgUrl, final SinkConfig sinkConfig, final String clientRole, - AuthenticationDataHttps clientAuthenticationDataHttps, + AuthenticationDataSource clientAuthenticationDataHttps, UpdateOptionsImpl updateOptions) { if (!isWorkerServiceAvailable()) { diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SourcesImpl.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SourcesImpl.java index 602881cc4367d..ee3e06533c1cd 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SourcesImpl.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SourcesImpl.java @@ -38,7 +38,6 @@ import javax.ws.rs.core.Response; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; -import org.apache.pulsar.broker.authentication.AuthenticationDataHttps; import org.apache.pulsar.broker.authentication.AuthenticationDataSource; import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.common.functions.UpdateOptionsImpl; @@ -80,7 +79,7 @@ public void registerSource(final String tenant, final String sourcePkgUrl, final SourceConfig sourceConfig, final String clientRole, - AuthenticationDataHttps clientAuthenticationDataHttps) { + AuthenticationDataSource clientAuthenticationDataHttps) { if (!isWorkerServiceAvailable()) { throwUnavailableException(); @@ -256,15 +255,15 @@ public void registerSource(final String tenant, @Override public void updateSource(final String tenant, - final String namespace, - final String sourceName, - final InputStream uploadedInputStream, - final FormDataContentDisposition fileDetail, - final String sourcePkgUrl, - final SourceConfig sourceConfig, - final String clientRole, - AuthenticationDataHttps clientAuthenticationDataHttps, - UpdateOptionsImpl updateOptions) { + final String namespace, + final String sourceName, + final InputStream uploadedInputStream, + final FormDataContentDisposition fileDetail, + final String sourcePkgUrl, + final SourceConfig sourceConfig, + final String clientRole, + AuthenticationDataSource clientAuthenticationDataHttps, + UpdateOptionsImpl updateOptions) { if (!isWorkerServiceAvailable()) { throwUnavailableException(); diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/service/api/Component.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/service/api/Component.java index df6f7f664ff8c..c305d64b9f3cf 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/service/api/Component.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/service/api/Component.java @@ -44,7 +44,21 @@ void deregisterFunction(String tenant, String namespace, String componentName, String clientRole, - AuthenticationDataHttps clientAuthenticationDataHttps); + AuthenticationDataSource clientAuthenticationDataHttps); + + @Deprecated + default void deregisterFunction(String tenant, + String namespace, + String componentName, + String clientRole, + AuthenticationDataHttps clientAuthenticationDataHttps) { + deregisterFunction( + tenant, + namespace, + componentName, + clientRole, + (AuthenticationDataSource) clientAuthenticationDataHttps); + } FunctionConfig getFunctionInfo(String tenant, String namespace, @@ -145,13 +159,34 @@ void uploadFunction(InputStream uploadedInputStream, StreamingOutput downloadFunction(String path, String clientRole, - AuthenticationDataHttps clientAuthenticationDataHttps); + AuthenticationDataSource clientAuthenticationDataHttps); + + @Deprecated + default StreamingOutput downloadFunction(String path, + String clientRole, + AuthenticationDataHttps clientAuthenticationDataHttps) { + return downloadFunction(path, clientRole, (AuthenticationDataSource) clientAuthenticationDataHttps); + } StreamingOutput downloadFunction(String tenant, String namespace, String componentName, String clientRole, - AuthenticationDataHttps clientAuthenticationDataHttps); + AuthenticationDataSource clientAuthenticationDataHttps); + + @Deprecated + default StreamingOutput downloadFunction(String tenant, + String namespace, + String componentName, + String clientRole, + AuthenticationDataHttps clientAuthenticationDataHttps) { + return downloadFunction( + tenant, + namespace, + componentName, + clientRole, + (AuthenticationDataSource) clientAuthenticationDataHttps); + } List getListOfConnectors(); diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/service/api/Functions.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/service/api/Functions.java index baac5f32a7ac2..ac77b76ec2e30 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/service/api/Functions.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/service/api/Functions.java @@ -34,6 +34,18 @@ */ public interface Functions extends Component { + /** + * Register a new function. + * @param tenant The tenant of a Pulsar Function + * @param namespace The namespace of a Pulsar Function + * @param functionName The name of a Pulsar Function + * @param uploadedInputStream Input stream of bytes + * @param fileDetail A form-data content disposition header + * @param functionPkgUrl URL path of the Pulsar Function package + * @param functionConfig Configuration of Pulsar Function + * @param clientRole Client role for running the pulsar function + * @param clientAuthenticationDataHttps Authentication status of the http client + */ void registerFunction(String tenant, String namespace, String functionName, @@ -42,8 +54,48 @@ void registerFunction(String tenant, String functionPkgUrl, FunctionConfig functionConfig, String clientRole, - AuthenticationDataHttps clientAuthenticationDataHttps); + AuthenticationDataSource clientAuthenticationDataHttps); + /** + * This method uses an incorrect signature 'AuthenticationDataHttps' that prevents the extension of auth status, + * so it is marked as deprecated and kept here only for backward compatibility. Please use the method that accepts + * the signature of the AuthenticationDataSource. + */ + @Deprecated + default void registerFunction(String tenant, + String namespace, + String functionName, + InputStream uploadedInputStream, + FormDataContentDisposition fileDetail, + String functionPkgUrl, + FunctionConfig functionConfig, + String clientRole, + AuthenticationDataHttps clientAuthenticationDataHttps) { + registerFunction( + tenant, + namespace, + functionName, + uploadedInputStream, + fileDetail, + functionPkgUrl, + functionConfig, + clientRole, + (AuthenticationDataSource) clientAuthenticationDataHttps); + } + + /** + * Update a function. + * @param tenant The tenant of a Pulsar Function + * @param namespace The namespace of a Pulsar Function + * @param functionName The name of a Pulsar Function + * @param uploadedInputStream Input stream of bytes + * @param fileDetail A form-data content disposition header + * @param functionPkgUrl URL path of the Pulsar Function package + * @param functionConfig Configuration of Pulsar Function + * @param clientRole Client role for running the Pulsar Function + * @param clientAuthenticationDataHttps Authentication status of the http client + * @param updateOptions Options while updating the function + */ void updateFunction(String tenant, String namespace, String functionName, @@ -52,9 +104,38 @@ void updateFunction(String tenant, String functionPkgUrl, FunctionConfig functionConfig, String clientRole, - AuthenticationDataHttps clientAuthenticationDataHttps, + AuthenticationDataSource clientAuthenticationDataHttps, UpdateOptionsImpl updateOptions); + /** + * This method uses an incorrect signature 'AuthenticationDataHttps' that prevents the extension of auth status, + * so it is marked as deprecated and kept here only for backward compatibility. Please use the method that accepts + * the signature of the AuthenticationDataSource. + */ + @Deprecated + default void updateFunction(String tenant, + String namespace, + String functionName, + InputStream uploadedInputStream, + FormDataContentDisposition fileDetail, + String functionPkgUrl, + FunctionConfig functionConfig, + String clientRole, + AuthenticationDataHttps clientAuthenticationDataHttps, + UpdateOptionsImpl updateOptions) { + updateFunction( + tenant, + namespace, + functionName, + uploadedInputStream, + fileDetail, + functionPkgUrl, + functionConfig, + clientRole, + (AuthenticationDataSource) clientAuthenticationDataHttps, + updateOptions); + } + void updateFunctionOnWorkerLeader(String tenant, String namespace, String functionName, diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/service/api/Sinks.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/service/api/Sinks.java index 214cde2bb45d2..d97a2856cd2c6 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/service/api/Sinks.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/service/api/Sinks.java @@ -37,6 +37,18 @@ */ public interface Sinks extends Component { + /** + * Update a function. + * @param tenant The tenant of a Pulsar Sink + * @param namespace The namespace of a Pulsar Sink + * @param sinkName The name of a Pulsar Sink + * @param uploadedInputStream Input stream of bytes + * @param fileDetail A form-data content disposition header + * @param sinkPkgUrl URL path of the Pulsar Sink package + * @param sinkConfig Configuration of Pulsar Sink + * @param clientRole Client role for running the Pulsar Sink + * @param clientAuthenticationDataHttps Authentication status of the http client + */ void registerSink(String tenant, String namespace, String sinkName, @@ -45,8 +57,48 @@ void registerSink(String tenant, String sinkPkgUrl, SinkConfig sinkConfig, String clientRole, - AuthenticationDataHttps clientAuthenticationDataHttps); + AuthenticationDataSource clientAuthenticationDataHttps); + /** + * This method uses an incorrect signature 'AuthenticationDataHttps' that prevents the extension of auth status, + * so it is marked as deprecated and kept here only for backward compatibility. Please use the method that accepts + * the signature of the AuthenticationDataSource. + */ + @Deprecated + default void registerSink(String tenant, + String namespace, + String sinkName, + InputStream uploadedInputStream, + FormDataContentDisposition fileDetail, + String sinkPkgUrl, + SinkConfig sinkConfig, + String clientRole, + AuthenticationDataHttps clientAuthenticationDataHttps) { + registerSink( + tenant, + namespace, + sinkName, + uploadedInputStream, + fileDetail, + sinkPkgUrl, + sinkConfig, + clientRole, + (AuthenticationDataSource) clientAuthenticationDataHttps); + } + + /** + * Update a function. + * @param tenant The tenant of a Pulsar Sink + * @param namespace The namespace of a Pulsar Sink + * @param sinkName The name of a Pulsar Sink + * @param uploadedInputStream Input stream of bytes + * @param fileDetail A form-data content disposition header + * @param sinkPkgUrl URL path of the Pulsar Sink package + * @param sinkConfig Configuration of Pulsar Sink + * @param clientRole Client role for running the Pulsar Sink + * @param clientAuthenticationDataHttps Authentication status of the http client + * @param updateOptions Options while updating the sink + */ void updateSink(String tenant, String namespace, String sinkName, @@ -55,9 +107,38 @@ void updateSink(String tenant, String sinkPkgUrl, SinkConfig sinkConfig, String clientRole, - AuthenticationDataHttps clientAuthenticationDataHttps, + AuthenticationDataSource clientAuthenticationDataHttps, UpdateOptionsImpl updateOptions); + /** + * This method uses an incorrect signature 'AuthenticationDataHttps' that prevents the extension of auth status, + * so it is marked as deprecated and kept here only for backward compatibility. Please use the method that accepts + * the signature of the AuthenticationDataSource. + */ + @Deprecated + default void updateSink(String tenant, + String namespace, + String sinkName, + InputStream uploadedInputStream, + FormDataContentDisposition fileDetail, + String sinkPkgUrl, + SinkConfig sinkConfig, + String clientRole, + AuthenticationDataHttps clientAuthenticationDataHttps, + UpdateOptionsImpl updateOptions) { + updateSink( + tenant, + namespace, + sinkName, + uploadedInputStream, + fileDetail, + sinkPkgUrl, + sinkConfig, + clientRole, + (AuthenticationDataSource) clientAuthenticationDataHttps, + updateOptions); + } + SinkInstanceStatusData getSinkInstanceStatus(String tenant, String namespace, String sinkName, diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/service/api/Sources.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/service/api/Sources.java index 853eb1c95637b..089115228b2d9 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/service/api/Sources.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/service/api/Sources.java @@ -37,6 +37,18 @@ */ public interface Sources extends Component { + /** + * Update a function. + * @param tenant The tenant of a Pulsar Source + * @param namespace The namespace of a Pulsar Source + * @param sourceName The name of a Pulsar Source + * @param uploadedInputStream Input stream of bytes + * @param fileDetail A form-data content disposition header + * @param sourcePkgUrl URL path of the Pulsar Source package + * @param sourceConfig Configuration of Pulsar Source + * @param clientRole Client role for running the Pulsar Source + * @param clientAuthenticationDataHttps Authentication status of the http client + */ void registerSource(String tenant, String namespace, String sourceName, @@ -45,8 +57,48 @@ void registerSource(String tenant, String sourcePkgUrl, SourceConfig sourceConfig, String clientRole, - AuthenticationDataHttps clientAuthenticationDataHttps); + AuthenticationDataSource clientAuthenticationDataHttps); + /** + * This method uses an incorrect signature 'AuthenticationDataHttps' that prevents the extension of auth status, + * so it is marked as deprecated and kept here only for backward compatibility. Please use the method that accepts + * the signature of the AuthenticationDataSource. + */ + @Deprecated + default void registerSource(String tenant, + String namespace, + String sourceName, + InputStream uploadedInputStream, + FormDataContentDisposition fileDetail, + String sourcePkgUrl, + SourceConfig sourceConfig, + String clientRole, + AuthenticationDataHttps clientAuthenticationDataHttps) { + registerSource( + tenant, + namespace, + sourceName, + uploadedInputStream, + fileDetail, + sourcePkgUrl, + sourceConfig, + clientRole, + (AuthenticationDataSource) clientAuthenticationDataHttps); + } + + /** + * Update a function. + * @param tenant The tenant of a Pulsar Source + * @param namespace The namespace of a Pulsar Source + * @param sourceName The name of a Pulsar Source + * @param uploadedInputStream Input stream of bytes + * @param fileDetail A form-data content disposition header + * @param sourcePkgUrl URL path of the Pulsar Source package + * @param sourceConfig Configuration of Pulsar Source + * @param clientRole Client role for running the Pulsar Source + * @param clientAuthenticationDataHttps Authentication status of the http client + * @param updateOptions Options while updating the source + */ void updateSource(String tenant, String namespace, String sourceName, @@ -55,9 +107,38 @@ void updateSource(String tenant, String sourcePkgUrl, SourceConfig sourceConfig, String clientRole, - AuthenticationDataHttps clientAuthenticationDataHttps, + AuthenticationDataSource clientAuthenticationDataHttps, UpdateOptionsImpl updateOptions); + /** + * This method uses an incorrect signature 'AuthenticationDataHttps' that prevents the extension of auth status, + * so it is marked as deprecated and kept here only for backward compatibility. Please use the method that accepts + * the signature of the AuthenticationDataSource. + */ + @Deprecated + default void updateSource(String tenant, + String namespace, + String sourceName, + InputStream uploadedInputStream, + FormDataContentDisposition fileDetail, + String sourcePkgUrl, + SourceConfig sourceConfig, + String clientRole, + AuthenticationDataHttps clientAuthenticationDataHttps, + UpdateOptionsImpl updateOptions) { + updateSource( + tenant, + namespace, + sourceName, + uploadedInputStream, + fileDetail, + sourcePkgUrl, + sourceConfig, + clientRole, + (AuthenticationDataSource) clientAuthenticationDataHttps, + updateOptions); + } + SourceStatus getSourceStatus(String tenant, String namespace, diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/AbstractWebSocketHandler.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/AbstractWebSocketHandler.java index 5217091b18fb4..ea4cc128a764c 100644 --- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/AbstractWebSocketHandler.java +++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/AbstractWebSocketHandler.java @@ -30,6 +30,7 @@ import org.apache.commons.lang3.StringUtils; import org.apache.pulsar.broker.authentication.AuthenticationDataHttps; import org.apache.pulsar.broker.authentication.AuthenticationDataSource; +import org.apache.pulsar.broker.authentication.AuthenticationState; import org.apache.pulsar.client.api.PulsarClientException.AuthenticationException; import org.apache.pulsar.client.api.PulsarClientException.AuthorizationException; import org.apache.pulsar.client.api.PulsarClientException.ConsumerBusyException; @@ -59,7 +60,7 @@ public abstract class AbstractWebSocketHandler extends WebSocketAdapter implemen protected final TopicName topic; protected final Map queryParams; - + private static final String PULSAR_AUTH_METHOD_NAME = "X-Pulsar-Auth-Method-Name"; public AbstractWebSocketHandler(WebSocketService service, HttpServletRequest request, @@ -76,9 +77,21 @@ public AbstractWebSocketHandler(WebSocketService service, protected boolean checkAuth(ServletUpgradeResponse response) { String authRole = ""; + String authMethodName = request.getHeader(PULSAR_AUTH_METHOD_NAME); + AuthenticationState authenticationState = null; if (service.isAuthenticationEnabled()) { try { - authRole = service.getAuthenticationService().authenticateHttpRequest(request); + if (authMethodName != null + && service.getAuthenticationService().getAuthenticationProvider(authMethodName) != null) { + authenticationState = service.getAuthenticationService() + .getAuthenticationProvider(authMethodName).newHttpAuthState(request); + } + if (authenticationState != null) { + authRole = service.getAuthenticationService() + .authenticateHttpRequest(request, authenticationState.getAuthDataSource()); + } else { + authRole = service.getAuthenticationService().authenticateHttpRequest(request); + } log.info("[{}:{}] Authenticated WebSocket client {} on topic {}", request.getRemoteAddr(), request.getRemotePort(), authRole, topic); @@ -96,7 +109,12 @@ protected boolean checkAuth(ServletUpgradeResponse response) { } if (service.isAuthorizationEnabled()) { - AuthenticationDataSource authenticationData = new AuthenticationDataHttps(request); + AuthenticationDataSource authenticationData; + if (authenticationState != null) { + authenticationData = authenticationState.getAuthDataSource(); + } else { + authenticationData = new AuthenticationDataHttps(request); + } try { if (!isAuthorized(authRole, authenticationData)) { log.warn("[{}:{}] WebSocket Client [{}] is not authorized on topic {}", request.getRemoteAddr(), diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/admin/WebSocketWebResource.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/admin/WebSocketWebResource.java index aea70bb1bb51f..b6a0f43a01aae 100644 --- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/admin/WebSocketWebResource.java +++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/admin/WebSocketWebResource.java @@ -26,6 +26,8 @@ import javax.ws.rs.core.Response.Status; import javax.ws.rs.core.UriInfo; import org.apache.pulsar.broker.authentication.AuthenticationDataHttps; +import org.apache.pulsar.broker.authentication.AuthenticationDataSource; +import org.apache.pulsar.broker.web.AuthenticationFilter; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.util.RestException; import org.apache.pulsar.websocket.WebSocketService; @@ -50,7 +52,7 @@ public class WebSocketWebResource { private WebSocketService socketService; private String clientId; - private AuthenticationDataHttps authData; + private AuthenticationDataSource authenticationDataSource; protected WebSocketService service() { if (socketService == null) { @@ -67,7 +69,18 @@ protected WebSocketService service() { public String clientAppId() { if (isBlank(clientId)) { try { - clientId = service().getAuthenticationService().authenticateHttpRequest(httpRequest); + String authMethodName = httpRequest.getHeader(AuthenticationFilter.PULSAR_AUTH_METHOD_NAME); + if (authMethodName != null + && service().getAuthenticationService().getAuthenticationProvider(authMethodName) != null) { + authenticationDataSource = service().getAuthenticationService() + .getAuthenticationProvider(authMethodName) + .newHttpAuthState(httpRequest).getAuthDataSource(); + clientId = service().getAuthenticationService().authenticateHttpRequest( + httpRequest, authenticationDataSource); + } else { + clientId = service().getAuthenticationService().authenticateHttpRequest(httpRequest); + authenticationDataSource = new AuthenticationDataHttps(httpRequest); + } } catch (AuthenticationException e) { if (service().getConfig().isAuthenticationEnabled()) { throw new RestException(Status.UNAUTHORIZED, "Failed to get clientId from request"); @@ -81,11 +94,8 @@ public String clientAppId() { return clientId; } - public AuthenticationDataHttps authData() { - if (authData == null) { - authData = new AuthenticationDataHttps(httpRequest); - } - return authData; + public AuthenticationDataSource authData() throws AuthenticationException { + return authenticationDataSource; } /** diff --git a/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/admin/WebSocketWebResourceTest.java b/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/admin/WebSocketWebResourceTest.java index 59e681de51322..6c8156ef79fe9 100644 --- a/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/admin/WebSocketWebResourceTest.java +++ b/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/admin/WebSocketWebResourceTest.java @@ -121,6 +121,10 @@ public void setup(Method method) throws Exception { // Mock ServletContext when(servletContext.getAttribute(anyString())).thenReturn(socketService); + // Mock HttpServletRequest + when(httpRequest.getRemoteAddr()).thenReturn("127.0.0.1"); + when(httpRequest.getRemotePort()).thenReturn(8080); + // Mock UriInfo when(uri.getRequestUri()).thenReturn(null);