PIP-25: Token based authentication

conf/broker.conf

@@ -290,15 +290,13 @@ anonymousUserRole=
 # The key can be specified like:
 # tokenSecretKey=data:base64,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:base64,xxxxxxxxx
 # tokenPublicKey=file:///my/public.key
-# tokenPublicKey=env:MY_PUBLIC_KEY_VAR
 tokenPublicKey=
 
 ### --- BookKeeper Client --- ###

conf/proxy.conf

@@ -136,15 +136,13 @@ tlsRequireTrustedClientCertOnConnect=false
 # The key can be specified like:
 # tokenSecretKey=data:base64,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:base64,xxxxxxxxx
 # tokenPublicKey=file:///my/public.key
-# tokenPublicKey=env:MY_PUBLIC_KEY_VAR
 tokenPublicKey=
 
 

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

@@ -28,6 +28,7 @@
 
 import javax.naming.AuthenticationException;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.broker.authentication.utils.AuthTokenUtils;
 
@@ -102,10 +103,12 @@ private static Key getValidationKey(ServiceConfiguration conf) throws IOExceptio
         final boolean isPublicKey;
         final String validationKeyConfig;
 
-        if (conf.getProperty(CONF_TOKEN_SECRET_KEY) != null) {
+        if (conf.getProperty(CONF_TOKEN_SECRET_KEY) != null
+                && !StringUtils.isBlank((String) conf.getProperty(CONF_TOKEN_SECRET_KEY))) {
             isPublicKey = false;
             validationKeyConfig = (String) conf.getProperty(CONF_TOKEN_SECRET_KEY);
-        } else if (conf.getProperty(CONF_TOKEN_PUBLIC_KEY) != null) {
+        } else if (conf.getProperty(CONF_TOKEN_PUBLIC_KEY) != null
+                && !StringUtils.isBlank((String) conf.getProperty(CONF_TOKEN_PUBLIC_KEY))) {
             isPublicKey = true;
             validationKeyConfig = (String) conf.getProperty(CONF_TOKEN_PUBLIC_KEY);
         } else {

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

@@ -98,9 +98,6 @@ public static byte[] readKeyFromUrl(String keyConfUrl) throws IOException {
             } catch (Exception e) {
                 throw new IOException(e);
             }
-        } else if (keyConfUrl.startsWith("env:")) {
-            String envVarName = keyConfUrl.substring("env:".length());
-            return Decoders.BASE64.decode(System.getenv(envVarName));
         } else {
             // Assume the key content was passed in base64
             return Decoders.BASE64.decode(keyConfUrl);

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

@@ -23,6 +23,10 @@
 import com.beust.jcommander.Parameters;
 import com.google.common.base.Charsets;
 
+import io.jsonwebtoken.Claims;
+import io.jsonwebtoken.Jwt;
+import io.jsonwebtoken.JwtException;
+import io.jsonwebtoken.Jwts;
 import io.jsonwebtoken.SignatureAlgorithm;
 import io.jsonwebtoken.io.Decoders;
 import io.jsonwebtoken.io.Encoders;
@@ -40,6 +44,7 @@
 import java.util.concurrent.TimeUnit;
 
 import javax.crypto.SecretKey;
+import javax.naming.AuthenticationException;
 
 import lombok.Cleanup;
 
@@ -130,7 +135,7 @@ public void run() throws Exception {
                 System.exit(1);
             } else if (secretKey != null && privateKey != null) {
                 System.err.println(
-                        "Only one between --secret-key and --private-key needs to be passed for signing a token");
+                        "Only one of --secret-key and --private-key needs to be passed for signing a token");
                 System.exit(1);
             }
 
@@ -184,7 +189,7 @@ public void run() throws Exception {
                 token = System.getenv("TOKEN");
             } else {
                 System.err.println(
-                        "Token needs to be either passed as an argument or through `--stdin`, `--token-file` or by `TOKEN` environment variable");
+                        "Token needs to be either passed as an argument or through `--stdin`, `--token-file` or by the `TOKEN` environment variable");
                 System.exit(1);
                 return;
             }
@@ -196,6 +201,77 @@ public void run() throws Exception {
         }
     }
 
