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
 
@@ -325,6 +326,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

@@ -274,6 +274,24 @@ athenzDomainNames=
 # When this parameter is not empty, unauthenticated users perform as anonymousUserRole
 anonymousUserRole=
 
+### --- Token Authentication Provider --- ###
+
+## Simmetric key
+# Configure the secret key directly in this config file
+tokenSecretKey=
+# Read secret key from an environment variable
+tokenSecretKeyFromEnv=
+# Read secret key from a file
+tokenSecretKeyFromFile=
+
+## Asymmetric public/private key pair
+# Configure the public key directly in this config file
+tokenPublicKey=
+# Read public key from an environment variable
+tokenPublicKeyFromEnv=
+# Read public key from a file
+tokenPublicKeyFromFile=
+
 ### --- BookKeeper Client --- ###
 
 # Authentication plugin to use when connecting to bookies

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

@@ -469,6 +469,10 @@ 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
 
 
 BSD 3-clause "New" or "Revised" License

pom.xml

@@ -181,6 +181,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>
 
     <!-- test dependencies -->
     <arquillian-cube.version>1.15.1</arquillian-cube.version>
@@ -728,6 +729,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,163 @@
+/**
+ * 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 static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.common.base.Charsets;
+
+import io.jsonwebtoken.Claims;
+import io.jsonwebtoken.Jwt;
+import io.jsonwebtoken.JwtException;
+import io.jsonwebtoken.Jwts;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.security.Key;
+
+import javax.naming.AuthenticationException;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.authentication.utils.AuthTokenUtils;
+
+@Slf4j
+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";
+    final static String CONF_TOKEN_SECRET_KEY_FROM_ENV = "tokenSecretKeyFromEnv";
+    final static String CONF_TOKEN_SECRET_KEY_FROM_FILE = "tokenSecretKeyFromFile";
+
+    // When public/private key pair is configured
+    final static String CONF_TOKEN_PUBLIC_KEY = "tokenPublicKey";
+    final static String CONF_TOKEN_PUBLIC_KEY_FROM_ENV = "tokenPublicKeyFromEnv";
+    final static String CONF_TOKEN_PUBLIC_KEY_FROM_FILE = "tokenPublicKeyFromFile";
+
+    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.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 {
+
+        // First check for secret key confs
+        if (conf.getProperty(CONF_TOKEN_SECRET_KEY) != null) {
+            // Secret key was specified directly in confg file
+            Object secretKey = conf.getProperty(CONF_TOKEN_SECRET_KEY);
+            checkNotNull(secretKey);
+            checkArgument(secretKey instanceof String);
+            log.info("Reading secret key from config file");
+            return AuthTokenUtils.decodeSecretKey((String) secretKey);
+
+        } else if (conf.getProperty(CONF_TOKEN_SECRET_KEY_FROM_ENV) != null) {
+            // Secret key was specified in ENV variable
+            String variableName = (String) conf.getProperty(CONF_TOKEN_SECRET_KEY_FROM_ENV);
+            log.info("Reading secret key from env variable '{}'", variableName);
+            return AuthTokenUtils.decodeSecretKey(System.getenv(variableName));
+
+        } else if (conf.getProperty(CONF_TOKEN_SECRET_KEY_FROM_FILE) != null) {
+            String filePath = (String) conf.getProperty(CONF_TOKEN_SECRET_KEY_FROM_FILE);
+            log.info("Reading secret key from file '{}'", filePath);
+            return AuthTokenUtils
+                    .decodeSecretKey(new String(Files.readAllBytes(Paths.get(filePath)), Charsets.UTF_8));
+
+            // Then check for public key conf
+        } else if (conf.getProperty(CONF_TOKEN_PUBLIC_KEY) != null) {
+            // Public key was specified directly in confg file
+            Object publicKey = conf.getProperty(CONF_TOKEN_PUBLIC_KEY);
+            checkNotNull(publicKey);
+            checkArgument(publicKey instanceof String);
+            log.info("Reading public key from config file");
+            return AuthTokenUtils.decodePublicKey((String) publicKey);
+
+        } else if (conf.getProperty(CONF_TOKEN_PUBLIC_KEY_FROM_ENV) != null) {
+            // Public key was specified in ENV variable
+            String variableName = (String) conf.getProperty(CONF_TOKEN_PUBLIC_KEY_FROM_ENV);
+            log.info("Reading public key from env variable '{}'", variableName);
+            return AuthTokenUtils.decodePublicKey(System.getenv(variableName));
+
+        } else if (conf.getProperty(CONF_TOKEN_PUBLIC_KEY_FROM_FILE) != null) {
+            String filePath = (String) conf.getProperty(CONF_TOKEN_PUBLIC_KEY_FROM_FILE);
+            log.info("Reading public key from file '{}'", filePath);
+            return AuthTokenUtils
+                    .decodePublicKey(new String(Files.readAllBytes(Paths.get(filePath)), Charsets.UTF_8));
+
+        } else {
+            throw new IOException("No secret key was provided for token authentication");
+        }
+    }
+}

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

@@ -0,0 +1,89 @@
+/**
+ * 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.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 String createSecretKey(SignatureAlgorithm signatureAlgorithm) {
+        Key key = Keys.secretKeyFor(signatureAlgorithm);
+        return Encoders.BASE64.encode(key.getEncoded());
+    }
+
+    public static SecretKey decodeSecretKey(String secretKey) {
+        return Keys.hmacShaKeyFor(Decoders.BASE64.decode(secretKey));
+    }
+
+    public static PrivateKey decodePrivateKey(String key) throws IOException {
+        try {
+            PKCS8EncodedKeySpec spec = new PKCS8EncodedKeySpec(Decoders.BASE64.decode(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(String key) throws IOException {
+        try {
+            X509EncodedKeySpec spec = new X509EncodedKeySpec(Decoders.BASE64.decode(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 encodeKey(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();
+    }
+}