Implement configurable token auth claim (apache#3826)

conf/broker.conf

@@ -340,6 +340,9 @@ tokenSecretKey=
 # tokenPublicKey=file:///my/public.key
 tokenPublicKey=
 
+# The token "claim" that will be interpreted as the authentication "role" or "principal" by AuthenticationProviderToken (defaults to "sub" if blank)
+tokenAuthClaim=
+
 ### --- BookKeeper Client --- ###
 
 # Authentication plugin to use when connecting to bookies

conf/proxy.conf

@@ -183,6 +183,8 @@ tokenSecretKey=
 # tokenPublicKey=file:///my/public.key
 tokenPublicKey=
 
+# The token "claim" that will be interpreted as the authentication "role" or "principal" by AuthenticationProviderToken (defaults to "sub" if blank)
+tokenAuthClaim=
 
 ### --- Deprecated config variables --- ###
 

conf/standalone.conf

@@ -223,6 +223,9 @@ athenzDomainNames=
 # When this parameter is not empty, unauthenticated users perform as anonymousUserRole
 anonymousUserRole=
 
+# The token "claim" that will be interpreted as the authentication "role" or "principal" by AuthenticationProviderToken (defaults to "sub" if blank)
+tokenAuthClaim=
+
 ### --- BookKeeper Client --- ###
 
 # Authentication plugin to use when connecting to bookies

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

@@ -43,9 +43,13 @@ public class AuthenticationProviderToken implements AuthenticationProvider {
     // When public/private key pair is configured
     final static String CONF_TOKEN_PUBLIC_KEY = "tokenPublicKey";
 
+    // The token's claim that corresponds to the "role" string
+    final static String CONF_TOKEN_AUTH_CLAIM = "tokenAuthClaim";
+
     final static String TOKEN = "token";
 
     private Key validationKey;
+    private String roleClaim;
 
     @Override
     public void close() throws IOException {
@@ -55,6 +59,7 @@ public void close() throws IOException {
     @Override
     public void initialize(ServiceConfiguration config) throws IOException {
         this.validationKey = getValidationKey(config);
+        this.roleClaim = getTokenRoleClaim(config);
     }
 
     @Override
@@ -106,7 +111,7 @@ private String parseToken(final String token) throws AuthenticationException {
                     .setSigningKey(validationKey)
                     .parse(token);
 
-            return jwt.getBody().getSubject();
+            return jwt.getBody().get(roleClaim, String.class);
         } catch (JwtException e) {
             throw new AuthenticationException("Failed to authentication token: " + e.getMessage());
         }
@@ -130,4 +135,13 @@ private Key getValidationKey(ServiceConfiguration conf) throws IOException {
             throw new IOException("No secret key was provided for token authentication");
         }
     }
+
+    private String getTokenRoleClaim(ServiceConfiguration conf) throws IOException {
+        if (conf.getProperty(CONF_TOKEN_AUTH_CLAIM) != null
+                && StringUtils.isNotBlank((String) conf.getProperty(CONF_TOKEN_AUTH_CLAIM))) {
+            return (String) conf.getProperty(CONF_TOKEN_AUTH_CLAIM);
+        } else {
+            return Claims.SUBJECT;
+        }
+    }
 }

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

@@ -36,6 +36,7 @@
 import java.security.KeyPair;
 import java.security.PrivateKey;
 import java.sql.Date;
+import java.util.HashMap;
 import java.util.Optional;
 import java.util.Properties;
 import java.util.concurrent.TimeUnit;
@@ -293,6 +294,56 @@ public String getCommandData() {
         provider.close();
     }
 
+    @Test
+    public void testAuthSecretKeyPairWithCustomClaim() throws Exception {
+        String authRoleClaim = "customClaim";
+        String authRole = "my-test-role";
+
+        KeyPair keyPair = Keys.keyPairFor(SignatureAlgorithm.RS256);
+
+        String privateKeyStr = AuthTokenUtils.encodeKeyBase64(keyPair.getPrivate());
+        String publicKeyStr = AuthTokenUtils.encodeKeyBase64(keyPair.getPublic());
+
+        AuthenticationProviderToken provider = new AuthenticationProviderToken();
+
+        Properties properties = new Properties();
+        // Use public key for validation
+        properties.setProperty(AuthenticationProviderToken.CONF_TOKEN_PUBLIC_KEY, publicKeyStr);
+        // Set custom claim field
+        properties.setProperty(AuthenticationProviderToken.CONF_TOKEN_AUTH_CLAIM, authRoleClaim);
+
+        ServiceConfiguration conf = new ServiceConfiguration();
+        conf.setProperties(properties);
+        provider.initialize(conf);
+
+
+        // Use private key to generate token
+        PrivateKey privateKey = AuthTokenUtils.decodePrivateKey(Decoders.BASE64.decode(privateKeyStr));
+        String token = Jwts.builder()
+                .setClaims(new HashMap<String, Object>() {{
+                    put(authRoleClaim, authRole);
+                }})
+                .signWith(privateKey)
+                .compact();
+
+
+        // Pulsar protocol auth
+        String role = provider.authenticate(new AuthenticationDataSource() {
+            @Override
+            public boolean hasDataFromCommand() {
+                return true;
+            }
+
+            @Override
+            public String getCommandData() {
+                return token;
+            }
+        });
+        assertEquals(role, authRole);
+
+        provider.close();
+    }
+
     @Test(expectedExceptions = AuthenticationException.class)
     public void testAuthenticateWhenNoJwtPassed() throws AuthenticationException {
         AuthenticationProviderToken provider = new AuthenticationProviderToken();

site2/docs/reference-configuration.md

@@ -144,6 +144,7 @@ Pulsar brokers are responsible for handling incoming messages from producers, di
 |tlsCiphers|Specify the tls cipher the broker will use to negotiate during TLS Handshake. Multiple values can be specified, separated by commas. Example:- ```TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256```||
 |tokenSecretKey| Configure the secret key to be used to validate auth tokens. The key can be specified like: `tokenSecretKey=data:base64,xxxxxxxxx` or `tokenSecretKey=file:///my/secret.key`||
 |tokenPublicKey| Configure the public key to be used to validate auth tokens. The key can be specified like: `tokenPublicKey=data:base64,xxxxxxxxx` or `tokenPublicKey=file:///my/secret.key`||
+|tokenAuthClaim| Specify which of the token's claims will be used as the authentication "principal" or "role". The default "sub" claim will be used if this is left blank ||
 |maxUnackedMessagesPerConsumer| Max number of unacknowledged messages allowed to receive messages by a consumer on a shared subscription. Broker will stop sending messages to consumer once, this limit reaches until consumer starts acknowledging messages back. Using a value of 0, is disabling unackeMessage limit check and consumer can receive messages without any restriction  |50000|
 |maxUnackedMessagesPerSubscription| Max number of unacknowledged messages allowed per shared subscription. Broker will stop dispatching messages to all consumers of the subscription once this limit reaches until consumer starts acknowledging messages back and unack count reaches to limit/2. Using a value of 0, is disabling unackedMessage-limit check and dispatcher can dispatch messages without any restriction  |200000|
 |subscriptionRedeliveryTrackerEnabled| Enable subscription message redelivery tracker |true|
@@ -448,6 +449,7 @@ The [Pulsar proxy](concepts-architecture-overview.md#pulsar-proxy) can be config
 |tlsCiphers|Specify the tls cipher the broker will use to negotiate during TLS Handshake. Multiple values can be specified, separated by commas. Example:- ```TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256```||
 |tokenSecretKey| Configure the secret key to be used to validate auth tokens. The key can be specified like: `tokenSecretKey=data:base64,xxxxxxxxx` or `tokenSecretKey=file:///my/secret.key`||
 |tokenPublicKey| Configure the public key to be used to validate auth tokens. The key can be specified like: `tokenPublicKey=data:base64,xxxxxxxxx` or `tokenPublicKey=file:///my/secret.key`||
+|tokenAuthClaim| Specify the token claim that will be used as the authentication "principal" or "role". The "subject" field will be used if this is left blank ||
 
 ## ZooKeeper