PIP-25: Token based authentication

conf/broker.conf

@@ -285,18 +285,18 @@ anonymousUserRole=
 
 ### --- Token Authentication Provider --- ###
 
-## Simmetric key
+## Symmetric key
 # Configure the secret key to be used to validate auth tokens
 # The key can be specified like:
-# tokenSecretKey=data:xxxxxxxxx
+# 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:xxxxxxxxx
+# tokenPublicKey=data:base64,xxxxxxxxx
 # tokenPublicKey=file:///my/public.key
 # tokenPublicKey=env:MY_PUBLIC_KEY_VAR
 tokenPublicKey=

conf/proxy.conf

@@ -129,6 +129,24 @@ tlsHostnameVerificationEnabled=false
 # certificate isn't trusted.
 tlsRequireTrustedClientCertOnConnect=false
 
+### --- Token Authentication Provider --- ###
+
+## Symmetric key
+# 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
+# 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=
+
 
 ### --- Deprecated config variables --- ###
 

pulsar-broker-common/pom.xml

@@ -38,6 +38,12 @@
       <artifactId>pulsar-zookeeper-utils</artifactId>
       <version>${project.version}</version>
     </dependency>
+
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>pulsar-common</artifactId>
+      <version>${project.version}</version>
+    </dependency>
 
     <dependency>
       <groupId>com.google.guava</groupId>

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

@@ -18,6 +18,8 @@
  */
 package org.apache.pulsar.broker.authentication.utils;
 
+import com.google.common.io.ByteStreams;
+
 import io.jsonwebtoken.JwtBuilder;
 import io.jsonwebtoken.Jwts;
 import io.jsonwebtoken.SignatureAlgorithm;
@@ -26,9 +28,7 @@
 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.io.InputStream;
 import java.security.Key;
 import java.security.KeyFactory;
 import java.security.PrivateKey;
@@ -42,6 +42,8 @@
 
 import lombok.experimental.UtilityClass;
 
+import org.apache.pulsar.client.api.url.URL;
+
 @UtilityClass
 public class AuthTokenUtils {
 
@@ -90,39 +92,18 @@ public static String createToken(Key signingKey, String subject, Optional<Date>
     }
 
     public static byte[] readKeyFromUrl(String keyConfUrl) throws IOException {
-        if (keyConfUrl.startsWith("data:")) {
-            return readDataUrl(keyConfUrl);
+        if (keyConfUrl.startsWith("data:") || keyConfUrl.startsWith("file:")) {
+            try {
+                return ByteStreams.toByteArray((InputStream) new URL(keyConfUrl).getContent());
+            } 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 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();
-        }
-    }
 }

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

@@ -64,7 +64,7 @@ public static class CommandCreateSecretKey {
         String outputFile;
 
         @Parameter(names = {
-                "-b", "--base-64" }, description = "Encode the key in base64")
+                "-b", "--base64" }, description = "Encode the key in base64")
         boolean base64 = false;
 
         public void run() throws IOException {
@@ -115,22 +115,32 @@ public static class CommandCreateToken {
                 "--expiry-time" }, description = "Relative expiry time for the token (eg: 1h, 3d, 10y). (m=minutes) Default: no expiration")
         private String expiryTime;
 
-        @Parameter(names = { "-pk",
-                "--is-private-key" }, description = "Indicate the signing key is a private key (rather than a symmetric secret key)")
-        private Boolean isPrivateKey = false;
+        @Parameter(names = { "-sk",
+                "--secret-key" }, description = "Pass the secret key for signing the token. This can either be: data:, file:, etc..")
+        private String secretKey;
 
