[feat] Pulsar authentication for Client authentication using private_key_jwt method

distribution/server/src/assemble/LICENSE.bin.txt

@@ -490,9 +490,9 @@ The Apache Software License, Version 2.0
   * OpenHFT
     - net.openhft-zero-allocation-hashing-0.16.jar
   * Java JSON WebTokens
-    - io.jsonwebtoken-jjwt-api-0.11.1.jar
-    - io.jsonwebtoken-jjwt-impl-0.11.1.jar
-    - io.jsonwebtoken-jjwt-jackson-0.11.1.jar
+    - io.jsonwebtoken-jjwt-api-0.11.5.jar
+    - io.jsonwebtoken-jjwt-impl-0.11.5.jar
+    - io.jsonwebtoken-jjwt-jackson-0.11.5.jar
   * JCTools - Java Concurrency Tools for the JVM
     - org.jctools-jctools-core-2.1.2.jar
   * Vertx

distribution/shell/src/assemble/LICENSE.bin.txt

@@ -413,6 +413,8 @@ The Apache Software License, Version 2.0
     - websocket-api-9.4.57.v20241219.jar
     - websocket-client-9.4.57.v20241219.jar
     - websocket-common-9.4.57.v20241219.jar
+ * Java JSON WebTokens
+    - jjwt-api-0.11.5.jar
  * SnakeYaml -- snakeyaml-2.0.jar
  * Google Error Prone Annotations - error_prone_annotations-2.38.0.jar
  * Javassist -- javassist-3.25.0-GA.jar

pom.xml

@@ -243,7 +243,7 @@ flexible messaging model and an intuitive client API.</description>
     <debezium.mysql.version>8.0.33</debezium.mysql.version>
     <!-- Override version that brings CVE-2022-3143 with debezium -->
     <wildfly-elytron.version>1.15.16.Final</wildfly-elytron.version>
-    <jsonwebtoken.version>0.11.1</jsonwebtoken.version>
+    <jsonwebtoken.version>0.11.5</jsonwebtoken.version>
     <opencensus.version>0.28.0</opencensus.version>
     <hadoop3.version>3.4.0</hadoop3.version>
     <dnsjava3.version>3.6.2</dnsjava3.version>
@@ -311,6 +311,7 @@ flexible messaging model and an intuitive client API.</description>
 
     <!-- test dependencies -->
     <testcontainers.version>1.20.4</testcontainers.version>
+    <mock-oauth2-server.version>2.1.11</mock-oauth2-server.version>
     <!-- Set docker-java.version to the version of docker-java used in org.testcontainers:testcontainers pom -->
     <docker-java.version>3.4.0</docker-java.version>
     <hamcrest.version>2.2</hamcrest.version>
@@ -1203,17 +1204,7 @@ flexible messaging model and an intuitive client API.</description>
 
       <dependency>
         <groupId>io.jsonwebtoken</groupId>
-        <artifactId>jjwt-api</artifactId>
-        <version>${jsonwebtoken.version}</version>
-      </dependency>
-      <dependency>
-        <groupId>io.jsonwebtoken</groupId>
-        <artifactId>jjwt-impl</artifactId>
-        <version>${jsonwebtoken.version}</version>
-      </dependency>
-      <dependency>
-        <groupId>io.jsonwebtoken</groupId>
-        <artifactId>jjwt-jackson</artifactId>
+        <artifactId>jjwt</artifactId>
         <version>${jsonwebtoken.version}</version>
       </dependency>
 
@@ -1727,6 +1718,26 @@ flexible messaging model and an intuitive client API.</description>
         <artifactId>opentelemetry-semconv</artifactId>
         <version>${opentelemetry.semconv.version}</version>
       </dependency>
