Improve AuthTokenUtils to support a valid key file path. (apache#6607)

### Motivation

Some users may confuse by using `bin/pulsar tokens create` or `validate` with option `--secret-key` or `--private-key` at first time like below:

```
$ bin/pulsar tokens create --secret-key /opt/test-secret.key --subject test-user
Exception in thread "main" io.jsonwebtoken.io.DecodingException: Illegal base64 character: '-'
        at io.jsonwebtoken.io.Base64.ctoi(Base64.java:206)
        at io.jsonwebtoken.io.Base64.decodeFast(Base64.java:255)
        at io.jsonwebtoken.io.Base64Decoder.decode(Base64Decoder.java:21)
        at io.jsonwebtoken.io.Base64Decoder.decode(Base64Decoder.java:8)
        at io.jsonwebtoken.io.ExceptionPropagatingDecoder.decode(ExceptionPropagatingDecoder.java:21)
        at org.apache.pulsar.broker.authentication.utils.AuthTokenUtils.readKeyFromUrl(AuthTokenUtils.java:115)
        at org.apache.pulsar.utils.auth.tokens.TokensCliUtils$CommandCreateToken.run(TokensCliUtils.java:149)
        at org.apache.pulsar.utils.auth.tokens.TokensCliUtils.main(TokensCliUtils.java:320)
```
Current usage should specify file path starting with "file:///opt/cmss-secret.key".

### Modifications

Add support to read key from a valid file path in `AuthTokenUtils.readKeyFromUrl`.

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

@@ -24,11 +24,14 @@
 import io.jsonwebtoken.Jwts;
 import io.jsonwebtoken.SignatureAlgorithm;
 import io.jsonwebtoken.io.Decoders;
+import io.jsonwebtoken.io.DecodingException;
 import io.jsonwebtoken.io.Encoders;
 import io.jsonwebtoken.security.Keys;
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.security.Key;
 import java.security.KeyFactory;
 import java.security.PrivateKey;
@@ -42,6 +45,7 @@
 
 import lombok.experimental.UtilityClass;
 
+import org.apache.commons.codec.binary.Base64;
 import org.apache.pulsar.client.api.url.URL;
 
 @UtilityClass
@@ -108,9 +112,24 @@ public static byte[] readKeyFromUrl(String keyConfUrl) throws IOException {
             } catch (Exception e) {
                 throw new IOException(e);
             }
-        } else {
+        } else if (Files.exists(Paths.get(keyConfUrl))) {
+            // Assume the key content was passed in a valid file path
+            try {
+                return Files.readAllBytes(Paths.get(keyConfUrl));
+            } catch (IOException e) {
+                throw new IOException(e);
+            }
+        } else if (Base64.isBase64(keyConfUrl.getBytes())) {
             // Assume the key content was passed in base64
-            return Decoders.BASE64.decode(keyConfUrl);
+            try {
+                return Decoders.BASE64.decode(keyConfUrl);
+            } catch (DecodingException e) {
+                String msg = "Illegal base64 character or Key file " + keyConfUrl + " doesn't exist";
+                throw new IOException(msg, e);
+            }
+        } else {
+            String msg = "Secret/Public Key file " + keyConfUrl + " doesn't exist";
+            throw new IllegalArgumentException(msg);
         }
     }
 }

pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authentication/AuthenticationProviderTokenTest.java

@@ -229,6 +229,41 @@ public String getCommandData() {
         provider.close();
     }
 
+    @Test
+    public void testAuthSecretKeyFromValidFile() throws Exception {
+        SecretKey secretKey = AuthTokenUtils.createSecretKey(SignatureAlgorithm.HS256);
+
+        File secretKeyFile = File.createTempFile("pulsar-test-secret-key-valid", ".key");
+        secretKeyFile.deleteOnExit();
+        Files.write(Paths.get(secretKeyFile.toString()), secretKey.getEncoded());
+
+        AuthenticationProviderToken provider = new AuthenticationProviderToken();
+
+        Properties properties = new Properties();
+        properties.setProperty(AuthenticationProviderToken.CONF_TOKEN_SECRET_KEY, secretKeyFile.toString());
+
+        ServiceConfiguration conf = new ServiceConfiguration();
+        conf.setProperties(properties);
+        provider.initialize(conf);
+
+        String token = AuthTokenUtils.createToken(secretKey, SUBJECT, Optional.empty());
+
+        // Pulsar protocol auth
+        String subject = provider.authenticate(new AuthenticationDataSource() {
+            @Override
+            public boolean hasDataFromCommand() {
+                return true;
+            }
+
+            @Override
+            public String getCommandData() {
+                return token;
+            }
+        });
+        assertEquals(subject, SUBJECT);
+        provider.close();
+    }
+
     @Test
     public void testAuthSecretKeyFromDataBase64() throws Exception {
         SecretKey secretKey = AuthTokenUtils.createSecretKey(SignatureAlgorithm.HS256);
@@ -515,6 +550,33 @@ public void testInitializeWhenSecretKeyFilePathIsInvalid() throws IOException {
         new AuthenticationProviderToken().initialize(conf);
     }
 
+    @Test(expectedExceptions = IOException.class)
+    public void testInitializeWhenSecretKeyIsValidPathOrBase64() throws IOException {
+        Properties properties = new Properties();
+        properties.setProperty(AuthenticationProviderToken.CONF_TOKEN_SECRET_KEY,
+                "secret_key_file_not_exist");
+
+        ServiceConfiguration conf = new ServiceConfiguration();
+        conf.setProperties(properties);
+
+        new AuthenticationProviderToken().initialize(conf);
+    }
+
+    @Test(expectedExceptions = IllegalArgumentException.class)
+    public void testInitializeWhenSecretKeyFilePathIfNotExist() throws IOException {
+        File secretKeyFile = File.createTempFile("secret_key_file_not_exist", ".key");
+        assertTrue(secretKeyFile.delete());
+        assertFalse(secretKeyFile.exists());
+
+        Properties properties = new Properties();
+        properties.setProperty(AuthenticationProviderToken.CONF_TOKEN_SECRET_KEY, secretKeyFile.toString());
+
+        ServiceConfiguration conf = new ServiceConfiguration();
+        conf.setProperties(properties);
+
+        new AuthenticationProviderToken().initialize(conf);
+    }
+
     @Test(expectedExceptions = IOException.class)
     public void testInitializeWhenPublicKeyFilePathIsInvalid() throws IOException {
         Properties properties = new Properties();