Enable specifying allowed offset when verifying athenz role token (ap…

…ache#3187)

### Motivation

We are using Athenz for client authentication. Occasionally, the following error occurs and client authentication fails.

> [pulsar-web-28-7] ERROR com.yahoo.athenz.auth.token.Token - Token:validate: token=v=Z1;d=xxx;r=xxx;p=xxx;a=xxx;t=1544027514;e=1544034714;k=0;i=xxx.xxx.xxx.xxx : has future timestamp=1544027514 : current time=1544027513 : allowed offset=0

This means that the timestamp included in the authentication token is more future than the server time. Since the difference between them is only 1 second, I think that the time of either server or client is slightly off.

This error can be avoided by increasing the value of `allowed offset`. Currently, this value is set to 0 in Pulsar, but the default value in Athenz ZMS seems to be 300 seconds.
https://github.com/yahoo/athenz/blob/93fe62c17f3ab4556c71c5136c1646df4a874a5f/servers/zms/conf/zms.properties#L277-L280

### Modifications

* Changed the default value of `allowed offset` from 0 to 30 (I think 300 seconds is too long)
* Enabled specifying `allowed offset` using system property

### Result

Even if the time of the server or client is slightly off, the authentication will succeed.

pulsar-broker-auth-athenz/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderAthenz.java

@@ -31,6 +31,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Lists;
 import com.yahoo.athenz.auth.token.RoleToken;
 import com.yahoo.athenz.zpe.AuthZpeClient;
@@ -41,8 +42,10 @@ public class AuthenticationProviderAthenz implements AuthenticationProvider {
     private static final String DOMAIN_NAME_LIST = "athenzDomainNames";
 
     private static final String SYS_PROP_DOMAIN_NAME_LIST = "pulsar.athenz.domain.names";
+    private static final String SYS_PROP_ALLOWED_OFFSET = "pulsar.athenz.role.token_allowed_offset";
 
     private List<String> domainNameList = null;
+    private int allowedOffset = 30;
 
     @Override
     public void initialize(ServiceConfiguration config) throws IOException {
@@ -57,6 +60,20 @@ public void initialize(ServiceConfiguration config) throws IOException {
 
         domainNameList = Lists.newArrayList(domainNames.split(","));
         log.info("Supported domain names for athenz: {}", domainNameList);
+
+        if (!StringUtils.isEmpty(System.getProperty(SYS_PROP_ALLOWED_OFFSET))) {
+            try {
+                allowedOffset = Integer.parseInt(System.getProperty(SYS_PROP_ALLOWED_OFFSET));
+            } catch (NumberFormatException e) {
+                throw new IOException("Invalid allowed offset for athenz role token verification specified", e);
+            }
+
+            if (allowedOffset < 0) {
+                throw new IOException("Allowed offset for athenz role token verification must not be negative");
+            }
+        }
+
+        log.info("Allowed offset for athenz role token verification: {} sec", allowedOffset);
     }
 
     @Override
@@ -103,7 +120,6 @@ public String authenticate(AuthenticationDataSource authData) throws Authenticat
         // Synchronize for non-thread safe static calls inside athenz library
         synchronized (this) {
             PublicKey ztsPublicKey = AuthZpeClient.getZtsPublicKey(token.getKeyId());
-            int allowedOffset = 0;
 
             if (ztsPublicKey == null) {
                 throw new AuthenticationException("Unable to retrieve ZTS Public Key");
@@ -123,5 +139,10 @@ public String authenticate(AuthenticationDataSource authData) throws Authenticat
     public void close() throws IOException {
     }
 
+    @VisibleForTesting
+    int getAllowedOffset() {
+        return this.allowedOffset;
+    }
+
     private static final Logger log = LoggerFactory.getLogger(AuthenticationProviderAthenz.class);
 }

pulsar-broker-auth-athenz/src/test/java/org/apache/pulsar/broker/authentication/AuthenticationProviderAthenzTest.java

@@ -29,6 +29,7 @@
 import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
 import org.apache.pulsar.broker.authentication.AuthenticationProviderAthenz;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Properties;
@@ -70,12 +71,38 @@ public void testInitilizeFromSystemPropeties() {
         ServiceConfiguration emptyConf = new ServiceConfiguration();
         Properties emptyProp = new Properties();
         emptyConf.setProperties(emptyProp);
-        AuthenticationProviderAthenz sysPropProvider = new AuthenticationProviderAthenz();
+        AuthenticationProviderAthenz sysPropProvider1 = new AuthenticationProviderAthenz();
         try {
-            sysPropProvider.initialize(emptyConf);
+            sysPropProvider1.initialize(emptyConf);
+            assertEquals(sysPropProvider1.getAllowedOffset(), 30); // default allowed offset is 30 sec
         } catch (Exception e) {
             fail("Fail to Read pulsar.athenz.domain.names from System Properties");
         }
+
+        System.setProperty("pulsar.athenz.role.token_allowed_offset", "0");
+        AuthenticationProviderAthenz sysPropProvider2 = new AuthenticationProviderAthenz();
+        try {
+            sysPropProvider2.initialize(config);
+            assertEquals(sysPropProvider2.getAllowedOffset(), 0);
+        } catch (Exception e) {
+            fail("Failed to get allowd offset from system property");
+        }
+
+        System.setProperty("pulsar.athenz.role.token_allowed_offset", "invalid");
+        AuthenticationProviderAthenz sysPropProvider3 = new AuthenticationProviderAthenz();
+        try {
+            sysPropProvider3.initialize(config);
+            fail("Invalid allowed offset should not be specified");
+        } catch (IOException e) {
+        }
+
+        System.setProperty("pulsar.athenz.role.token_allowed_offset", "-1");
+        AuthenticationProviderAthenz sysPropProvider4 = new AuthenticationProviderAthenz();
+        try {
+            sysPropProvider4.initialize(config);
+            fail("Negative allowed offset should not be specified");
+        } catch (IOException e) {
+        }
     }
 
     @Test
@@ -133,4 +160,4 @@ public void testAuthenticateSignedTokenWithDifferentDomain() throws Exception {
             // OK, expected
         }
     }
-}
+}