+
+      <!-- JJWT -->
+      <dependency>
+        <groupId>io.jsonwebtoken</groupId>
+        <artifactId>jjwt-api</artifactId>
+        <version>${jsonwebtoken.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>io.jsonwebtoken</groupId>
+        <artifactId>jjwt-impl</artifactId>
+        <version>${jsonwebtoken.version}</version>
+        <scope>runtime</scope>
+      </dependency>
+      <dependency>
+        <groupId>io.jsonwebtoken</groupId>
+        <artifactId>jjwt-jackson</artifactId> <!-- or jjwt-gson if Gson is preferred -->
+        <version>${jsonwebtoken.version}</version>
+        <scope>runtime</scope>
+      </dependency>
+
     </dependencies>
   </dependencyManagement>
 

pulsar-broker-auth-oidc/pom.xml

@@ -34,10 +34,6 @@
   <description>Open ID Connect authentication plugin for broker</description>
   <name>Pulsar Broker Auth OIDC</name>
 
-  <properties>
-    <jsonwebtoken.version>0.11.5</jsonwebtoken.version>
-  </properties>
-
   <dependencies>
 
     <dependency>
@@ -106,13 +102,11 @@
     <dependency>
       <groupId>io.jsonwebtoken</groupId>
       <artifactId>jjwt-api</artifactId>
-      <version>${jsonwebtoken.version}</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>io.jsonwebtoken</groupId>
       <artifactId>jjwt-impl</artifactId>
-      <version>${jsonwebtoken.version}</version>
       <scope>test</scope>
     </dependency>
 

pulsar-broker-common/pom.xml

@@ -65,14 +65,19 @@
       <artifactId>jakarta.ws.rs-api</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>io.jsonwebtoken</groupId>
+      <artifactId>jjwt-api</artifactId>
+    </dependency>
     <dependency>
       <groupId>io.jsonwebtoken</groupId>
       <artifactId>jjwt-impl</artifactId>
+      <scope>runtime</scope>
     </dependency>
-
     <dependency>
       <groupId>io.jsonwebtoken</groupId>
       <artifactId>jjwt-jackson</artifactId>
+      <scope>runtime</scope>
     </dependency>
 
     <!-- test -->

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

@@ -35,6 +35,7 @@
 import java.security.Key;
 import java.util.Date;
 import java.util.List;
+import java.util.Set;
 import javax.naming.AuthenticationException;
 import javax.net.ssl.SSLSession;
 import javax.servlet.http.HttpServletRequest;
@@ -246,6 +247,14 @@ private Jwt<?, Claims> authenticateToken(final String token) throws Authenticati
                         throw new AuthenticationException("Audiences in token: ["
                                 + String.join(", ", audiences) + "] not contains this broker: " + audience);
                     }
+                } else if (object instanceof Set) {
+                        Set<String> audiences = (Set<String>) object;
+                        // audience not contains this broker, throw exception.
+                        if (!audiences.contains(audience)) {
+                            incrementFailureMetric(ErrorCode.INVALID_AUDIENCES);
+                            throw new AuthenticationException("Audiences in token: ["
+                                    + String.join(", ", audiences) + "] not contains this broker: " + audience);
+                        }
                 } else if (object instanceof String) {
                     if (!object.equals(audience)) {
                         incrementFailureMetric(ErrorCode.INVALID_AUDIENCES);

pulsar-broker/src/main/java/org/apache/pulsar/utils/auth/tokens/TokensCliUtils.java

@@ -20,7 +20,8 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import io.jsonwebtoken.Claims;
-import io.jsonwebtoken.Jwt;
+import io.jsonwebtoken.Jws;
+import io.jsonwebtoken.JwtException;
 import io.jsonwebtoken.Jwts;
 import io.jsonwebtoken.SignatureAlgorithm;
 import io.jsonwebtoken.io.Decoders;
@@ -290,13 +291,20 @@ public Integer call() throws Exception {
                 validationKey = AuthTokenUtils.decodeSecretKey(encodedKey);
             }
 
-            // Validate the token
-            Jwt<?, Claims> jwt = Jwts.parserBuilder()
-                    .setSigningKey(validationKey)
-                    .build()
-                    .parseClaimsJws(token);
+            try {
+                // Validate the token
+                Jws<Claims> jwt = Jwts.parserBuilder()
+                        .setSigningKey(validationKey)
+                        .build()
+                        .parseClaimsJws(token);
+
+                System.out.println(jwt.getBody());
+
+            } catch (JwtException e) {
+                System.err.println("JWT Signature verification failed: " + e.getMessage());
+                return -1;
+            }
 
-            System.out.println(jwt.getBody());
             return 0;
         }
     }

pulsar-broker/src/test/java/org/apache/pulsar/utils/auth/tokens/TokensCliUtilsTest.java

@@ -73,12 +73,12 @@ public void testCreateToken() {
             };
 
             new TokensCliUtils().execute(command);
-            String token = baoStream.toString();
+            String token = baoStream.toString().trim();
 
             Jwt<?, ?> jwt = Jwts.parserBuilder()
                     .setSigningKey(Decoders.BASE64.decode(secretKey))
                     .build()
-                    .parseClaimsJws(token);
+                    .parse(token);
 
             JwsHeader header = (JwsHeader) jwt.getHeader();
             String keyId = header.getKeyId();
@@ -108,7 +108,7 @@ public void commandCreateToken_WhenCreatingATokenWithExpiryTime_ShouldHaveTheDes
             };
 
             new TokensCliUtils().execute(command);
-            String token = baoStream.toString();
+            String token = baoStream.toString().trim();
 
             Instant start = (new Date().toInstant().plus(expireAsSec - 5, ChronoUnit.SECONDS));
             Instant stop = (new Date().toInstant().plus(expireAsSec + 5, ChronoUnit.SECONDS));

pulsar-client-messagecrypto-bc/src/main/java/org/apache/pulsar/client/impl/crypto/MessageCryptoBc.java

