PIP-25: Token based authentication

bin/pulsar

@@ -137,6 +137,7 @@ where command is one of:
     initialize-cluster-metadata     One-time metadata initialization
     compact-topic       Run compaction against a topic
     zookeeper-shell     Open a ZK shell client
+    tokens              Utility to create authentication tokens
 
     help                This help message
 
@@ -331,6 +332,8 @@ elif [ $COMMAND == "sql" ]; then
     exec $JAVA -cp "${PRESTO_HOME}/lib/*" com.facebook.presto.cli.Presto --server localhost:8081 "${@}"
 elif [ $COMMAND == "sql-worker" ]; then
     exec ${PRESTO_HOME}/bin/launcher --etc-dir ${PULSAR_PRESTO_CONF} "${@}"
+elif [ $COMMAND == "tokens" ]; then
+      exec $JAVA $OPTS org.apache.pulsar.utils.auth.tokens.TokensCliUtils $@
 elif [ $COMMAND == "help" ]; then
     pulsar_help;
 else

conf/broker.conf

@@ -283,6 +283,24 @@ athenzDomainNames=
 # When this parameter is not empty, unauthenticated users perform as anonymousUserRole
 anonymousUserRole=
 
+### --- Token Authentication Provider --- ###
+
+## Simmetric key
+# Configure the secret key to be used to validate auth tokens
+# The key can be specified like:
+# tokenSecretKey=data:xxxxxxxxx
+# tokenSecretKey=file:///my/secret.key
+# tokenSecretKey=env:MY_SECRET_KEY_VAR
+tokenSecretKey=
+
+## Asymmetric public/private key pair
+# Configure the public key to be used to validate auth tokens
+# The key can be specified like:
+# tokenPublicKey=data:xxxxxxxxx
+# tokenPublicKey=file:///my/public.key
+# tokenPublicKey=env:MY_PUBLIC_KEY_VAR
+tokenPublicKey=
+
 ### --- BookKeeper Client --- ###
 
 # Authentication plugin to use when connecting to bookies

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

@@ -469,10 +469,13 @@ The Apache Software License, Version 2.0
     - io.dropwizard.metrics-metrics-jvm-3.1.0.jar
   * Prometheus
     - io.prometheus-simpleclient_httpserver-0.5.0.jar
+  * Java JSON WebTokens
+    - io.jsonwebtoken-jjwt-api-0.10.5.jar
+    - io.jsonwebtoken-jjwt-impl-0.10.5.jar
+    - io.jsonwebtoken-jjwt-jackson-0.10.5.jar
   * JavaX Injection
     - javax.inject-javax.inject-1.jar
 
-
 BSD 3-clause "New" or "Revised" License
  * Google auth library
     - com.google.auth-google-auth-library-credentials-0.9.0.jar -- licenses/LICENSE-google-auth-library.txt

pom.xml

@@ -182,6 +182,7 @@ flexible messaging model and an intuitive client API.</description>
     <flink.version>1.6.0</flink.version>
     <scala.binary.version>2.11</scala.binary.version>
     <debezium.version>0.8.2</debezium.version>
+    <jsonwebtoken.version>0.10.5</jsonwebtoken.version>
     <opencensus.version>0.12.3</opencensus.version>
 
     <!-- test dependencies -->
@@ -750,6 +751,22 @@ flexible messaging model and an intuitive client API.</description>
           </exclusion>
         </exclusions>
       </dependency>
+
+      <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>
+        <version>${jsonwebtoken.version}</version>
+      </dependency>
 
       <dependency>
         <groupId>org.aspectj</groupId>

pulsar-broker-common/pom.xml

@@ -54,5 +54,14 @@
       <artifactId>javax.ws.rs-api</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>io.jsonwebtoken</groupId>
+      <artifactId>jjwt-impl</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>io.jsonwebtoken</groupId>
+      <artifactId>jjwt-jackson</artifactId>
+    </dependency>
   </dependencies>
 </project>

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