+    @Parameters(commandDescription = "Validate a token against a key")
+    public static class CommandValidateToken {
+
+        @Parameter(description = "The token string", arity = 1)
+        private java.util.List<String> args;
+
+        @Parameter(names = { "-i",
+                "--stdin" }, description = "Read token from standard input")
+        private Boolean stdin = false;
+
+        @Parameter(names = { "-f",
+                "--token-file" }, description = "Read token from a file")
+        private String tokenFile;
+
+        @Parameter(names = { "-sk",
+                "--secret-key" }, description = "Pass the secret key for validating the token. This can either be: data:, file:, etc..")
+        private String secretKey;
+
+        @Parameter(names = { "-pk",
+                "--public-key" }, description = "Pass the public key for validating the token. This can either be: data:, file:, etc..")
+        private String publicKey;
+
+        public void run() throws Exception {
+            if (secretKey == null && publicKey == null) {
+                System.err.println(
+                        "Either --secret-key or --public-key needs to be passed for signing a token");
+                System.exit(1);
+            } else if (secretKey != null && publicKey != null) {
+                System.err.println(
+                        "Only one of --secret-key and --public-key needs to be passed for signing a token");
+                System.exit(1);
+            }
+
+            String token;
+            if (args != null) {
+                token = args.get(0);
+            } else if (stdin) {
+                @Cleanup
+                BufferedReader r = new BufferedReader(new InputStreamReader(System.in));
+                token = r.readLine();
+            } else if (tokenFile != null) {
+                token = new String(Files.readAllBytes(Paths.get(tokenFile)), Charsets.UTF_8);
+            } else if (System.getenv("TOKEN") != null) {
+                token = System.getenv("TOKEN");
+            } else {
+                System.err.println(
+                        "Token needs to be either passed as an argument or through `--stdin`, `--token-file` or by the `TOKEN` environment variable");
+                System.exit(1);
+                return;
+            }
+
+            Key validationKey;
+
+            if (publicKey != null) {
+                byte[] encodedKey = AuthTokenUtils.readKeyFromUrl(publicKey);
+                validationKey = AuthTokenUtils.decodePublicKey(encodedKey);
+            } else {
+                byte[] encodedKey = AuthTokenUtils.readKeyFromUrl(secretKey);
+                validationKey = AuthTokenUtils.decodeSecretKey(encodedKey);
+            }
+
+            // Validate the token
+            @SuppressWarnings("unchecked")
+            Jwt<?, Claims> jwt = Jwts.parser()
+                    .setSigningKey(validationKey)
+                    .parse(token);
+
+            System.out.println(jwt.getBody());
+        }
+    }
+
     public static void main(String[] args) throws Exception {
         Arguments arguments = new Arguments();
         JCommander jcommander = new JCommander(arguments);
@@ -212,6 +288,9 @@ public static void main(String[] args) throws Exception {
         CommandShowToken commandShowToken = new CommandShowToken();
         jcommander.addCommand("show", commandShowToken);
 
+        CommandValidateToken commandValidateToken = new CommandValidateToken();
+        jcommander.addCommand("validate", commandValidateToken);
+
         try {
             jcommander.parse(args);
 
@@ -229,10 +308,14 @@ public static void main(String[] args) throws Exception {
 
         if (cmd.equals("create-secret-key")) {
             commandCreateSecretKey.run();
+        } else if (cmd.equals("create-key-pair")) {
+            commandCreateKeyPair.run();
         } else if (cmd.equals("create")) {
             commandCreateToken.run();
         } else if (cmd.equals("show")) {
             commandShowToken.run();
+        } else if (cmd.equals("validate")) {
+            commandValidateToken.run();
         } else {
             System.err.println("Invalid command: " + cmd);
             System.exit(1);

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

@@ -82,10 +82,6 @@ public void configure(String encodedAuthParamString) {
                     throw new RuntimeException("Failed to read token from file", e);
                 }
             };
-        } else if (encodedAuthParamString.startsWith("env:")) {
-            // Read token from environment variable
-            String envVarName = encodedAuthParamString.substring("env:".length());
-            this.tokenSupplier = () -> System.getenv(envVarName);
         } else {
             this.tokenSupplier = () -> encodedAuthParamString;
         }

pulsar-common/src/main/java/org/apache/pulsar/client/api/url/DataURLStreamHandler.java

@@ -68,10 +68,6 @@ public void connect() throws IOException {
                     this.contentType = "application/data";
                 }
 
-                for (int i =0; i < matcher.groupCount(); i++) {
-                    System.out.println("Group: " + matcher.group(i));
-                }
-
                 if (matcher.group("base64") == null) {
                     // Support Urlencode but not decode here because already decoded by URI class.
                     this.data = matcher.group("data").getBytes();

site2/docs/reference-configuration.md

@@ -142,8 +142,8 @@ Pulsar brokers are responsible for handling incoming messages from producers, di
 |tlsAllowInsecureConnection|  Accept untrusted TLS certificate from client  |false|
 |tlsProtocols|Specify the tls protocols the broker will use to negotiate during TLS Handshake. Multiple values can be specified, separated by commas. Example:- ```TLSv1.2```, ```TLSv1.1```, ```TLSv1``` ||
 |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`, `tokenSecretKey=file:///my/secret.key` or `tokenSecretKey=env:MY_SECRET_KEY_VAR`||
-|tokenPublicKey| Configure the public key to be used to validate auth tokens. The key can be specified like: `tokenPublicKey=data:base64,xxxxxxxxx`, `tokenPublicKey=file:///my/secret.key` or `tokenPublicKey=env:MY_SECRET_KEY_VAR`||
+|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`||
 |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|
 |maxConcurrentLookupRequest|  Max number of concurrent lookup request broker allows to throttle heavy incoming lookup traffic |50000|
@@ -440,8 +440,8 @@ The [Pulsar proxy](concepts-architecture-overview.md#pulsar-proxy) can be config
 |tlsRequireTrustedClientCertOnConnect|  Whether client certificates are required for TLS. Connections are rejected if the client certificate isn’t trusted. |false|
 |tlsProtocols|Specify the tls protocols the broker will use to negotiate during TLS Handshake. Multiple values can be specified, separated by commas. Example:- ```TLSv1.2```, ```TLSv1.1```, ```TLSv1``` ||
 |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`, `tokenSecretKey=file:///my/secret.key` or `tokenSecretKey=env:MY_SECRET_KEY_VAR`||
-|tokenPublicKey| Configure the public key to be used to validate auth tokens. The key can be specified like: `tokenPublicKey=data:base64,xxxxxxxxx`, `tokenPublicKey=file:///my/secret.key` or `tokenPublicKey=env:MY_SECRET_KEY_VAR`||
+|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`||
 
 ## ZooKeeper
 

tests/integration/src/test/java/org/apache/pulsar/tests/integration/auth/token/PulsarTokenAuthenticationBaseSuite.java

@@ -37,14 +37,17 @@
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.auth.AuthenticationToken;
 import org.apache.pulsar.common.policies.data.AuthAction;
 import org.apache.pulsar.common.policies.data.TenantInfo;
 import org.apache.pulsar.tests.integration.containers.BrokerContainer;
+import org.apache.pulsar.tests.integration.containers.ProxyContainer;
 import org.apache.pulsar.tests.integration.containers.PulsarContainer;
 import org.apache.pulsar.tests.integration.containers.ZKContainer;
 import org.apache.pulsar.tests.integration.topologies.PulsarCluster;
 import org.apache.pulsar.tests.integration.topologies.PulsarClusterSpec;
 import org.apache.pulsar.tests.integration.topologies.PulsarClusterTestBase;
+import org.testcontainers.containers.Network;
 import org.testng.ITest;
 import org.testng.annotations.AfterSuite;
 import org.testng.annotations.BeforeSuite;
@@ -54,20 +57,27 @@
 public abstract class PulsarTokenAuthenticationBaseSuite extends PulsarClusterTestBase implements ITest {
 
     protected String superUserAuthToken;
+    protected String proxyAuthToken;
     protected String clientAuthToken;
 
-    protected abstract void createKeysAndTokens(PulsarContainer container) throws Exception;
-
+    protected abstract void createKeysAndTokens(PulsarContainer<?> container) throws Exception;
     protected abstract void configureBroker(BrokerContainer brokerContainer) throws Exception;
+    protected abstract void configureProxy(ProxyContainer proxyContainer) throws Exception;
+
+    protected static final String SUPER_USER_ROLE = "super-user";
+    protected static final String PROXY_ROLE = "proxy";
+    protected static final String REGULAR_USER_ROLE = "client";
 
     @BeforeSuite
     @Override
     public void setupCluster() throws Exception {
         // Before starting the cluster, generate the secret key and the token
         // Use Zk container to have 1 container available before starting the cluster
-        try (ZKContainer zkContainer = new ZKContainer<>("cli-setup")) {
+        try (ZKContainer<?> zkContainer = new ZKContainer<>("cli-setup")) {
             zkContainer
-                    .withEnv("zkServers", ZKContainer.NAME);
+                .withNetwork(Network.newNetwork())
+                .withNetworkAliases(ZKContainer.NAME)
+                .withEnv("zkServers", ZKContainer.NAME);
             zkContainer.start();
 
             createKeysAndTokens(zkContainer);
@@ -80,6 +90,7 @@ public void setupCluster() throws Exception {
         PulsarClusterSpec spec = PulsarClusterSpec.builder()
                 .numBookies(2)
                 .numBrokers(1)
+                .numProxies(1)
                 .clusterName(clusterName)
                 .build();
 
@@ -92,11 +103,20 @@ public void setupCluster() throws Exception {
             configureBroker(brokerContainer);
             brokerContainer.withEnv("authenticationEnabled", "true");
             brokerContainer.withEnv("authenticationProviders",
-                    "org.apache.pulsar.broker.authentication.AuthenticationToken");
+                    "org.apache.pulsar.broker.authentication.AuthenticationProviderToken");
             brokerContainer.withEnv("authorizationEnabled", "true");
-            brokerContainer.withEnv("superUserRoles", "super-user");
+            brokerContainer.withEnv("superUserRoles", SUPER_USER_ROLE + "," + PROXY_ROLE);
         }
 
+        ProxyContainer proxyContainer = pulsarCluster.getProxy();
+        configureProxy(proxyContainer);
+        proxyContainer.withEnv("authenticationEnabled", "true");
+        proxyContainer.withEnv("authenticationProviders",
+                "org.apache.pulsar.broker.authentication.AuthenticationProviderToken");
+        proxyContainer.withEnv("authorizationEnabled", "true");
+        proxyContainer.withEnv("brokerClientAuthenticationPlugin", AuthenticationToken.class.getName());
+        proxyContainer.withEnv("brokerClientAuthenticationParameters", "token:" + proxyAuthToken);
+
         pulsarCluster.start();
 
         log.info("Cluster {} is setup", spec.clusterName());
@@ -125,12 +145,17 @@ public void testPublishWithTokenAuth() throws Exception {
                 .authentication(AuthenticationFactory.token(superUserAuthToken))
                 .build();
 
+        try {
         admin.tenants().createTenant(tenant,
-                new TenantInfo(Collections.singleton("regular-user"),
+                new TenantInfo(Collections.singleton(REGULAR_USER_ROLE),
                         Collections.singleton(pulsarCluster.getClusterName())));
 
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+
         admin.namespaces().createNamespace(namespace, Collections.singleton(pulsarCluster.getClusterName()));
-        admin.namespaces().grantPermissionOnNamespace(namespace, "regular-user", EnumSet.allOf(AuthAction.class));
+        admin.namespaces().grantPermissionOnNamespace(namespace, REGULAR_USER_ROLE, EnumSet.allOf(AuthAction.class));
 
         @Cleanup
         PulsarClient client = PulsarClient.builder()
@@ -139,7 +164,8 @@ public void testPublishWithTokenAuth() throws Exception {
                 .build();
 
         @Cleanup
-        Producer<String> producer = client.newProducer(Schema.STRING).topic(topic)
+        Producer<String> producer = client.newProducer(Schema.STRING)
+                .topic(topic)
                 .create();
 
         @Cleanup
@@ -168,7 +194,7 @@ public void testPublishWithTokenAuth() throws Exception {
                 .build();
 
         try {
-            client.newProducer(Schema.STRING).topic(topic)
+            clientNoAuth.newProducer(Schema.STRING).topic(topic)
                     .create();
             fail("Should have failed to create producer");
         } catch (PulsarClientException e) {