@@ -238,7 +238,7 @@ public static PublicKey loadPublicKey(byte[] keyBytes) throws Exception {
         return publicKey;
     }
 
-    private PrivateKey loadPrivateKey(byte[] keyBytes) throws Exception {
+    public static PrivateKey loadPrivateKey(byte[] keyBytes) throws Exception {
 
         Reader keyReader = new StringReader(new String(keyBytes));
         PrivateKey privateKey = null;

pulsar-client/pom.xml

@@ -191,6 +191,22 @@
       <optional>true</optional>
     </dependency>
 
+    <!-- JWT dependencies -->
+    <dependency>
+      <groupId>io.jsonwebtoken</groupId>
+      <artifactId>jjwt-api</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.jsonwebtoken</groupId>
+      <artifactId>jjwt-impl</artifactId>
+      <scope>runtime</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.jsonwebtoken</groupId>
+      <artifactId>jjwt-jackson</artifactId> <!-- or jjwt-gson if Gson is preferred -->
+      <scope>runtime</scope>
+    </dependency>
+
     <!-- Testing dependencies -->
     <dependency>
       <groupId>${project.groupId}</groupId>
@@ -212,6 +228,13 @@
       <scope>test</scope>
     </dependency>
 
+    <dependency>
+      <groupId>no.nav.security</groupId>
+      <artifactId>mock-oauth2-server</artifactId>
+      <version>${mock-oauth2-server.version}</version>
+      <scope>test</scope>
+    </dependency>
+
     <dependency>
       <groupId>org.roaringbitmap</groupId>
       <artifactId>RoaringBitmap</artifactId>

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

@@ -41,6 +41,7 @@ public class AuthenticationOAuth2 implements Authentication, EncodedAuthenticati
 
     public static final String CONFIG_PARAM_TYPE = "type";
     public static final String TYPE_CLIENT_CREDENTIALS = "client_credentials";
+    public static final String TYPE_JWT_BEARER = "jwt_bearer";
     public static final String AUTH_METHOD_NAME = "token";
     public static final double EXPIRY_ADJUSTMENT = 0.9;
     private static final long serialVersionUID = 1L;
@@ -80,6 +81,9 @@ public void configure(String encodedAuthParamString) {
             case TYPE_CLIENT_CREDENTIALS:
                 this.flow = ClientCredentialsFlow.fromParameters(params);
                 break;
+            case TYPE_JWT_BEARER:
+                this.flow = JwtBearerFlow.fromParameters(params);
+                break;
             default:
                 throw new IllegalArgumentException("Unsupported authentication type: " + type);
         }

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

@@ -19,17 +19,10 @@
 package org.apache.pulsar.client.impl.auth.oauth2;
 
 import java.io.IOException;
-import java.io.InputStream;
-import java.io.InputStreamReader;
-import java.io.Reader;
-import java.net.URISyntaxException;
 import java.net.URL;
-import java.net.URLConnection;
-import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import lombok.Builder;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.io.IOUtils;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.impl.auth.oauth2.protocol.ClientCredentialsExchangeRequest;
 import org.apache.pulsar.client.impl.auth.oauth2.protocol.ClientCredentialsExchanger;
@@ -81,7 +74,7 @@ public TokenResult authenticate() throws PulsarClientException {
         // read the private key from storage
         KeyFile keyFile;
         try {
-            keyFile = loadPrivateKey(this.privateKey);
+            keyFile = KeyFile.loadPrivateKey(this.privateKey);
         } catch (IOException e) {
             throw new PulsarClientException.AuthenticationException("Unable to read private key: " + e.getMessage());
         }
@@ -133,33 +126,4 @@ public static ClientCredentialsFlow fromParameters(Map<String, String> params) {
                 .build();
     }
 
-    /**
-     * Loads the private key from the given URL.
-     * @param privateKeyURL
-     * @return
-     * @throws IOException
-     */
-    private static KeyFile loadPrivateKey(String privateKeyURL) throws IOException {
-        try {
-            URLConnection urlConnection = new org.apache.pulsar.client.api.url.URL(privateKeyURL).openConnection();
-            try {
-                String protocol = urlConnection.getURL().getProtocol();
-                String contentType = urlConnection.getContentType();
-                if ("data".equals(protocol) && !"application/json".equals(contentType)) {
-                    throw new IllegalArgumentException(
-                            "Unsupported media type or encoding format: " + urlConnection.getContentType());
-                }
-                KeyFile privateKey;
-                try (Reader r = new InputStreamReader((InputStream) urlConnection.getContent(),
-                        StandardCharsets.UTF_8)) {
-                    privateKey = KeyFile.fromJson(r);
-                }
-                return privateKey;
-            } finally {
-                IOUtils.close(urlConnection);
-            }
-        } catch (URISyntaxException | InstantiationException | IllegalAccessException e) {
-            throw new IOException("Invalid privateKey format", e);
-        }
-    }
 }