You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/09/11 20:11:50 UTC
(pulsar) branch master updated: [improve][broker] Support OpenID multi-role authorization and improve proxy authentication defaults (#26549)
This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 891cf2e40f05 [improve][broker] Support OpenID multi-role authorization and improve proxy authentication defaults (#26549)
891cf2e40f05 is described below
commit 891cf2e40f0581787ecf73ebb4bd5165783218be
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Fri Sep 11 23:11:43 2026 +0300
[improve][broker] Support OpenID multi-role authorization and improve proxy authentication defaults (#26549)
---
conf/broker.conf | 6 +-
conf/proxy.conf | 2 +-
conf/standalone.conf | 6 +-
deployment/terraform-ansible/templates/broker.conf | 6 +-
deployment/terraform-ansible/templates/proxy.conf | 2 +-
.../oidc/AuthenticationProviderOpenID.java | 19 +-
...uthenticationProviderOpenIDIntegrationTest.java | 124 ++++++++++++
.../ProxySaslAuthenticationTest.java | 2 +
.../apache/pulsar/broker/ServiceConfiguration.java | 8 +-
.../authentication/AuthenticationProviderList.java | 58 ++++--
.../AuthenticationProviderToken.java | 31 ++-
.../TokenAuthenticationProvider.java | 37 ++++
.../authentication/utils/AuthTokenUtils.java | 24 +++
.../authorization/AuthorizationProvider.java | 20 ++
.../broker/authorization/AuthorizationService.java | 12 +-
.../MultiRolesTokenAuthorizationProvider.java | 145 ++++++--------
.../authorization/PulsarAuthorizationProvider.java | 12 +-
.../AuthenticationProviderTokenTest.java | 28 +++
.../authorization/AuthorizationServiceTest.java | 37 ++++
.../MultiRolesTokenAuthorizationProviderTest.java | 211 +++++++++++++++++++--
.../pulsar/broker/service/BrokerService.java | 7 +-
.../pulsar/proxy/server/ProxyConfiguration.java | 2 +-
.../apache/pulsar/proxy/server/ProxyService.java | 2 +-
.../ProxyAuthenticatedProducerConsumerTest.java | 2 +
.../proxy/server/ProxyForwardAuthDataTest.java | 1 +
.../server/ProxyWithAuthorizationNegTest.java | 2 +
.../proxy/server/ProxyWithAuthorizationTest.java | 2 +
.../server/ProxyWithJwtAuthorizationTest.java | 97 ++++++++--
.../server/ProxyWithoutServiceDiscoveryTest.java | 2 +
.../apache/pulsar/websocket/WebSocketService.java | 7 +-
30 files changed, 745 insertions(+), 169 deletions(-)
diff --git a/conf/broker.conf b/conf/broker.conf
index 716527f7217f..2fa57137c138 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -855,7 +855,11 @@ proxyRoles=
# If this flag is set then the broker authenticates the original Auth data
# else it just accepts the originalPrincipal and authorizes it (if required).
-authenticateOriginalAuthData=false
+# For TLS client-certificate authentication through a proxy, set false:
+# the broker's TLS connection carries the proxy certificate, not the client certificate.
+# Also set false for SASL authentication through a proxy: the client-proxy handshake
+# cannot be replayed as a separate client-broker handshake.
+authenticateOriginalAuthData=true
# Tls cert refresh duration in seconds. Set 0 to disable the background rotation check, so the
# TLS material loaded at startup is kept until restart.
diff --git a/conf/proxy.conf b/conf/proxy.conf
index 4ec8d7d3db39..ef073d8178a6 100644
--- a/conf/proxy.conf
+++ b/conf/proxy.conf
@@ -204,7 +204,7 @@ authorizationProvider=org.apache.pulsar.broker.authorization.PulsarAuthorization
# Whether client authorization credentials are forwarded to the broker for re-authorization.
# Authentication must be enabled via authenticationEnabled=true for this to take effect.
-forwardAuthorizationCredentials=false
+forwardAuthorizationCredentials=true
### --- Authentication --- ###
diff --git a/conf/standalone.conf b/conf/standalone.conf
index 77b912e8424b..168cd6d33bf8 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -703,7 +703,11 @@ proxyRoles=
# If this flag is set then the broker authenticates the original Auth data
# else it just accepts the originalPrincipal and authorizes it (if required).
-authenticateOriginalAuthData=false
+# For TLS client-certificate authentication through a proxy, set false:
+# the broker's TLS connection carries the proxy certificate, not the client certificate.
+# Also set false for SASL authentication through a proxy: the client-proxy handshake
+# cannot be replayed as a separate client-broker handshake.
+authenticateOriginalAuthData=true
# Enable authentication
authenticationEnabled=false
diff --git a/deployment/terraform-ansible/templates/broker.conf b/deployment/terraform-ansible/templates/broker.conf
index 51d5bbae0316..8e32e48143bf 100644
--- a/deployment/terraform-ansible/templates/broker.conf
+++ b/deployment/terraform-ansible/templates/broker.conf
@@ -437,7 +437,11 @@ proxyRoles=
# If this flag is set then the broker authenticates the original Auth data
# else it just accepts the originalPrincipal and authorizes it (if required).
-authenticateOriginalAuthData=false
+# For TLS client-certificate authentication through a proxy, set false:
+# the broker's TLS connection carries the proxy certificate, not the client certificate.
+# Also set false for SASL authentication through a proxy: the client-proxy handshake
+# cannot be replayed as a separate client-broker handshake.
+authenticateOriginalAuthData=true
# Deprecated - Use webServicePortTls and brokerServicePortTls instead
tlsEnabled=false
diff --git a/deployment/terraform-ansible/templates/proxy.conf b/deployment/terraform-ansible/templates/proxy.conf
index 5a64be5b4a28..98949a46bd3e 100644
--- a/deployment/terraform-ansible/templates/proxy.conf
+++ b/deployment/terraform-ansible/templates/proxy.conf
@@ -86,7 +86,7 @@ authorizationProvider=org.apache.pulsar.broker.authorization.PulsarAuthorization
# Whether client authorization credentials are forwared to the broker for re-authorization.
# Authentication must be enabled via authenticationEnabled=true for this to take effect.
-forwardAuthorizationCredentials=false
+forwardAuthorizationCredentials=true
### --- Authentication --- ###
diff --git a/pulsar-broker-auth-oidc/src/main/java/org/apache/pulsar/broker/authentication/oidc/AuthenticationProviderOpenID.java b/pulsar-broker-auth-oidc/src/main/java/org/apache/pulsar/broker/authentication/oidc/AuthenticationProviderOpenID.java
index 44e009e8eb57..2fd204bbc738 100644
--- a/pulsar-broker-auth-oidc/src/main/java/org/apache/pulsar/broker/authentication/oidc/AuthenticationProviderOpenID.java
+++ b/pulsar-broker-auth-oidc/src/main/java/org/apache/pulsar/broker/authentication/oidc/AuthenticationProviderOpenID.java
@@ -58,10 +58,11 @@ import okhttp3.OkHttpClient;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
-import org.apache.pulsar.broker.authentication.AuthenticationProvider;
import org.apache.pulsar.broker.authentication.AuthenticationProviderToken;
import org.apache.pulsar.broker.authentication.AuthenticationState;
+import org.apache.pulsar.broker.authentication.TokenAuthenticationProvider;
import org.apache.pulsar.broker.authentication.metrics.AuthenticationMetrics;
+import org.apache.pulsar.broker.authentication.utils.AuthTokenUtils;
import org.apache.pulsar.common.api.AuthData;
import org.asynchttpclient.AsyncHttpClient;
import org.asynchttpclient.AsyncHttpClientConfig;
@@ -69,7 +70,7 @@ import org.asynchttpclient.DefaultAsyncHttpClient;
import org.asynchttpclient.DefaultAsyncHttpClientConfig;
/**
- * An {@link AuthenticationProvider} implementation that supports the usage of a JSON Web Token (JWT)
+ * A {@link TokenAuthenticationProvider} implementation that supports the usage of a JSON Web Token (JWT)
* for client authentication. This implementation retrieves the PublicKey from the JWT issuer (assuming the
* issuer is in the configured allowed list) and then uses that Public Key to verify the validity of the JWT's
* signature.
@@ -86,9 +87,8 @@ import org.asynchttpclient.DefaultAsyncHttpClientConfig;
* this RFC: https://datatracker.ietf.org/doc/html/rfc7518#section-3.1.
*/
@CustomLog
-public class AuthenticationProviderOpenID implements AuthenticationProvider {
+public class AuthenticationProviderOpenID implements TokenAuthenticationProvider {
// Must match the value used by the OAuth2 Client Plugin.
- private static final String AUTH_METHOD_NAME = "token";
// This is backed by an ObjectMapper, which is thread safe. It is an optimization
// to share this for decoding JWTs for all connections to this broker.
@@ -204,6 +204,17 @@ public class AuthenticationProviderOpenID implements AuthenticationProvider {
authenticationMetrics.recordFailure(errorCode);
}
+ @Override
+ public CompletableFuture<Set<String>> authenticateRolesAsync(AuthenticationDataSource authData, String roleClaim) {
+ try {
+ return authenticateTokenAsync(authData)
+ .thenApply(jwt -> AuthTokenUtils.rolesFromClaim(
+ jwt.getClaim(roleClaim).as(Object.class)));
+ } catch (Exception e) {
+ return CompletableFuture.failedFuture(e);
+ }
+ }
+
/**
* Authenticate the parameterized {@link AuthenticationDataSource} by verifying the issuer is an allowed issuer,
* then retrieving the JWKS URI from the issuer, then retrieving the Public key from the JWKS URI, and finally
diff --git a/pulsar-broker-auth-oidc/src/test/java/org/apache/pulsar/broker/authentication/oidc/AuthenticationProviderOpenIDIntegrationTest.java b/pulsar-broker-auth-oidc/src/test/java/org/apache/pulsar/broker/authentication/oidc/AuthenticationProviderOpenIDIntegrationTest.java
index 4ce912425510..724c4d4ae9b9 100644
--- a/pulsar-broker-auth-oidc/src/test/java/org/apache/pulsar/broker/authentication/oidc/AuthenticationProviderOpenIDIntegrationTest.java
+++ b/pulsar-broker-auth-oidc/src/test/java/org/apache/pulsar/broker/authentication/oidc/AuthenticationProviderOpenIDIntegrationTest.java
@@ -23,6 +23,11 @@ import static com.github.tomakehurst.wiremock.client.WireMock.get;
import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
import static com.github.tomakehurst.wiremock.client.WireMock.urlMatching;
import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNull;
@@ -31,6 +36,7 @@ import static org.testng.Assert.fail;
import com.github.tomakehurst.wiremock.WireMockServer;
import com.github.tomakehurst.wiremock.stubbing.Scenario;
import com.google.common.io.Resources;
+import io.jsonwebtoken.Jwts;
import io.jsonwebtoken.SignatureAlgorithm;
import io.jsonwebtoken.impl.DefaultJwtBuilder;
import io.jsonwebtoken.io.Decoders;
@@ -44,23 +50,35 @@ import java.security.interfaces.RSAPublicKey;
import java.util.Base64;
import java.util.Date;
import java.util.HashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
+import javax.crypto.SecretKey;
import javax.naming.AuthenticationException;
import lombok.Cleanup;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.authentication.AuthenticationDataCommand;
import org.apache.pulsar.broker.authentication.AuthenticationProvider;
+import org.apache.pulsar.broker.authentication.AuthenticationProviderList;
import org.apache.pulsar.broker.authentication.AuthenticationProviderToken;
import org.apache.pulsar.broker.authentication.AuthenticationService;
import org.apache.pulsar.broker.authentication.AuthenticationState;
+import org.apache.pulsar.broker.authentication.TokenAuthenticationProvider;
import org.apache.pulsar.broker.authentication.utils.AuthTokenUtils;
+import org.apache.pulsar.broker.authorization.AuthorizationProvider;
+import org.apache.pulsar.broker.authorization.AuthorizationService;
+import org.apache.pulsar.broker.authorization.MultiRolesTokenAuthorizationProvider;
+import org.apache.pulsar.broker.resources.PulsarResources;
import org.apache.pulsar.common.api.AuthData;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
/**
@@ -274,6 +292,112 @@ public class AuthenticationProviderOpenIDIntegrationTest {
server.resetScenarios();
}
+ @DataProvider
+ public Object[][] roleClaims() {
+ return new Object[][]{
+ {"writer", Set.of("writer")},
+ {List.of("reader", "writer", "reader"), Set.of("reader", "writer")},
+ {List.of(), Set.of()},
+ {null, Set.of()},
+ {123, Set.of()},
+ {List.of("writer", 123), Set.of()}
+ };
+ }
+
+ @Test(dataProvider = "roleClaims")
+ public void testAuthenticateRolesFromValidatedClaims(Object claim, Set<String> expectedRoles) throws Exception {
+ HashMap<String, Object> claims = new HashMap<>();
+ claims.put("permissions", claim);
+ String token = generateToken(validJwk, issuer, "user", "allowed-audience", 0L, 0L, 10000L, claims);
+ assertThat(provider.authenticateRolesAsync(new AuthenticationDataCommand(token), "permissions").get())
+ .isEqualTo(expectedRoles);
+ }
+
+ @DataProvider
+ public Object[][] multiRoleProviders() {
+ return new Object[][]{{false, false}, {true, false}, {true, true}};
+ }
+
+ @Test(dataProvider = "multiRoleProviders")
+ public void testMultiRoleAuthorizationWithOpenID(boolean includeTokenProvider, boolean tokenFirst)
+ throws Exception {
+ ServiceConfiguration conf = new ServiceConfiguration();
+ conf.setAuthenticationEnabled(true);
+ List<String> providerClasses = includeTokenProvider
+ ? (tokenFirst
+ ? List.of(AuthenticationProviderToken.class.getName(),
+ AuthenticationProviderOpenID.class.getName())
+ : List.of(AuthenticationProviderOpenID.class.getName(),
+ AuthenticationProviderToken.class.getName()))
+ : List.of(AuthenticationProviderOpenID.class.getName());
+ conf.setAuthenticationProviders(new LinkedHashSet<>(providerClasses));
+ conf.setAuthorizationProvider(MultiRolesTokenAuthorizationProvider.class.getName());
+ conf.setSuperUserRoles(Set.of("admin"));
+ conf.getProperties().setProperty(AuthenticationProviderOpenID.ISSUER_TRUST_CERTS_FILE_PATH, caCert);
+ conf.getProperties().setProperty(AuthenticationProviderOpenID.ALLOWED_AUDIENCES, "allowed-audience");
+ conf.getProperties().setProperty(AuthenticationProviderOpenID.ALLOWED_TOKEN_ISSUERS, issuer);
+ SecretKey tokenKey = Jwts.SIG.HS256.key().build();
+ if (includeTokenProvider) {
+ conf.getProperties().setProperty("tokenSecretKey", AuthTokenUtils.encodeKeyBase64(tokenKey));
+ }
+ @Cleanup
+ AuthenticationService authenticationService = spy(new AuthenticationService(conf));
+ PulsarResources resources = mock(PulsarResources.class);
+ AuthorizationService authorizationService = new AuthorizationService(conf, resources, authenticationService);
+ verify(authenticationService).getAuthenticationProvider("token");
+ if (includeTokenProvider) {
+ AuthenticationProvider sharedProvider = authenticationService.getAuthenticationProvider("token");
+ assertThat(sharedProvider).isInstanceOf(AuthenticationProviderList.class);
+ AuthenticationProviderList providerList = (AuthenticationProviderList) sharedProvider;
+ assertThat(providerList.getAuthMethodName()).isEqualTo(TokenAuthenticationProvider.AUTH_METHOD_NAME);
+ assertThat(providerList.getProviders()).extracting(p -> p.getClass().getName())
+ .containsExactlyElementsOf(providerClasses);
+ }
+
+ HashMap<String, Object> roles = new HashMap<>(Map.of("roles", List.of("user", "admin")));
+ String valid = generateToken(validJwk, issuer, "user", "allowed-audience", 0L, 0L, 10000L, roles);
+ assertThat(authorizationService.isSuperUser("user", new AuthenticationDataCommand(valid)).get()).isTrue();
+ for (String invalid : List.of(
+ generateToken(invalidJwk, issuer, "user", "allowed-audience", 0L, 0L, 10000L, roles),
+ generateToken(validJwk, issuer, "user", "another-audience", 0L, 0L, 10000L, roles),
+ generateToken(validJwk, issuerThatFails, "user", "allowed-audience", 0L, 0L, 10000L, roles),
+ generateToken(validJwk, issuer, "user", "allowed-audience", -120000L, -120000L, -60000L, roles))) {
+ assertThat(authorizationService.isSuperUser("user", new AuthenticationDataCommand(invalid)).get())
+ .isFalse();
+ }
+
+ @Cleanup
+ MultiRolesTokenAuthorizationProvider multiRoles = new MultiRolesTokenAuthorizationProvider();
+ multiRoles.initialize(new AuthorizationProvider.InitialContext(conf, resources, authenticationService));
+ String normal = generateToken(validJwk, issuer, "user", "allowed-audience", 0L, 0L, 10000L,
+ new HashMap<>(Map.of("roles", List.of("user", "writer"))));
+ assertThat(multiRoles.authorize("user", new AuthenticationDataCommand(normal),
+ role -> CompletableFuture.completedFuture(role.equals("writer"))).get()).isTrue();
+ assertThat(multiRoles.authorize("user", new AuthenticationDataCommand(normal),
+ role -> CompletableFuture.completedFuture(role.equals("other-user"))).get()).isFalse();
+ if (includeTokenProvider) {
+ String staticToken = Jwts.builder().subject("user").claim("roles", List.of("user", "writer"))
+ .signWith(tokenKey).compact();
+ AuthenticationProviderList providerList =
+ (AuthenticationProviderList) authenticationService.getAuthenticationProvider("token");
+ for (String token : List.of(normal, staticToken)) {
+ AuthenticationDataCommand authData = new AuthenticationDataCommand(token);
+ assertThat(providerList.authenticateAsync(authData).get()).isEqualTo("user");
+ assertThat(providerList.authenticateRolesAsync(authData, "roles").get())
+ .containsExactlyInAnyOrder("user", "writer");
+ assertThat(multiRoles.authorize("user", authData,
+ role -> CompletableFuture.completedFuture(role.equals("writer"))).get()).isTrue();
+ assertThat(multiRoles.authorize("user", authData,
+ role -> CompletableFuture.completedFuture(role.equals("other-user"))).get()).isFalse();
+ }
+ }
+ multiRoles.close();
+ verify(authenticationService, never()).close();
+ // The same initialized provider remains usable after authorization closes.
+ assertThat(authenticationService.getAuthenticationProvider("token")
+ .authenticateAsync(new AuthenticationDataCommand(normal)).get()).isEqualTo("user");
+ }
+
@Test
public void testTokenWithValidJWK() throws Exception {
String role = "superuser";
diff --git a/pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/ProxySaslAuthenticationTest.java b/pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/ProxySaslAuthenticationTest.java
index 4197cc723114..5dd55727167f 100644
--- a/pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/ProxySaslAuthenticationTest.java
+++ b/pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/ProxySaslAuthenticationTest.java
@@ -179,6 +179,8 @@ public class ProxySaslAuthenticationTest extends ProducerConsumerBase {
isTcpLookup = true;
conf.setAdvertisedAddress(localHostname);
conf.setAuthenticationEnabled(true);
+ // The client SASL handshake terminates at the proxy, which forwards the original principal.
+ conf.setAuthenticateOriginalAuthData(false);
conf.setSaslJaasClientAllowedIds(".*" + localHostname + ".*");
conf.setSaslJaasServerSectionName("PulsarBroker");
brokerSecretKeyFile = File.createTempFile("saslRoleTokenSignerSecret", ".key");
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index c6e098d10356..8a729ab5138c 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -2219,8 +2219,12 @@ public class ServiceConfiguration implements PulsarConfiguration {
@FieldContext(
category = CATEGORY_AUTHORIZATION,
doc = "If this flag is set then the broker authenticates the original Auth data"
- + " else it just accepts the originalPrincipal and authorizes it (if required)")
- private boolean authenticateOriginalAuthData = false;
+ + " else it just accepts the originalPrincipal and authorizes it (if required)."
+ + " Set false for TLS client-certificate authentication through a proxy, since the broker"
+ + " receives the proxy certificate rather than the client certificate."
+ + " Also set false for SASL authentication through a proxy, since the client-proxy handshake"
+ + " cannot be replayed as a separate client-broker handshake.")
+ private boolean authenticateOriginalAuthData = true;
@FieldContext(
category = CATEGORY_AUTHORIZATION,
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 1bd527a26099..8c4a505cb6da 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
@@ -24,7 +24,9 @@ import java.io.IOException;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
import javax.naming.AuthenticationException;
import javax.net.ssl.SSLSession;
import lombok.CustomLog;
@@ -37,7 +39,7 @@ import org.apache.pulsar.common.api.AuthData;
*/
@CustomLog
@SuppressWarnings("deprecation")
-public class AuthenticationProviderList implements AuthenticationProvider {
+public class AuthenticationProviderList implements TokenAuthenticationProvider {
private AuthenticationMetrics authenticationMetrics;
@@ -249,15 +251,30 @@ public class AuthenticationProviderList implements AuthenticationProvider {
@Override
public CompletableFuture<String> authenticateAsync(AuthenticationDataSource authData) {
- CompletableFuture<String> roleFuture = new CompletableFuture<>();
- authenticateRemainingAuthProviders(roleFuture, authData, null, providers.isEmpty() ? -1 : 0);
- return roleFuture;
+ return authenticateAsync(provider -> provider.authenticateAsync(authData));
}
- private void authenticateRemainingAuthProviders(CompletableFuture<String> roleFuture,
- AuthenticationDataSource authData,
- Throwable previousException,
- int index) {
+ @Override
+ public CompletableFuture<Set<String>> authenticateRolesAsync(AuthenticationDataSource authData, String roleClaim) {
+ return authenticateAsync(provider -> {
+ if (provider instanceof TokenAuthenticationProvider tokenProvider) {
+ return tokenProvider.authenticateRolesAsync(authData, roleClaim);
+ }
+ return CompletableFuture.failedFuture(
+ new AuthenticationException("Authentication provider does not support token roles"));
+ });
+ }
+
+ private <T> CompletableFuture<T> authenticateAsync(
+ Function<AuthenticationProvider, CompletableFuture<T>> authenticate) {
+ CompletableFuture<T> result = new CompletableFuture<>();
+ authenticateRemainingAuthProviders(result, authenticate, null, providers.isEmpty() ? -1 : 0);
+ return result;
+ }
+
+ private <T> void authenticateRemainingAuthProviders(CompletableFuture<T> roleFuture,
+ Function<AuthenticationProvider, CompletableFuture<T>> authenticate,
+ Throwable previousException, int index) {
if (index < 0 || index >= providers.size()) {
if (previousException == null) {
previousException = new AuthenticationException("Authentication required");
@@ -268,17 +285,22 @@ public class AuthenticationProviderList implements AuthenticationProvider {
return;
}
AuthenticationProvider provider = providers.get(index);
- provider.authenticateAsync(authData)
- .whenComplete((role, ex) -> {
- if (ex == null) {
- roleFuture.complete(role);
- } else {
- log.debug().attr("authProvider", provider.getClass()).exception(ex)
- .log("Authentication failed for auth provider");
- authenticateRemainingAuthProviders(roleFuture, authData, ex, index + 1);
- }
- });
+ CompletableFuture<T> authentication;
+ try {
+ authentication = authenticate.apply(provider);
+ } catch (Exception e) {
+ authentication = CompletableFuture.failedFuture(e);
}
+ authentication.whenComplete((role, ex) -> {
+ if (ex == null) {
+ roleFuture.complete(role);
+ } else {
+ log.debug().attr("authProvider", provider.getClass()).exception(ex)
+ .log("Authentication failed for auth provider");
+ authenticateRemainingAuthProviders(roleFuture, authenticate, ex, index + 1);
+ }
+ });
+ }
@Override
public String authenticate(AuthenticationDataSource authData) throws AuthenticationException {
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 b7c733d18551..232acf90163c 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,8 @@ import java.security.Key;
import java.util.Collection;
import java.util.Date;
import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import javax.naming.AuthenticationException;
import javax.net.ssl.SSLSession;
import org.apache.commons.lang3.StringUtils;
@@ -47,7 +49,7 @@ import org.apache.pulsar.broker.authentication.utils.AuthTokenUtils;
import org.apache.pulsar.common.api.AuthData;
@SuppressWarnings({"deprecation", "unchecked"})
-public class AuthenticationProviderToken implements AuthenticationProvider {
+public class AuthenticationProviderToken implements TokenAuthenticationProvider {
static final String HTTP_HEADER_NAME = "Authorization";
static final String HTTP_HEADER_VALUE_PREFIX = "Bearer ";
@@ -76,7 +78,7 @@ public class AuthenticationProviderToken implements AuthenticationProvider {
// token validation.
static final String CONF_TOKEN_ALLOWED_CLOCK_SKEW_SECONDS = "tokenAllowedClockSkewSeconds";
- static final String TOKEN = "token";
+ static final String TOKEN = AUTH_METHOD_NAME;
private Key validationKey;
private String roleClaim;
@@ -152,7 +154,7 @@ public class AuthenticationProviderToken implements AuthenticationProvider {
@Override
public String getAuthMethodName() {
- return TOKEN;
+ return AUTH_METHOD_NAME;
}
@Override
@@ -162,6 +164,24 @@ public class AuthenticationProviderToken implements AuthenticationProvider {
@Override
public String authenticate(AuthenticationDataSource authData) throws AuthenticationException {
+ String role = getPrincipal(authenticateClaims(authData));
+ authenticationMetricsToken.recordSuccess();
+ return role;
+ }
+
+ @Override
+ public CompletableFuture<Set<String>> authenticateRolesAsync(AuthenticationDataSource authData, String roleClaim) {
+ try {
+ Set<String> roles = AuthTokenUtils.rolesFromClaim(
+ authenticateClaims(authData).getPayload().get(roleClaim));
+ authenticationMetricsToken.recordSuccess();
+ return CompletableFuture.completedFuture(roles);
+ } catch (Exception e) {
+ return CompletableFuture.failedFuture(e);
+ }
+ }
+
+ private Jws<Claims> authenticateClaims(AuthenticationDataSource authData) throws AuthenticationException {
String token;
try {
// Get Token
@@ -170,10 +190,7 @@ public class AuthenticationProviderToken implements AuthenticationProvider {
incrementFailureMetric(ErrorCode.INVALID_AUTH_DATA);
throw exception;
}
- // Parse Token by validating
- String role = getPrincipal(authenticateToken(token));
- authenticationMetricsToken.recordSuccess();
- return role;
+ return authenticateToken(token);
}
@Override
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/TokenAuthenticationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/TokenAuthenticationProvider.java
new file mode 100644
index 000000000000..bddcd136b43e
--- /dev/null
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/TokenAuthenticationProvider.java
@@ -0,0 +1,37 @@
+/*
+ * 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.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * A JWT authentication provider that can return all roles from a validated token.
+ */
+public interface TokenAuthenticationProvider extends AuthenticationProvider {
+
+ String AUTH_METHOD_NAME = "token";
+
+ /**
+ * Validate the token and extract the requested role claim from the same parsed token.
+ * Implementations use their configured signature, issuer, audience and time checks.
+ * Failures must be returned as failed futures without blocking the calling thread.
+ */
+ CompletableFuture<Set<String>> authenticateRolesAsync(AuthenticationDataSource authData, String roleClaim);
+}
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/utils/AuthTokenUtils.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/utils/AuthTokenUtils.java
index c0ab48605654..318a184fa821 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/utils/AuthTokenUtils.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/utils/AuthTokenUtils.java
@@ -34,9 +34,12 @@ import java.security.PrivateKey;
import java.security.PublicKey;
import java.security.spec.PKCS8EncodedKeySpec;
import java.security.spec.X509EncodedKeySpec;
+import java.util.Collection;
import java.util.Date;
+import java.util.HashSet;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import javax.crypto.SecretKey;
import lombok.experimental.UtilityClass;
import org.apache.commons.codec.binary.Base64;
@@ -47,6 +50,27 @@ import org.apache.pulsar.client.api.url.URL;
@UtilityClass
public class AuthTokenUtils {
+ /**
+ * Interpret a role claim as a string or a collection of strings.
+ * Missing claims and other value types do not supply any roles.
+ */
+ public static Set<String> rolesFromClaim(Object claim) {
+ if (claim instanceof String role) {
+ return Set.of(role);
+ }
+ if (claim instanceof Collection<?> values) {
+ Set<String> roles = new HashSet<>();
+ for (Object value : values) {
+ if (!(value instanceof String role)) {
+ return Set.of();
+ }
+ roles.add(role);
+ }
+ return Set.copyOf(roles);
+ }
+ return Set.of();
+ }
+
public static SecretKey createSecretKey(SignatureAlgorithm signatureAlgorithm) {
return Keys.secretKeyFor(signatureAlgorithm);
}
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java
index cb61292f8e38..f3cce0a2a5a2 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java
@@ -24,8 +24,10 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import lombok.Builder;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
+import org.apache.pulsar.broker.authentication.AuthenticationService;
import org.apache.pulsar.broker.resources.PulsarResources;
import org.apache.pulsar.client.admin.GrantTopicPermissionOptions;
import org.apache.pulsar.client.admin.RevokeTopicPermissionOptions;
@@ -75,6 +77,22 @@ public interface AuthorizationProvider extends Closeable {
&& tenantInfo.getAdminRoles().contains(role));
}
+ /**
+ * Initialization dependencies for an authorization provider.
+ * The authentication service is already initialized and owned by the enclosing service.
+ */
+ @Builder
+ record InitialContext(ServiceConfiguration config, PulsarResources pulsarResources,
+ AuthenticationService authenticationService) {
+ }
+
+ /**
+ * Initialize the authorization provider with its shared dependencies.
+ */
+ default void initialize(InitialContext context) throws IOException {
+ initialize(context.config(), context.pulsarResources());
+ }
+
/**
* Perform initialization for the authorization provider.
*
@@ -82,9 +100,11 @@ public interface AuthorizationProvider extends Closeable {
* broker config object
* @param pulsarResources
* Resources component for access to metadata
+ * @deprecated use {@link #initialize(InitialContext)} instead
* @throws IOException
* if the initialization fails
*/
+ @Deprecated
default void initialize(ServiceConfiguration conf, PulsarResources pulsarResources) throws IOException {
}
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java
index c134ffcd553b..c8a7a4d244ae 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java
@@ -33,6 +33,7 @@ import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
import org.apache.pulsar.broker.authentication.AuthenticationParameters;
+import org.apache.pulsar.broker.authentication.AuthenticationService;
import org.apache.pulsar.broker.resources.PulsarResources;
import org.apache.pulsar.client.admin.GrantTopicPermissionOptions;
import org.apache.pulsar.client.admin.RevokeTopicPermissionOptions;
@@ -64,13 +65,22 @@ public class AuthorizationService {
public AuthorizationService(ServiceConfiguration conf, PulsarResources pulsarResources)
throws PulsarServerException {
+ this(conf, pulsarResources, null);
+ }
+
+ public AuthorizationService(ServiceConfiguration conf, PulsarResources pulsarResources,
+ AuthenticationService authenticationService) throws PulsarServerException {
this.conf = conf;
try {
final String providerClassname = conf.getAuthorizationProvider();
if (StringUtils.isNotBlank(providerClassname)) {
provider = (AuthorizationProvider) Class.forName(providerClassname)
.getDeclaredConstructor().newInstance();
- provider.initialize(conf, pulsarResources);
+ provider.initialize(AuthorizationProvider.InitialContext.builder()
+ .config(conf)
+ .pulsarResources(pulsarResources)
+ .authenticationService(authenticationService)
+ .build());
this.resources = pulsarResources;
log.info().attr("providerClassname", providerClassname).log("Loaded authorization provider");
} else {
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java
index 0e0cfe309a6e..50ca3a25464d 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java
@@ -18,19 +18,10 @@
*/
package org.apache.pulsar.broker.authorization;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.node.ObjectNode;
-import io.jsonwebtoken.JwtParser;
-import io.jsonwebtoken.Jwts;
import jakarta.ws.rs.core.Response;
import java.io.IOException;
-import java.io.UncheckedIOException;
import java.util.ArrayList;
-import java.util.Base64;
-import java.util.Collection;
import java.util.Collections;
-import java.util.HashSet;
import java.util.List;
import java.util.Optional;
import java.util.Set;
@@ -41,6 +32,8 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
import org.apache.pulsar.broker.authentication.AuthenticationDataSubscription;
+import org.apache.pulsar.broker.authentication.AuthenticationProvider;
+import org.apache.pulsar.broker.authentication.TokenAuthenticationProvider;
import org.apache.pulsar.broker.resources.PulsarResources;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
@@ -55,12 +48,16 @@ import org.apache.pulsar.common.util.RestException;
import org.apache.pulsar.metadata.api.MetadataStoreException;
+/**
+ * Authorizes roles in JWTs authenticated by a {@link TokenAuthenticationProvider}.
+ *
+ * <p>Authorization obtains roles asynchronously from the existing authentication provider's validated
+ * token, using its configured keys, issuer, audience and clock-skew handling. Proxies must forward
+ * the original credentials and brokers must authenticate them.
+ */
@CustomLog
public class MultiRolesTokenAuthorizationProvider extends PulsarAuthorizationProvider {
- static final String HTTP_HEADER_NAME = "Authorization";
- static final String HTTP_HEADER_VALUE_PREFIX = "Bearer ";
-
// When symmetric key is configured
static final String CONF_TOKEN_SETTING_PREFIX = "tokenSettingPrefix";
@@ -69,18 +66,18 @@ public class MultiRolesTokenAuthorizationProvider extends PulsarAuthorizationPro
static final String DEFAULT_ROLE_CLAIM = "roles";
- private static final ObjectMapper mapper = new ObjectMapper();
-
- private final JwtParser parser;
- private String roleClaim;
+ private String roleClaim = DEFAULT_ROLE_CLAIM;
+ private TokenAuthenticationProvider authenticationProvider;
- public MultiRolesTokenAuthorizationProvider() {
- this.roleClaim = DEFAULT_ROLE_CLAIM;
- this.parser = Jwts.parser().unsecured().build();
+ @Deprecated
+ @Override
+ public void initialize(ServiceConfiguration conf, PulsarResources pulsarResources) throws IOException {
+ initialize(new InitialContext(conf, pulsarResources, null));
}
@Override
- public void initialize(ServiceConfiguration conf, PulsarResources pulsarResources) throws IOException {
+ public void initialize(InitialContext context) throws IOException {
+ ServiceConfiguration conf = context.config();
String prefix = (String) conf.getProperty(CONF_TOKEN_SETTING_PREFIX);
if (null == prefix) {
prefix = "";
@@ -91,7 +88,21 @@ public class MultiRolesTokenAuthorizationProvider extends PulsarAuthorizationPro
this.roleClaim = (String) tokenAuthClaim;
}
- super.initialize(conf, pulsarResources);
+ if (!conf.isAuthenticationEnabled()) {
+ throw new IOException("MultiRolesTokenAuthorizationProvider requires authenticationEnabled=true");
+ }
+ // Token and OpenID share the "token" method; the service returns their AuthenticationProviderList
+ // when both are configured.
+ AuthenticationProvider sharedProvider = context.authenticationService() == null ? null
+ : context.authenticationService()
+ .getAuthenticationProvider(TokenAuthenticationProvider.AUTH_METHOD_NAME);
+ if (!(sharedProvider instanceof TokenAuthenticationProvider tokenProvider)) {
+ throw new IOException("MultiRolesTokenAuthorizationProvider requires an initialized token authentication "
+ + "provider in AuthorizationProvider.InitialContext");
+ }
+ authenticationProvider = tokenProvider;
+
+ super.initialize(context);
}
@Override
@@ -105,22 +116,21 @@ public class MultiRolesTokenAuthorizationProvider extends PulsarAuthorizationPro
if (role != null && superUserRoles.contains(role)) {
return CompletableFuture.completedFuture(true);
}
- Set<String> roles = getRoles(role, authenticationData);
- if (roles.isEmpty()) {
- return CompletableFuture.completedFuture(false);
- }
- return CompletableFuture.completedFuture(roles.stream().anyMatch(superUserRoles::contains));
+ return getRolesAsync(role, authenticationData)
+ .thenApply(roles -> roles.stream().anyMatch(superUserRoles::contains));
}
@Override
public CompletableFuture<Boolean> validateTenantAdminAccess(String tenantName, String role,
AuthenticationDataSource authData) {
- return isSuperUser(role, authData, conf)
- .thenCompose(isSuperUser -> {
- if (isSuperUser) {
+ if (role != null && conf.getSuperUserRoles().contains(role)) {
+ return CompletableFuture.completedFuture(true);
+ }
+ return getRolesAsync(role, authData)
+ .thenCompose(roles -> {
+ if (roles.stream().anyMatch(conf.getSuperUserRoles()::contains)) {
return CompletableFuture.completedFuture(true);
}
- Set<String> roles = getRoles(role, authData);
if (roles.isEmpty()) {
return CompletableFuture.completedFuture(false);
}
@@ -159,72 +169,33 @@ public class MultiRolesTokenAuthorizationProvider extends PulsarAuthorizationPro
});
}
- @SuppressWarnings({"deprecation", "unchecked"})
- private Set<String> getRoles(String role, AuthenticationDataSource authData) {
- if (authData == null || (authData instanceof AuthenticationDataSubscription
- && ((AuthenticationDataSubscription) authData).getAuthData() == null)) {
- return Collections.singleton(role);
+ private CompletableFuture<Set<String>> getRolesAsync(String role, AuthenticationDataSource authData) {
+ if (authData == null || (authData instanceof AuthenticationDataSubscription subscription
+ && subscription.getAuthData() == null)) {
+ return CompletableFuture.completedFuture(
+ role == null ? Collections.emptySet() : Collections.singleton(role));
}
-
- String token = null;
-
- if (authData.hasDataFromCommand()) {
- // Authenticate Pulsar binary connection
- token = authData.getCommandData();
- if (StringUtils.isBlank(token)) {
- return Collections.emptySet();
- }
- } else if (authData.hasDataFromHttp()) {
- // The format here should be compliant to RFC-6750
- // (https://tools.ietf.org/html/rfc6750#section-2.1). Eg: Authorization: Bearer xxxxxxxxxxxxx
- String httpHeaderValue = authData.getHttpHeader(HTTP_HEADER_NAME);
- if (httpHeaderValue == null || !httpHeaderValue.startsWith(HTTP_HEADER_VALUE_PREFIX)) {
- return Collections.emptySet();
- }
-
- // Remove prefix
- token = httpHeaderValue.substring(HTTP_HEADER_VALUE_PREFIX.length());
- }
-
- if (token == null) {
- return Collections.emptySet();
- }
-
- String[] splitToken = token.split("\\.");
- if (splitToken.length < 2) {
- log.warn("Unable to extract additional roles from JWT token");
- return Collections.emptySet();
- }
- String unsignedToken = replaceAlgWithNoneInHeader(splitToken[0]) + "." + splitToken[1] + ".";
- Object jwtRole = parser.parseUnsecuredClaims(unsignedToken).getBody().get(roleClaim);
- if (jwtRole instanceof String) {
- return new HashSet<String>(Collections.singletonList((String) jwtRole));
- } else if (jwtRole instanceof Collection) {
- return new HashSet<>((Collection<String>) jwtRole);
- } else {
- return Collections.emptySet();
- }
- }
-
- // replace alg with none in header so that it can be parsed in unsecured mode
- private static String replaceAlgWithNoneInHeader(String header) {
try {
- JsonNode jsonNode = mapper.readTree(Base64.getDecoder().decode(header));
- ((ObjectNode) jsonNode).put("alg", "none");
- return Base64.getEncoder().withoutPadding().encodeToString(mapper.writeValueAsBytes(jsonNode));
- } catch (IOException e) {
- throw new UncheckedIOException(e);
+ return authenticationProvider.authenticateRolesAsync(authData, roleClaim)
+ .exceptionally(error -> {
+ log.debug().log("Unable to extract additional roles from JWT token");
+ return Collections.emptySet();
+ });
+ } catch (RuntimeException e) {
+ return CompletableFuture.completedFuture(Collections.emptySet());
}
}
public CompletableFuture<Boolean> authorize(String role, AuthenticationDataSource authenticationData,
Function<String, CompletableFuture<Boolean>> authorizeFunc) {
- return isSuperUser(role, authenticationData, conf)
- .thenCompose(superUser -> {
- if (superUser) {
+ if (role != null && conf.getSuperUserRoles().contains(role)) {
+ return CompletableFuture.completedFuture(true);
+ }
+ return getRolesAsync(role, authenticationData)
+ .thenCompose(roles -> {
+ if (roles.stream().anyMatch(conf.getSuperUserRoles()::contains)) {
return CompletableFuture.completedFuture(true);
}
- Set<String> roles = getRoles(role, authenticationData);
if (roles.isEmpty()) {
return CompletableFuture.completedFuture(false);
}
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
index 9ae57db36653..b20da66594df 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
@@ -73,12 +73,16 @@ public class PulsarAuthorizationProvider implements AuthorizationProvider {
initialize(conf, resources);
}
+ @Deprecated
@Override
public void initialize(ServiceConfiguration conf, PulsarResources pulsarResources) throws IOException {
- requireNonNull(conf, "ServiceConfiguration can't be null");
- requireNonNull(pulsarResources, "PulsarResources can't be null");
- this.conf = conf;
- this.pulsarResources = pulsarResources;
+ initialize(InitialContext.builder().config(conf).pulsarResources(pulsarResources).build());
+ }
+
+ @Override
+ public void initialize(InitialContext context) throws IOException {
+ this.conf = requireNonNull(context.config(), "ServiceConfiguration can't be null");
+ this.pulsarResources = requireNonNull(context.pulsarResources(), "PulsarResources can't be null");
}
/**
diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authentication/AuthenticationProviderTokenTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authentication/AuthenticationProviderTokenTest.java
index 7377c8fc3823..f8053962a8d8 100644
--- a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authentication/AuthenticationProviderTokenTest.java
+++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authentication/AuthenticationProviderTokenTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.authentication;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
@@ -51,6 +52,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Optional;
import java.util.Properties;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import javax.crypto.SecretKey;
import javax.naming.AuthenticationException;
@@ -61,12 +63,38 @@ import org.apache.pulsar.broker.authentication.metrics.AuthenticationMetricsToke
import org.apache.pulsar.broker.authentication.utils.AuthTokenUtils;
import org.apache.pulsar.common.api.AuthData;
import org.mockito.Mockito;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
public class AuthenticationProviderTokenTest {
private static final String SUBJECT = "my-test-subject";
+ @DataProvider
+ public Object[][] roleClaims() {
+ return new Object[][]{
+ {"writer", Set.of("writer")},
+ {List.of("reader", "writer", "reader"), Set.of("reader", "writer")},
+ {List.of(), Set.of()},
+ {null, Set.of()},
+ {123, Set.of()},
+ {List.of("writer", 123), Set.of()}
+ };
+ }
+
+ @Test(dataProvider = "roleClaims")
+ public void testAuthenticateRolesFromValidatedClaims(Object claim, Set<String> expectedRoles) throws Exception {
+ SecretKey key = Jwts.SIG.HS256.key().build();
+ ServiceConfiguration config = new ServiceConfiguration();
+ config.getProperties().setProperty("tokenSecretKey", AuthTokenUtils.encodeKeyBase64(key));
+ @Cleanup
+ TokenAuthenticationProvider provider = new AuthenticationProviderToken();
+ provider.initialize(AuthenticationProvider.Context.builder().config(config).build());
+ String token = Jwts.builder().subject(SUBJECT).claim("permissions", claim).signWith(key).compact();
+ assertThat(provider.authenticateRolesAsync(new AuthenticationDataCommand(token), "permissions").get())
+ .isEqualTo(expectedRoles);
+ }
+
@Test
public void testInvalidInitialize() throws Exception {
AuthenticationProviderToken provider = new AuthenticationProviderToken();
diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/AuthorizationServiceTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/AuthorizationServiceTest.java
index 6f9dffa11b94..ebdbaa9e53cc 100644
--- a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/AuthorizationServiceTest.java
+++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/AuthorizationServiceTest.java
@@ -18,11 +18,17 @@
*/
package org.apache.pulsar.broker.authorization;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.testng.AssertJUnit.assertFalse;
import static org.testng.AssertJUnit.assertTrue;
import java.util.HashSet;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.authentication.AuthenticationService;
+import org.apache.pulsar.broker.resources.PulsarResources;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.NamespaceOperation;
@@ -51,6 +57,37 @@ public class AuthorizationServiceTest {
authorizationService = new AuthorizationService(conf, null);
}
+ @Test
+ @SuppressWarnings("deprecation")
+ public void testContextDelegatesToLegacyInitializer() throws Exception {
+ ServiceConfiguration config = new ServiceConfiguration();
+ PulsarResources resources = mock(PulsarResources.class);
+ AuthenticationService authenticationService = mock(AuthenticationService.class);
+ AuthorizationProvider provider = mock(AuthorizationProvider.class, CALLS_REAL_METHODS);
+ provider.initialize(AuthorizationProvider.InitialContext.builder()
+ .config(config).pulsarResources(resources).authenticationService(authenticationService).build());
+ verify(provider).initialize(config, resources);
+ }
+
+ @Test
+ @SuppressWarnings("deprecation")
+ public void testPulsarProviderInitializers() throws Exception {
+ ServiceConfiguration config = new ServiceConfiguration();
+ PulsarResources resources = mock(PulsarResources.class);
+ try (PulsarAuthorizationProvider provider = new PulsarAuthorizationProvider()) {
+ provider.initialize(AuthorizationProvider.InitialContext.builder()
+ .config(config).pulsarResources(resources).build());
+ assertThat(provider.conf).isSameAs(config);
+ assertThat(provider.pulsarResources).isSameAs(resources);
+
+ ServiceConfiguration legacyConfig = new ServiceConfiguration();
+ PulsarResources legacyResources = mock(PulsarResources.class);
+ provider.initialize(legacyConfig, legacyResources);
+ assertThat(provider.conf).isSameAs(legacyConfig);
+ assertThat(provider.pulsarResources).isSameAs(legacyResources);
+ }
+ }
+
/**
* See {@link MockAuthorizationProvider} for the implementation of the mock authorization provider.
*/
diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProviderTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProviderTest.java
index 103e7ded3017..2ce701f600ac 100644
--- a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProviderTest.java
+++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProviderTest.java
@@ -18,7 +18,13 @@
*/
package org.apache.pulsar.broker.authorization;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
@@ -27,6 +33,8 @@ import static org.testng.Assert.expectThrows;
import io.jsonwebtoken.Jwts;
import io.jsonwebtoken.SignatureAlgorithm;
import jakarta.ws.rs.core.Response;
+import java.io.IOException;
+import java.util.Date;
import java.util.List;
import java.util.Optional;
import java.util.Properties;
@@ -39,16 +47,174 @@ import lombok.Cleanup;
import org.apache.logging.log4j.Level;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.authentication.AuthenticationDataCommand;
import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
import org.apache.pulsar.broker.authentication.AuthenticationDataSubscription;
+import org.apache.pulsar.broker.authentication.AuthenticationProvider;
+import org.apache.pulsar.broker.authentication.AuthenticationProviderTls;
+import org.apache.pulsar.broker.authentication.AuthenticationProviderToken;
+import org.apache.pulsar.broker.authentication.AuthenticationService;
+import org.apache.pulsar.broker.authentication.TokenAuthenticationProvider;
import org.apache.pulsar.broker.authentication.utils.AuthTokenUtils;
import org.apache.pulsar.broker.resources.PulsarResources;
import org.apache.pulsar.broker.resources.TenantResources;
import org.apache.pulsar.common.util.RestException;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
public class MultiRolesTokenAuthorizationProviderTest {
+ private static ServiceConfiguration tokenConfiguration(SecretKey key) {
+ ServiceConfiguration conf = new ServiceConfiguration();
+ conf.setAuthenticationEnabled(true);
+ conf.setAuthenticationProviders(Set.of(AuthenticationProviderToken.class.getName()));
+ conf.getProperties().setProperty("tokenSecretKey", AuthTokenUtils.encodeKeyBase64(key));
+ return conf;
+ }
+
+ private static void initializeProvider(MultiRolesTokenAuthorizationProvider provider, ServiceConfiguration conf,
+ PulsarResources resources) throws IOException {
+ AuthenticationProviderToken authenticationProvider = new AuthenticationProviderToken();
+ authenticationProvider.initialize(AuthenticationProvider.Context.builder().config(conf).build());
+ provider.initialize(new AuthorizationProvider.InitialContext(conf, resources,
+ authenticationService(authenticationProvider)));
+ }
+
+ private static AuthenticationService authenticationService(AuthenticationProvider provider) {
+ AuthenticationService service = mock(AuthenticationService.class);
+ when(service.getAuthenticationProvider("token")).thenReturn(provider);
+ return service;
+ }
+
+ @DataProvider
+ public Object[][] validationSettings() {
+ return new Object[][]{
+ {""}, {"custom_"}
+ };
+ }
+
+ @Test(dataProvider = "validationSettings")
+ public void testTokenValidation(String prefix) throws Exception {
+ SecretKey key = Jwts.SIG.HS256.key().build();
+ ServiceConfiguration conf = tokenConfiguration(key);
+ conf.setSuperUserRoles(Set.of("admin"));
+ conf.getProperties().setProperty("tokenSettingPrefix", prefix);
+ conf.getProperties().setProperty(prefix + "tokenSecretKey", AuthTokenUtils.encodeKeyBase64(key));
+ @Cleanup
+ MultiRolesTokenAuthorizationProvider provider = new MultiRolesTokenAuthorizationProvider();
+ initializeProvider(provider, conf, mock(PulsarResources.class));
+
+ String valid = Jwts.builder().claim("roles", List.of("user", "admin")).signWith(key).compact();
+ String otherKey = Jwts.builder().claim("roles", List.of("user", "admin"))
+ .signWith(Jwts.SIG.HS256.key().build()).compact();
+ String unsigned = Jwts.builder().claim("roles", List.of("user", "admin")).compact();
+ String expired = Jwts.builder().claim("roles", List.of("user", "admin"))
+ .expiration(new Date(0)).signWith(key).compact();
+
+ assertThat(provider.isSuperUser("user", new AuthenticationDataCommand(valid), conf).get()).isTrue();
+ for (String token : List.of(otherKey, unsigned)) {
+ AuthenticationDataSource data = new AuthenticationDataCommand(token);
+ assertThat(provider.isSuperUser("user", data, conf).get()).isFalse();
+ assertThat(provider.authorize("user", data, role -> CompletableFuture.completedFuture(true)).get())
+ .isFalse();
+ }
+ for (String token : List.of(expired, "invalid.token.value", "not-a-token")) {
+ assertThat(provider.isSuperUser("user", new AuthenticationDataCommand(token), conf).get()).isFalse();
+ assertThat(provider.authorize("user", new AuthenticationDataCommand(token),
+ role -> CompletableFuture.completedFuture(true)).get()).isFalse();
+ }
+ }
+
+ @Test
+ public void testRequiresSharedAuthenticationProvider() throws Exception {
+ ServiceConfiguration conf = tokenConfiguration(Jwts.SIG.HS256.key().build());
+ @Cleanup
+ MultiRolesTokenAuthorizationProvider provider = new MultiRolesTokenAuthorizationProvider();
+ assertThatThrownBy(() -> provider.initialize(new AuthorizationProvider.InitialContext(
+ conf, mock(PulsarResources.class), null)))
+ .isInstanceOf(IOException.class).hasMessageContaining("initialized token authentication provider");
+ assertThatThrownBy(() -> provider.initialize(conf, mock(PulsarResources.class)))
+ .isInstanceOf(IOException.class);
+ }
+
+ @DataProvider
+ public Object[][] unsupportedProviders() {
+ return new Object[][]{
+ {null}, {mock(AuthenticationProvider.class)}, {mock(AuthenticationProviderTls.class)}
+ };
+ }
+
+ public static class CustomTokenProvider extends AuthenticationProviderToken {
+ }
+
+ @Test
+ public void testCustomTokenProvider() throws Exception {
+ SecretKey key = Jwts.SIG.HS256.key().build();
+ ServiceConfiguration conf = tokenConfiguration(key);
+ conf.setAuthenticationProviders(Set.of(CustomTokenProvider.class.getName()));
+ @Cleanup
+ AuthenticationService service = new AuthenticationService(conf);
+ @Cleanup
+ MultiRolesTokenAuthorizationProvider provider = new MultiRolesTokenAuthorizationProvider();
+ provider.initialize(new AuthorizationProvider.InitialContext(conf, mock(PulsarResources.class), service));
+ String token = Jwts.builder().claim("roles", List.of("user", "other")).signWith(key).compact();
+ assertThat(provider.authorize("user", new AuthenticationDataCommand(token),
+ role -> CompletableFuture.completedFuture("other".equals(role))).get()).isTrue();
+ }
+
+ @Test(dataProvider = "unsupportedProviders")
+ public void testRejectUnsupportedProviders(AuthenticationProvider authenticationProvider) throws IOException {
+ ServiceConfiguration conf = tokenConfiguration(Jwts.SIG.HS256.key().build());
+ @Cleanup
+ MultiRolesTokenAuthorizationProvider provider = new MultiRolesTokenAuthorizationProvider();
+ assertThatThrownBy(() -> provider.initialize(new AuthorizationProvider.InitialContext(
+ conf, null, authenticationService(authenticationProvider))))
+ .isInstanceOf(IOException.class).hasMessageContaining("initialized token authentication provider");
+ }
+
+ @Test
+ public void testRejectDisabledAuthentication() throws IOException {
+ ServiceConfiguration conf = tokenConfiguration(Jwts.SIG.HS256.key().build());
+ @Cleanup
+ MultiRolesTokenAuthorizationProvider provider = new MultiRolesTokenAuthorizationProvider();
+ conf.setAuthenticationEnabled(false);
+ assertThatThrownBy(() -> provider.initialize(new AuthorizationProvider.InitialContext(
+ conf, null, authenticationService(mock(TokenAuthenticationProvider.class)))))
+ .isInstanceOf(IOException.class).hasMessageContaining("authenticationEnabled=true");
+ }
+
+ @DataProvider
+ public Object[][] validationOutcomes() {
+ return new Object[][]{{true}, {false}};
+ }
+
+ @Test(dataProvider = "validationOutcomes")
+ public void testSharedProviderValidationIsAsync(boolean succeeds) throws Exception {
+ ServiceConfiguration conf = tokenConfiguration(Jwts.SIG.HS256.key().build());
+ conf.setSuperUserRoles(Set.of("admin"));
+ TokenAuthenticationProvider authenticationProvider = mock(TokenAuthenticationProvider.class);
+ CompletableFuture<Set<String>> validation = new CompletableFuture<>();
+ when(authenticationProvider.authenticateRolesAsync(any(), eq("roles"))).thenReturn(validation);
+ @Cleanup
+ MultiRolesTokenAuthorizationProvider provider = new MultiRolesTokenAuthorizationProvider();
+ provider.initialize(new AuthorizationProvider.InitialContext(conf, mock(PulsarResources.class),
+ authenticationService(authenticationProvider)));
+ String token = Jwts.builder().claim("roles", List.of("user", "admin")).compact();
+ CompletableFuture<Boolean> authorized = provider.authorize("user", new AuthenticationDataCommand(token),
+ role -> CompletableFuture.completedFuture(false));
+ assertThat(authorized).isNotDone();
+ if (succeeds) {
+ validation.complete(Set.of("user", "admin"));
+ } else {
+ validation.completeExceptionally(new IllegalArgumentException("Token validation failed"));
+ }
+ assertThat(authorized.get()).isEqualTo(succeeds);
+ verify(authenticationProvider).authenticateRolesAsync(any(), eq("roles"));
+ provider.close();
+ verify(authenticationProvider, never()).initialize(any(AuthenticationProvider.Context.class));
+ verify(authenticationProvider, never()).close();
+ }
+
@SuppressWarnings("deprecation")
@Test
public void testMultiRolesAuthz() throws Exception {
@@ -58,8 +224,8 @@ public class MultiRolesTokenAuthorizationProviderTest {
String token = Jwts.builder().claim("roles", new String[]{userA, userB}).signWith(secretKey).compact();
MultiRolesTokenAuthorizationProvider provider = new MultiRolesTokenAuthorizationProvider();
- ServiceConfiguration conf = new ServiceConfiguration();
- provider.initialize(conf, mock(PulsarResources.class));
+ ServiceConfiguration conf = tokenConfiguration(secretKey);
+ initializeProvider(provider, conf, mock(PulsarResources.class));
AuthenticationDataSource ads = new AuthenticationDataSource() {
@Override
@@ -100,8 +266,8 @@ public class MultiRolesTokenAuthorizationProviderTest {
String token = Jwts.builder().claim("roles", new String[]{}).signWith(secretKey).compact();
MultiRolesTokenAuthorizationProvider provider = new MultiRolesTokenAuthorizationProvider();
- ServiceConfiguration conf = new ServiceConfiguration();
- provider.initialize(conf, mock(PulsarResources.class));
+ ServiceConfiguration conf = tokenConfiguration(secretKey);
+ initializeProvider(provider, conf, mock(PulsarResources.class));
AuthenticationDataSource ads = new AuthenticationDataSource() {
@Override
@@ -130,8 +296,8 @@ public class MultiRolesTokenAuthorizationProviderTest {
String token = Jwts.builder().claim("roles", testRole).signWith(secretKey).compact();
MultiRolesTokenAuthorizationProvider provider = new MultiRolesTokenAuthorizationProvider();
- ServiceConfiguration conf = new ServiceConfiguration();
- provider.initialize(conf, mock(PulsarResources.class));
+ ServiceConfiguration conf = tokenConfiguration(secretKey);
+ initializeProvider(provider, conf, mock(PulsarResources.class));
AuthenticationDataSource ads = new AuthenticationDataSource() {
@Override
@@ -165,9 +331,9 @@ public class MultiRolesTokenAuthorizationProviderTest {
// broker will use "roles" as the claim by default.
final String token = Jwts.builder()
.claim("whatever", testRole).signWith(secretKey).compact();
- ServiceConfiguration conf = new ServiceConfiguration();
+ ServiceConfiguration conf = tokenConfiguration(secretKey);
final MultiRolesTokenAuthorizationProvider provider = new MultiRolesTokenAuthorizationProvider();
- provider.initialize(conf, mock(PulsarResources.class));
+ initializeProvider(provider, conf, mock(PulsarResources.class));
final AuthenticationDataSource ads = new AuthenticationDataSource() {
@Override
public boolean hasDataFromHttp() {
@@ -194,11 +360,12 @@ public class MultiRolesTokenAuthorizationProviderTest {
@Test
public void testMultiRolesAuthzWithAnonymousUser() throws Exception {
+ SecretKey secretKey = Jwts.SIG.HS256.key().build();
@Cleanup
MultiRolesTokenAuthorizationProvider provider = new MultiRolesTokenAuthorizationProvider();
- ServiceConfiguration conf = new ServiceConfiguration();
+ ServiceConfiguration conf = tokenConfiguration(secretKey);
- provider.initialize(conf, mock(PulsarResources.class));
+ initializeProvider(provider, conf, mock(PulsarResources.class));
Function<String, CompletableFuture<Boolean>> authorizeFunc = (String role) -> {
if (role.equals("test-role")) {
@@ -214,11 +381,12 @@ public class MultiRolesTokenAuthorizationProviderTest {
@Test
public void testMultiRolesNotFailNonJWT() throws Exception {
+ SecretKey secretKey = Jwts.SIG.HS256.key().build();
String token = "a-non-jwt-token";
MultiRolesTokenAuthorizationProvider provider = new MultiRolesTokenAuthorizationProvider();
- ServiceConfiguration conf = new ServiceConfiguration();
- provider.initialize(conf, mock(PulsarResources.class));
+ ServiceConfiguration conf = tokenConfiguration(secretKey);
+ initializeProvider(provider, conf, mock(PulsarResources.class));
AuthenticationDataSource ads = new AuthenticationDataSource() {
@Override
@@ -250,11 +418,12 @@ public class MultiRolesTokenAuthorizationProviderTest {
Properties properties = new Properties();
properties.setProperty("tokenSettingPrefix", "prefix_");
properties.setProperty("prefix_tokenAuthClaim", customRolesClaims);
- ServiceConfiguration conf = new ServiceConfiguration();
+ properties.setProperty("prefix_tokenSecretKey", AuthTokenUtils.encodeKeyBase64(secretKey));
+ ServiceConfiguration conf = tokenConfiguration(secretKey);
conf.setProperties(properties);
MultiRolesTokenAuthorizationProvider provider = new MultiRolesTokenAuthorizationProvider();
- provider.initialize(conf, mock(PulsarResources.class));
+ initializeProvider(provider, conf, mock(PulsarResources.class));
AuthenticationDataSource ads = new AuthenticationDataSource() {
@Override
@@ -287,11 +456,11 @@ public class MultiRolesTokenAuthorizationProviderTest {
String testAdminRole = "admin";
String token = Jwts.builder().claim("roles", testAdminRole).signWith(secretKey).compact();
- ServiceConfiguration conf = new ServiceConfiguration();
+ ServiceConfiguration conf = tokenConfiguration(secretKey);
conf.setSuperUserRoles(Set.of(testAdminRole));
MultiRolesTokenAuthorizationProvider provider = new MultiRolesTokenAuthorizationProvider();
- provider.initialize(conf, mock(PulsarResources.class));
+ initializeProvider(provider, conf, mock(PulsarResources.class));
AuthenticationDataSource ads = new AuthenticationDataSource() {
@Override
@@ -340,8 +509,8 @@ public class MultiRolesTokenAuthorizationProviderTest {
.signWith(secretKey).compact();
MultiRolesTokenAuthorizationProvider provider = new MultiRolesTokenAuthorizationProvider();
- ServiceConfiguration conf = new ServiceConfiguration();
- provider.initialize(conf, mock(PulsarResources.class));
+ ServiceConfiguration conf = tokenConfiguration(secretKey);
+ initializeProvider(provider, conf, mock(PulsarResources.class));
AuthenticationDataSource ads = new AuthenticationDataSource() {
@Override
@@ -395,8 +564,8 @@ public class MultiRolesTokenAuthorizationProviderTest {
.signWith(secretKey).compact();
MultiRolesTokenAuthorizationProvider provider = new MultiRolesTokenAuthorizationProvider();
- ServiceConfiguration conf = new ServiceConfiguration();
- provider.initialize(conf, mock(PulsarResources.class));
+ ServiceConfiguration conf = tokenConfiguration(secretKey);
+ initializeProvider(provider, conf, mock(PulsarResources.class));
AuthenticationDataSource ads = new AuthenticationDataSource() {
@Override
@@ -445,7 +614,7 @@ public class MultiRolesTokenAuthorizationProviderTest {
String token = Jwts.builder().claim("roles", new String[]{"user-a"}).signWith(secretKey).compact();
MultiRolesTokenAuthorizationProvider provider = new MultiRolesTokenAuthorizationProvider();
- provider.initialize(new ServiceConfiguration(), pulsarResources);
+ initializeProvider(provider, tokenConfiguration(secretKey), pulsarResources);
AuthenticationDataSource ads = new AuthenticationDataSource() {
@Override
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 692527a3536f..9b4bcfed2a62 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -397,8 +397,11 @@ public class BrokerService implements Closeable {
this.workerGroup = eventLoopGroup;
this.statsUpdater = new SingleThreadNonConcurrentFixedRateScheduler("pulsar-stats-updater");
+ this.authenticationService = new AuthenticationService(pulsar.getConfiguration(),
+ pulsar.getOpenTelemetry().getOpenTelemetry());
this.authorizationService = new AuthorizationService(
- pulsar.getConfiguration(), pulsar().getPulsarResources());
+ pulsar.getConfiguration(), pulsar().getPulsarResources(),
+ authenticationService);
this.entryFilterProvider = new EntryFilterProvider(pulsar.getConfiguration());
pulsar.getLocalMetadataStore().registerListener(this::handleMetadataChanges);
@@ -415,8 +418,6 @@ public class BrokerService implements Closeable {
this.backlogQuotaChecker = new SingleThreadNonConcurrentFixedRateScheduler("pulsar-backlog-quota-checker");
this.subscriptionBacklogAgeChecker =
new SingleThreadNonConcurrentFixedRateScheduler("pulsar-subscription-backlog-age-checker");
- this.authenticationService = new AuthenticationService(pulsar.getConfiguration(),
- pulsar.getOpenTelemetry().getOpenTelemetry());
this.topicFactory = createPersistentTopicFactory();
// update dynamic configuration and register-listener
updateConfigurationAndRegisterListeners();
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java
index fbdf62bbf943..169c9acb56d7 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java
@@ -413,7 +413,7 @@ public class ProxyConfiguration implements PulsarConfiguration {
+ "Authentication must be enabled via configuring `authenticationEnabled` to be true for this"
+ "to take effect"
)
- private boolean forwardAuthorizationCredentials = false;
+ private boolean forwardAuthorizationCredentials = true;
@FieldContext(
category = CATEGORY_AUTHENTICATION,
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java
index fb781c35df60..c16a3b8a04cc 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java
@@ -276,7 +276,7 @@ public class ProxyService implements Closeable {
pulsarResources = new PulsarResources(localMetadataStore, configMetadataStore);
discoveryProvider = new BrokerDiscoveryProvider(this.proxyConfig, pulsarResources);
authorizationService = new AuthorizationService(PulsarConfigurationLoader.convertFrom(proxyConfig),
- pulsarResources);
+ pulsarResources, authenticationService);
}
ServerBootstrap bootstrap = new ServerBootstrap();
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyAuthenticatedProducerConsumerTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyAuthenticatedProducerConsumerTest.java
index f740dd16ff81..9c2fed3ed459 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyAuthenticatedProducerConsumerTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyAuthenticatedProducerConsumerTest.java
@@ -85,6 +85,8 @@ public class ProxyAuthenticatedProducerConsumerTest extends ProducerConsumerBase
// enable tls and auth&auth at broker
conf.setAuthenticationEnabled(true);
+ // TLS client certificates are authenticated by the proxy, which forwards the original principal.
+ conf.setAuthenticateOriginalAuthData(false);
conf.setAuthorizationEnabled(true);
conf.setBrokerServicePortTls(Optional.of(0));
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyForwardAuthDataTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyForwardAuthDataTest.java
index 77324f4a87b2..e413924b9869 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyForwardAuthDataTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyForwardAuthDataTest.java
@@ -101,6 +101,7 @@ public class ProxyForwardAuthDataTest extends ProducerConsumerBase {
// Step 2: Run Pulsar Proxy without forwarding authData - expect Exception
ProxyConfiguration proxyConfig = new ProxyConfiguration();
+ proxyConfig.setForwardAuthorizationCredentials(false);
proxyConfig.setAuthenticationEnabled(true);
proxyConfig.setServicePort(Optional.of(0));
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationNegTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationNegTest.java
index 83c9bb556dce..396e6180367c 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationNegTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationNegTest.java
@@ -91,6 +91,8 @@ public class ProxyWithAuthorizationNegTest extends ProducerConsumerBase {
conf.setTopicLevelPoliciesEnabled(false);
conf.setAuthenticationEnabled(true);
+ // TLS client certificates are authenticated at the proxy and cannot be forwarded to the broker.
+ conf.setAuthenticateOriginalAuthData(false);
conf.setAuthorizationEnabled(true);
conf.setBrokerServicePortTls(Optional.of(0));
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java
index ba90173b0171..005f3ed04551 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java
@@ -182,6 +182,8 @@ public class ProxyWithAuthorizationTest extends ProducerConsumerBase {
super.doInitConf();
// enable tls and auth&auth at broker
conf.setAuthenticationEnabled(true);
+ // TLS client certificates are authenticated at the proxy and cannot be forwarded to the broker.
+ conf.setAuthenticateOriginalAuthData(false);
conf.setAuthorizationEnabled(true);
conf.setTopicLevelPoliciesEnabled(false);
conf.setProxyRoles(Collections.singleton("Proxy"));
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithJwtAuthorizationTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithJwtAuthorizationTest.java
index 6c1b874f3c90..d2b9ee7fbdd1 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithJwtAuthorizationTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithJwtAuthorizationTest.java
@@ -18,6 +18,8 @@
*/
package org.apache.pulsar.proxy.server;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.spy;
import com.google.common.collect.Sets;
import io.jsonwebtoken.Jwts;
@@ -36,6 +38,7 @@ import lombok.CustomLog;
import org.apache.pulsar.broker.authentication.AuthenticationProviderToken;
import org.apache.pulsar.broker.authentication.AuthenticationService;
import org.apache.pulsar.broker.authentication.utils.AuthTokenUtils;
+import org.apache.pulsar.broker.authorization.MultiRolesTokenAuthorizationProvider;
import org.apache.pulsar.broker.resources.PulsarResources;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.Authentication;
@@ -61,6 +64,8 @@ import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Factory;
import org.testng.annotations.Test;
@CustomLog
@@ -73,15 +78,28 @@ public class ProxyWithJwtAuthorizationTest extends ProducerConsumerBase {
private static final String CLIENT_ROLE = "client";
@SuppressWarnings("deprecation")
private static final SecretKey SECRET_KEY = AuthTokenUtils.createSecretKey(SignatureAlgorithm.HS256);
-@SuppressWarnings("deprecation")
- private static final String ADMIN_TOKEN = Jwts.builder().setSubject(ADMIN_ROLE).signWith(SECRET_KEY).compact();
- @SuppressWarnings("deprecation")
- private static final String PROXY_TOKEN = Jwts.builder().setSubject(PROXY_ROLE).signWith(SECRET_KEY).compact();
- @SuppressWarnings("deprecation")
- private static final String BROKER_TOKEN = Jwts.builder().setSubject(BROKER_ROLE).signWith(SECRET_KEY).compact();
- @SuppressWarnings("deprecation")
- private static final String CLIENT_TOKEN = Jwts.builder().setSubject(CLIENT_ROLE).signWith(SECRET_KEY).compact();
+ private final String adminToken;
+ private final String proxyToken;
+ private final String brokerToken;
+ private final String clientToken;
+ private final boolean multiRoles;
+ private final String roleClaim;
+
+ @Factory(dataProvider = "authorizationModes")
+ public ProxyWithJwtAuthorizationTest(boolean multiRoles, String roleClaim) {
+ this.multiRoles = multiRoles;
+ this.roleClaim = roleClaim;
+ adminToken = Jwts.builder().claim(roleClaim, ADMIN_ROLE).signWith(SECRET_KEY).compact();
+ proxyToken = Jwts.builder().claim(roleClaim, PROXY_ROLE).signWith(SECRET_KEY).compact();
+ brokerToken = Jwts.builder().claim(roleClaim, BROKER_ROLE).signWith(SECRET_KEY).compact();
+ clientToken = Jwts.builder().claim(roleClaim, CLIENT_ROLE).signWith(SECRET_KEY).compact();
+ }
+
+ @DataProvider
+ public static Object[][] authorizationModes() {
+ return new Object[][]{{false, "sub"}, {false, "roles"}, {true, "roles"}};
+ }
private ProxyService proxyService;
private WebServer webServer;
@@ -94,6 +112,12 @@ public class ProxyWithJwtAuthorizationTest extends ProducerConsumerBase {
// enable auth&auth and use JWT at broker
conf.setAuthenticationEnabled(true);
conf.setAuthorizationEnabled(true);
+ if (multiRoles) {
+ conf.setAuthorizationProvider(MultiRolesTokenAuthorizationProvider.class.getName());
+ }
+ if (!"sub".equals(roleClaim)) {
+ conf.getProperties().setProperty("tokenAuthClaim", roleClaim);
+ }
conf.getProperties().setProperty("tokenSecretKey", "data:;base64,"
+ Base64.getEncoder().encodeToString(SECRET_KEY.getEncoded()));
@@ -105,7 +129,7 @@ public class ProxyWithJwtAuthorizationTest extends ProducerConsumerBase {
conf.setProxyRoles(Collections.singleton(PROXY_ROLE));
conf.setBrokerClientAuthenticationPlugin(AuthenticationToken.class.getName());
- conf.setBrokerClientAuthenticationParameters(BROKER_TOKEN);
+ conf.setBrokerClientAuthenticationParameters(brokerToken);
Set<String> providers = new HashSet<>();
providers.add(AuthenticationProviderToken.class.getName());
conf.setAuthenticationProviders(providers);
@@ -118,6 +142,9 @@ public class ProxyWithJwtAuthorizationTest extends ProducerConsumerBase {
// start proxy service
proxyConfig.setAuthenticationEnabled(true);
proxyConfig.setAuthorizationEnabled(false);
+ if (!"sub".equals(roleClaim)) {
+ proxyConfig.getProperties().setProperty("tokenAuthClaim", roleClaim);
+ }
proxyConfig.getProperties().setProperty("tokenSecretKey", "data:;base64,"
+ Base64.getEncoder().encodeToString(SECRET_KEY.getEncoded()));
proxyConfig.setBrokerServiceURL(pulsar.getBrokerServiceUrl());
@@ -130,7 +157,7 @@ public class ProxyWithJwtAuthorizationTest extends ProducerConsumerBase {
// enable auth&auth and use JWT at proxy
proxyConfig.setBrokerClientAuthenticationPlugin(AuthenticationToken.class.getName());
- proxyConfig.setBrokerClientAuthenticationParameters(PROXY_TOKEN);
+ proxyConfig.setBrokerClientAuthenticationParameters(proxyToken);
proxyConfig.setAuthenticationProviders(providers);
proxyConfig.setStatusFilePath("./src/test/resources/vip_status.html");
@@ -160,6 +187,52 @@ public class ProxyWithJwtAuthorizationTest extends ProducerConsumerBase {
webServer.start();
}
+ @Test
+ public void testMultiRoleClientPermissionsWithSuperUserProxy() throws Exception {
+ assertThat(conf.isAuthenticateOriginalAuthData()).isTrue();
+ assertThat(proxyConfig.isForwardAuthorizationCredentials()).isTrue();
+ startProxy();
+ createAdminClient();
+ admin.clusters().createCluster(CLUSTER_NAME,
+ ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
+ admin.tenants().createTenant("multi-role",
+ new TenantInfoImpl(Set.of(ADMIN_ROLE), Set.of(CLUSTER_NAME)));
+ admin.namespaces().createNamespace("multi-role/ns");
+ String allowedTopic = "persistent://multi-role/ns/allowed";
+ String deniedTopic = "persistent://multi-role/ns/denied";
+ admin.topics().createNonPartitionedTopic(allowedTopic);
+ admin.topics().createNonPartitionedTopic(deniedTopic);
+ admin.topics().grantPermission(allowedTopic, CLIENT_ROLE, Set.of(AuthAction.produce, AuthAction.consume));
+ admin.topics().grantPermission(deniedTopic, "other-client", Set.of(AuthAction.produce, AuthAction.consume));
+
+ String token = Jwts.builder()
+ .claim(roleClaim, multiRoles ? new String[]{"unprivileged", CLIENT_ROLE} : CLIENT_ROLE)
+ .signWith(SECRET_KEY).compact();
+ @Cleanup
+ PulsarClient client = PulsarClient.builder().serviceUrl(proxyService.getServiceUrl())
+ .authentication(AuthenticationFactory.token(token)).operationTimeout(5, TimeUnit.SECONDS).build();
+ @Cleanup
+ Consumer<byte[]> consumer = client.newConsumer().topic(allowedTopic).subscriptionName("sub").subscribe();
+ @Cleanup
+ Producer<byte[]> producer = client.newProducer().topic(allowedTopic).create();
+ producer.send(new byte[]{1});
+ Message<byte[]> message = consumer.receive(5, TimeUnit.SECONDS);
+ assertThat(message).isNotNull();
+ assertThat(message.getData()).containsExactly((byte) 1);
+
+ assertThatThrownBy(() -> {
+ try (Producer<byte[]> ignored = client.newProducer().topic(deniedTopic).create()) {
+ // Creation should be rejected.
+ }
+ }).isInstanceOf(PulsarClientException.AuthorizationException.class);
+ assertThatThrownBy(() -> {
+ try (Consumer<byte[]> ignored = client.newConsumer().topic(deniedTopic)
+ .subscriptionName("sub").subscribe()) {
+ // Subscription should be rejected.
+ }
+ }).isInstanceOf(PulsarClientException.AuthorizationException.class);
+ }
+
/**
* <pre>
* It verifies jwt + Authentication + Authorization (client -> proxy -> broker).
@@ -523,14 +596,14 @@ public class ProxyWithJwtAuthorizationTest extends ProducerConsumerBase {
private void createAdminClient() throws Exception {
closeAdmin();
admin = spy(PulsarAdmin.builder().serviceHttpUrl(webServer.getServiceUri().toString())
- .authentication(AuthenticationFactory.token(ADMIN_TOKEN)).build());
+ .authentication(AuthenticationFactory.token(adminToken)).build());
}
@SuppressWarnings("deprecation")
private PulsarClient createPulsarClient(String proxyServiceUrl, ClientBuilder clientBuilder)
throws PulsarClientException {
return clientBuilder.serviceUrl(proxyServiceUrl).statsInterval(0, TimeUnit.SECONDS)
- .authentication(AuthenticationFactory.token(CLIENT_TOKEN))
+ .authentication(AuthenticationFactory.token(clientToken))
.operationTimeout(1000, TimeUnit.MILLISECONDS).build();
}
}
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithoutServiceDiscoveryTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithoutServiceDiscoveryTest.java
index 07a65fcce6e2..5e5483a13c69 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithoutServiceDiscoveryTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithoutServiceDiscoveryTest.java
@@ -63,6 +63,8 @@ public class ProxyWithoutServiceDiscoveryTest extends ProducerConsumerBase {
// enable tls and auth&auth at broker
conf.setAuthenticationEnabled(true);
+ // TLS client certificates are authenticated by the proxy, which forwards the original principal.
+ conf.setAuthenticateOriginalAuthData(false);
conf.setAuthorizationEnabled(true);
conf.setBrokerServicePortTls(Optional.of(0));
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java
index 1d49286467f0..5ba8c8747576 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java
@@ -109,16 +109,17 @@ public class WebSocketService implements Closeable {
pulsarResources = new PulsarResources(null, configMetadataStore);
}
+ // Start authentication first so authorization can reuse the initialized token provider.
+ authenticationService = new AuthenticationService(this.config);
// start authorizationService
if (config.isAuthorizationEnabled()) {
if (pulsarResources == null) {
throw new PulsarServerException(
"Failed to initialize authorization manager due to empty ConfigurationStoreServers");
}
- authorizationService = new AuthorizationService(this.config, pulsarResources);
+ authorizationService = new AuthorizationService(this.config, pulsarResources,
+ authenticationService);
}
- // start authentication service
- authenticationService = new AuthenticationService(this.config);
// initialize crypto key reader
String cryptoFactoryClassName = (String) config.getProperties().get("cryptoKeyReaderFactoryClassName");
if (StringUtils.isNotBlank(cryptoFactoryClassName)) {