-        @Parameter(names = { "-k",
-                "--signing-key" }, description = "Pass the signing key. This can either be: data:, file:, etc..", required = true)
-        private String key;
+        @Parameter(names = { "-pk",
+                "--private-key" }, description = "Pass the private key for signing the token. This can either be: data:, file:, etc..")
+        private String privateKey;
 
         public void run() throws Exception {
-            byte[] encodedKey = AuthTokenUtils.readKeyFromUrl(key);
+            if (secretKey == null && privateKey == null) {
+                System.err.println(
+                        "Either --secret-key or --private-key needs to be passed for signing a token");
+                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");
+                System.exit(1);
+            }
 
             Key signingKey;
 
-            if (isPrivateKey) {
+            if (privateKey != null) {
+                byte[] encodedKey = AuthTokenUtils.readKeyFromUrl(privateKey);
                 signingKey = AuthTokenUtils.decodePrivateKey(encodedKey);
             } else {
+                byte[] encodedKey = AuthTokenUtils.readKeyFromUrl(secretKey);
                 signingKey = AuthTokenUtils.decodeSecretKey(encodedKey);
             }
 
@@ -157,7 +167,7 @@ public static class CommandShowToken {
         private Boolean stdin = false;
 
         @Parameter(names = { "-f",
-                "--token-file" }, description = "Read tokn from a file")
+                "--token-file" }, description = "Read token from a file")
         private String tokenFile;
 
         public void run() throws Exception {
@@ -174,7 +184,7 @@ public void run() throws Exception {
                 token = System.getenv("TOKEN");
             } else {
                 System.err.println(
-                        "Token needs to be either passed 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 `TOKEN` environment variable");
                 System.exit(1);
                 return;
             }

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

@@ -33,12 +33,15 @@
 
 public class DataURLStreamHandler extends URLStreamHandler {
 
-    class DataURLConnection extends URLConnection {
+    static class DataURLConnection extends URLConnection {
         private boolean parsed = false;
         private String contentType;
-        private String data;
+        private byte[] data;
         private URI uri;
 
+        private static final Pattern pattern = Pattern.compile(
+              "(?<mimeType>[^;,]+)?(;(?<charset>charset=[^;,]+))?(;(?<base64>base64))?,(?<data>.+)", Pattern.DOTALL);
+
         protected DataURLConnection(URL url) {
             super(url);
             try {
@@ -57,20 +60,23 @@ public void connect() throws IOException {
             if (this.uri == null) {
                 throw new IOException();
             }
-            Pattern pattern = Pattern.compile(
-                    "(?<mimeType>.+?)(;(?<charset>charset=.+?))?(;(?<base64>base64?))?,(?<data>.+)", Pattern.DOTALL);
+
             Matcher matcher = pattern.matcher(this.uri.getSchemeSpecificPart());
             if (matcher.matches()) {
                 this.contentType = matcher.group("mimeType");
-                String charset = matcher.group("charset");
-                if (charset == null) {
-                    charset = "US-ASCII";
+                if (contentType == null) {
+                    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 = new String(matcher.group("data").getBytes(), charset);
+                    this.data = matcher.group("data").getBytes();
                 } else {
-                    this.data = new String(Base64.getDecoder().decode(matcher.group("data")), charset);
+                    this.data = Base64.getDecoder().decode(matcher.group("data"));
                 }
             } else {
                 throw new MalformedURLException();
@@ -83,7 +89,7 @@ public long getContentLengthLong() {
             long length;
             try {
                 this.connect();
-                length = this.data.length();
+                length = this.data.length;
             } catch (IOException e) {
                 length = -1;
             }
@@ -109,7 +115,7 @@ public String getContentEncoding() {
 
         public InputStream getInputStream() throws IOException {
             this.connect();
-            return new ByteArrayInputStream(this.data.getBytes());
+            return new ByteArrayInputStream(this.data);
         }
     }
 

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

@@ -24,7 +24,7 @@
 import java.util.Map;
 
 public class PulsarURLStreamHandlerFactory implements URLStreamHandlerFactory {
-    static Map<String, Class<? extends URLStreamHandler>> handlers;
+    private static final Map<String, Class<? extends URLStreamHandler>> handlers;
     static {
         handlers = new HashMap<>();
         handlers.put("data", DataURLStreamHandler.class);

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

@@ -26,8 +26,8 @@
 import java.net.URLStreamHandlerFactory;
 
 public class URL {
-    private static URLStreamHandlerFactory urlStreamHandlerFactory = new PulsarURLStreamHandlerFactory();
-    private java.net.URL url;
+    private static final URLStreamHandlerFactory urlStreamHandlerFactory = new PulsarURLStreamHandlerFactory();
+    private final java.net.URL url;
 
     public URL(String spec)
             throws MalformedURLException, URISyntaxException, InstantiationException, IllegalAccessException {
@@ -47,7 +47,7 @@ public Object getContent() throws IOException {
         return this.url.getContent();
     }
 
-    public Object getContent(Class[] classes) throws IOException {
+    public Object getContent(Class<?>[] classes) throws IOException {
         return this.url.getContent(classes);
     }
 

pulsar-common/src/main/java/org/apache/pulsar/common/util/RelativeTimeUtil.java

@@ -29,15 +29,21 @@ public static long parseRelativeTimeInSeconds(String relativeTime) {
             throw new IllegalArgumentException("exipiry time cannot be empty");
         }
 
-        char lastChar = relativeTime.charAt(relativeTime.length() - 1);
+        int lastIndex=  relativeTime.length() - 1;
+        char lastChar = relativeTime.charAt(lastIndex);
+        final char timeUnit;
 
         if (!Character.isAlphabetic(lastChar)) {
-            throw new IllegalArgumentException("Relative time should contain time unit. eg: 3h or 5d");
+            // No unit specified, assume seconds
+            timeUnit = 's';
+            lastIndex = relativeTime.length();
+        } else {
+            timeUnit = Character.toLowerCase(lastChar);
         }
 
-        long duration = Long.parseLong(relativeTime.substring(0, relativeTime.length() - 1));
+        long duration = Long.parseLong(relativeTime.substring(0, lastIndex));
 
-        switch (lastChar) {
+        switch (timeUnit) {
         case 's':
             return duration;
         case 'm':

pulsar-common/src/test/java/org/apache/pulsar/common/util/RelativeTimeUtilTest.java

@@ -28,11 +28,19 @@
 public class RelativeTimeUtilTest {
     @Test
     public void testParseRelativeTime() {
+        assertEquals(RelativeTimeUtil.parseRelativeTimeInSeconds("-1"), -1);
+        assertEquals(RelativeTimeUtil.parseRelativeTimeInSeconds("7"), 7);
         assertEquals(RelativeTimeUtil.parseRelativeTimeInSeconds("3s"), 3);
+        assertEquals(RelativeTimeUtil.parseRelativeTimeInSeconds("3S"), 3);
         assertEquals(RelativeTimeUtil.parseRelativeTimeInSeconds("5m"), TimeUnit.MINUTES.toSeconds(5));
+        assertEquals(RelativeTimeUtil.parseRelativeTimeInSeconds("5M"), TimeUnit.MINUTES.toSeconds(5));
         assertEquals(RelativeTimeUtil.parseRelativeTimeInSeconds("7h"), TimeUnit.HOURS.toSeconds(7));
+        assertEquals(RelativeTimeUtil.parseRelativeTimeInSeconds("7H"), TimeUnit.HOURS.toSeconds(7));
         assertEquals(RelativeTimeUtil.parseRelativeTimeInSeconds("9d"), TimeUnit.DAYS.toSeconds(9));
+        assertEquals(RelativeTimeUtil.parseRelativeTimeInSeconds("9D"), TimeUnit.DAYS.toSeconds(9));
+        assertEquals(RelativeTimeUtil.parseRelativeTimeInSeconds("3w"), 7 * TimeUnit.DAYS.toSeconds(3));
         assertEquals(RelativeTimeUtil.parseRelativeTimeInSeconds("11y"), 365 * TimeUnit.DAYS.toSeconds(11));
+        assertEquals(RelativeTimeUtil.parseRelativeTimeInSeconds("11Y"), 365 * TimeUnit.DAYS.toSeconds(11));
 
         // Negative interval
         assertEquals(RelativeTimeUtil.parseRelativeTimeInSeconds("-5m"), -TimeUnit.MINUTES.toSeconds(5));
@@ -44,14 +52,6 @@ public void testParseRelativeTime() {
             // expected
         }
 
-        try {
-            // No time unit specified
-            RelativeTimeUtil.parseRelativeTimeInSeconds("1234");
-            fail("should have failed");
-        } catch (IllegalArgumentException e) {
-            // expected
-        }
-
         try {
             // Invalid time unit specified
             RelativeTimeUtil.parseRelativeTimeInSeconds("1234x");