@@ -0,0 +1,123 @@
+/**
+ * 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 io.jsonwebtoken.Claims;
+import io.jsonwebtoken.Jwt;
+import io.jsonwebtoken.JwtException;
+import io.jsonwebtoken.Jwts;
+
+import java.io.IOException;
+import java.security.Key;
+
+import javax.naming.AuthenticationException;
+
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.authentication.utils.AuthTokenUtils;
+
+public class AuthenticationProviderToken implements AuthenticationProvider {
+
+    public final static String HTTP_HEADER_NAME = "Authorization";
+    final static String HTTP_HEADER_VALUE_PREFIX = "Bearer ";
+
+    // When simmetric key is configured
+    final static String CONF_TOKEN_SECRET_KEY = "tokenSecretKey";
+
+    // When public/private key pair is configured
+    final static String CONF_TOKEN_PUBLIC_KEY = "tokenPublicKey";
+
+    private Key validationKey;
+
+    @Override
+    public void close() throws IOException {
+        // noop
+    }
+
+    @Override
+    public void initialize(ServiceConfiguration config) throws IOException {
+        this.validationKey = getValidationKey(config);
+    }
+
+    @Override
+    public String getAuthMethodName() {
+        return "token";
+    }
+
+    @Override
+    public String authenticate(AuthenticationDataSource authData) throws AuthenticationException {
+        String token = null;
+
+        if (authData.hasDataFromCommand()) {
+            // Authenticate Pulsar binary connection
+            token = authData.getCommandData();
+        } else if (authData.hasDataFromHttp()) {
+            // Authentication HTTP request. 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)) {
+                throw new AuthenticationException("Invalid HTTP Authorization header");
+            }
+
+            // Remove prefix
+            token = httpHeaderValue.substring(HTTP_HEADER_VALUE_PREFIX.length());
+        } else {
+            throw new AuthenticationException("No token credentials passed");
+        }
+
+        // Validate the token
+        try {
+            @SuppressWarnings("unchecked")
+            Jwt<?, Claims> jwt = Jwts.parser()
+                    .setSigningKey(validationKey)
+                    .parse(token);
+
+            return jwt.getBody().getSubject();
+        } catch (JwtException e) {
+            throw new AuthenticationException("Failed to authentication token: " + e.getMessage());
+        }
+    }
+
+    /**
+     * Try to get the validation key for tokens from several possible config options.
+     */
+    private static Key getValidationKey(ServiceConfiguration conf) throws IOException {
+        final boolean isPublicKey;
+        final String validationKeyConfig;
+
+        if (conf.getProperty(CONF_TOKEN_SECRET_KEY) != null) {
+            isPublicKey = false;
+            validationKeyConfig = (String) conf.getProperty(CONF_TOKEN_SECRET_KEY);
+        } else if (conf.getProperty(CONF_TOKEN_PUBLIC_KEY) != null) {
+            isPublicKey = true;
+            validationKeyConfig = (String) conf.getProperty(CONF_TOKEN_PUBLIC_KEY);
+        } else {
+            throw new IOException("No secret key was provided for token authentication");
+        }
+
+        byte[] validationKey = AuthTokenUtils.readKeyFromUrl(validationKeyConfig);
+
+        if (isPublicKey) {
+            return AuthTokenUtils.decodePublicKey(validationKey);
+        } else {
+            return AuthTokenUtils.decodeSecretKey(validationKey);
+        }
+    }
+}

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

@@ -0,0 +1,128 @@
+/**
+ * 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.utils;
+
+import io.jsonwebtoken.JwtBuilder;
+import io.jsonwebtoken.Jwts;
+import io.jsonwebtoken.SignatureAlgorithm;
+import io.jsonwebtoken.io.Decoders;
+import io.jsonwebtoken.io.Encoders;
+import io.jsonwebtoken.security.Keys;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.security.Key;
+import java.security.KeyFactory;
+import java.security.PrivateKey;
+import java.security.PublicKey;
+import java.security.spec.PKCS8EncodedKeySpec;
+import java.security.spec.X509EncodedKeySpec;
+import java.util.Date;
+import java.util.Optional;
+
+import javax.crypto.SecretKey;
+
+import lombok.experimental.UtilityClass;
+
+@UtilityClass
+public class AuthTokenUtils {
+
+    public static SecretKey createSecretKey(SignatureAlgorithm signatureAlgorithm) {
+        return Keys.secretKeyFor(signatureAlgorithm);
+    }
+
+    public static SecretKey decodeSecretKey(byte[] secretKey) {
+        return Keys.hmacShaKeyFor(secretKey);
+    }
+
+    public static PrivateKey decodePrivateKey(byte[] key) throws IOException {
+        try {
+            PKCS8EncodedKeySpec spec = new PKCS8EncodedKeySpec(key);
+            KeyFactory kf = KeyFactory.getInstance("RSA");
+            return kf.generatePrivate(spec);
+        } catch (Exception e) {
+            throw new IOException("Failed to decode private key", e);
+        }
+    }
+
+    public static PublicKey decodePublicKey(byte[] key) throws IOException {
+        try {
+            X509EncodedKeySpec spec = new X509EncodedKeySpec(key);
+            KeyFactory kf = KeyFactory.getInstance("RSA");
+            return kf.generatePublic(spec);
+        } catch (Exception e) {
+            throw new IOException("Failed to decode public key", e);
+        }
+    }
+
+    public static String encodeKeyBase64(Key key) {
+        return Encoders.BASE64.encode(key.getEncoded());
+    }
+
+    public static String createToken(Key signingKey, String subject, Optional<Date> expiryTime) {
+        JwtBuilder builder = Jwts.builder()
+                .setSubject(subject)
+                .signWith(signingKey);
+
+        if (expiryTime.isPresent()) {
+            builder.setExpiration(expiryTime.get());
+        }
+
+        return builder.compact();
+    }
+
+    public static byte[] readKeyFromUrl(String keyConfUrl) throws IOException {
+        if (keyConfUrl.startsWith("data:")) {
+            return readDataUrl(keyConfUrl);
+        } else if (keyConfUrl.startsWith("env:")) {
+            String envVarName = keyConfUrl.substring("env:".length());
+            return Decoders.BASE64.decode(System.getenv(envVarName));
+        } else if (keyConfUrl.startsWith("file:")) {
+            URI filePath = URI.create(keyConfUrl);
+            return Files.readAllBytes(Paths.get(filePath));
+        } else {
+            // Assume the key content was passed in base64
+            return Decoders.BASE64.decode(keyConfUrl);
+        }
+    }
+
+    private static byte[] readDataUrl(String data) throws IOException {
+        // Expected format is:
+        // data:[<mediatype>][;base64],<data>
+
+        // Ignore mediatype and only support base64 encoding
+        String[] parts = data.split(",", 2);
+        String header = parts[0];
+        String encodedData = parts[1];
+
+        if (header.contains(";")) {
+            String encodingType = header.split(";")[1];
+            if (!"base64".equals(encodingType)) {
+                throw new IOException("Data url encoding not supported: " + encodingType);
+            }
+
+            return Decoders.BASE64.decode(encodedData);
+        } else {
+            // No encoding specified
+            return encodedData.getBytes();
+        }
+    }
+}