Add Athenz authentication plugin (apache#178)

conf/broker.conf

@@ -122,6 +122,8 @@ superUserRoles=
 brokerClientAuthenticationPlugin=
 brokerClientAuthenticationParameters=
 
+# Supported Athenz provider domain names(comma separated) for authentication
+athenzDomainNames=
 
 ### --- BookKeeper Client --- ###
 

conf/standalone.conf

@@ -95,6 +95,8 @@ superUserRoles=
 brokerClientAuthenticationPlugin=
 brokerClientAuthenticationParameters=
 
+# Supported Athenz provider domain names(comma separated) for authentication
+athenzDomainNames=
 
 ### --- BookKeeper Client --- ###
 

pom.xml

@@ -81,6 +81,8 @@ flexible messaging model and an intuitive client API.</description>
     <module>pulsar-zookeeper-utils</module>
     <module>pulsar-checksum</module>
     <module>pulsar-testclient</module>
+    <module>pulsar-broker-auth-athenz</module>
+    <module>pulsar-client-auth-athenz</module>
     <module>all</module>
   </modules>
 
@@ -388,6 +390,19 @@ flexible messaging model and an intuitive client API.</description>
         <artifactId>caffeine</artifactId>
         <version>2.3.3</version>
       </dependency>
+
+      <dependency>
+        <groupId>com.yahoo.athenz</groupId>
+        <artifactId>zts_java_client</artifactId>
+        <version>1.1.3</version>
+      </dependency>
+
+      <dependency>
+        <groupId>com.yahoo.athenz</groupId>
+        <artifactId>zpe_java_client</artifactId>
+        <version>1.1.3</version>
+      </dependency>
+
     </dependencies>
   </dependencyManagement>
 
@@ -710,6 +725,14 @@ flexible messaging model and an intuitive client API.</description>
       <id>bookkeeper-yahoo-mvn-repo</id>
       <url>https://raw.githubusercontent.com/yahoo/bookkeeper/mvn-repo</url>
     </repository>
+    <repository>
+      <snapshots>
+        <enabled>false</enabled>
+      </snapshots>
+      <id>bintray-yahoo-maven</id>
+      <name>bintray</name>
+      <url>http://yahoo.bintray.com/maven</url>
+    </repository>
   </repositories>
 
   <distributionManagement>

pulsar-broker-auth-athenz/pom.xml

@@ -0,0 +1,47 @@
+<?xml version="1.0"?>
+<!--
+
+    Copyright 2016 Yahoo Inc.
+
+    Licensed 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.
+
+-->
+<project
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
+  xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>com.yahoo.pulsar</groupId>
+    <artifactId>pulsar</artifactId>
+    <version>1.17-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>pulsar-broker-auth-athenz</artifactId>
+  <packaging>jar</packaging>
+  <description>Athenz authentication plugin for broker</description>
+
+  <dependencies>
+
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>pulsar-broker</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>com.yahoo.athenz</groupId>
+      <artifactId>zpe_java_client</artifactId>
+    </dependency>
+
+  </dependencies>
+</project>

pulsar-broker-auth-athenz/src/main/java/com/yahoo/pulsar/broker/authentication/AuthenticationProviderAthenz.java

@@ -0,0 +1,116 @@
+/**
+ * Copyright 2016 Yahoo Inc.
+ *
+ * Licensed 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 com.yahoo.pulsar.broker.authentication;
+
+import java.io.IOException;
+import java.net.SocketAddress;
+import java.util.List;
+import java.security.PublicKey;
+
+import javax.naming.AuthenticationException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+import com.yahoo.athenz.auth.token.RoleToken;
+import com.yahoo.athenz.zpe.AuthZpeClient;
+import com.yahoo.pulsar.broker.ServiceConfiguration;
+import com.yahoo.pulsar.broker.authentication.AuthenticationDataSource;
+import com.yahoo.pulsar.broker.authentication.AuthenticationProvider;
+
+public class AuthenticationProviderAthenz implements AuthenticationProvider {
+
+    private static final String DOMAIN_NAME_LIST = "athenzDomainNames";
+
+    private List<String> domainNameList = null;
+
+    @Override
+    public void initialize(ServiceConfiguration config) throws IOException {
+        if (config.getProperty(DOMAIN_NAME_LIST) == null) {
+            throw new IOException("No athenz domain name specified");
+        }
+        String domainNames = (String) config.getProperty(DOMAIN_NAME_LIST);
+        domainNameList = Lists.newArrayList(domainNames.split(","));
+        log.info("Supported domain names for athenz: {}", domainNameList);
+    }
+
+    @Override
+    public String getAuthMethodName() {
+        return "athenz";
+    }
+
+    @Override
+    public String authenticate(AuthenticationDataSource authData) throws AuthenticationException {
+        SocketAddress clientAddress;
+        String roleToken;
+
+        if (authData.hasDataFromPeer()) {
+            clientAddress = authData.getPeerAddress();
+        } else {
+            throw new AuthenticationException("Authentication data source does not have a client address");
+        }
+
+        if (authData.hasDataFromCommand()) {
+            roleToken = authData.getCommandData();
+        } else if (authData.hasDataFromHttp()) {
+            roleToken = authData.getHttpHeader(AuthZpeClient.ZPE_TOKEN_HDR);
+        } else {
+            throw new AuthenticationException("Authentication data source does not have a role token");
+        }
+
+        if (roleToken == null) {
+            throw new AuthenticationException("Athenz token is null, can't authenticate");
+        }
+        if (roleToken.isEmpty()) {
+            throw new AuthenticationException("Athenz RoleToken is empty, Server is Using Athenz Authentication");
+        }
+        if (log.isDebugEnabled()) {
+            log.debug("Athenz RoleToken : [{}] received from Client: {}", roleToken, clientAddress);
+        }
+
+        RoleToken token = new RoleToken(roleToken);
+
+        if (!domainNameList.contains(token.getDomain())) {
+            throw new AuthenticationException(
+                    String.format("Athenz RoleToken Domain mismatch, Expected: %s, Found: %s", domainNameList.toString(), token.getDomain()));
+        }
+
+        // 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");
+            }
+
+            if (token.validate(ztsPublicKey, allowedOffset, null)) {
+                log.info("Athenz Role Token : {}, Authorized for Client: {}", roleToken, clientAddress);
+                return token.getPrincipal();
+            } else {
+                throw new AuthenticationException(
+                        String.format("Athenz Role Token Not Authorized from Client: %s", clientAddress));
+            }
+        }
+    }
+
+    @Override
+    public void close() throws IOException {
+    }
+
+    private static final Logger log = LoggerFactory.getLogger(AuthenticationProviderAthenz.class);
+}

pulsar-client-auth-athenz/pom.xml

@@ -0,0 +1,47 @@
+<!--
+
+    Copyright 2016 Yahoo Inc.
+
+    Licensed 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>com.yahoo.pulsar</groupId>
+    <artifactId>pulsar</artifactId>
+    <version>1.17-SNAPSHOT</version>
+    <relativePath>..</relativePath>
+  </parent>
+
+  <artifactId>pulsar-client-auth-athenz</artifactId>
+  <packaging>jar</packaging>
+  <description>Athenz authentication plugin for java client</description>
+
+  <dependencies>
+
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>pulsar-client</artifactId>
+      <version>${project.parent.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>com.yahoo.athenz</groupId>
+      <artifactId>zts_java_client</artifactId>
+    </dependency>
+
+  </dependencies>
+</project>

pulsar-client-auth-athenz/src/main/java/com/yahoo/pulsar/client/impl/auth/AuthenticationAthenz.java

@@ -0,0 +1,107 @@
+/**
+ * Copyright 2016 Yahoo Inc.
+ *
+ * Licensed 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 com.yahoo.pulsar.client.impl.auth;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.security.PrivateKey;
+
+import com.yahoo.athenz.zts.RoleToken;
+import com.yahoo.athenz.zts.ZTSClient;
+import com.yahoo.athenz.auth.ServiceIdentityProvider;
+import com.yahoo.athenz.auth.impl.SimpleServiceIdentityProvider;
+import com.yahoo.athenz.auth.util.Crypto;
+import com.yahoo.pulsar.client.api.Authentication;
+import com.yahoo.pulsar.client.api.AuthenticationDataProvider;
+import com.yahoo.pulsar.client.api.PulsarClientException;
+import com.yahoo.pulsar.client.api.PulsarClientException.GettingAuthenticationDataException;
+
+public class AuthenticationAthenz implements Authentication {
+
+    private transient ZTSClient ztsClient = null;
+    private String tenantDomain;
+    private String tenantService;
+    private String providerDomain;
+    private String privateKeyPath;
+    private String keyId = "0";
+    private long cachedRoleTokenTimestamp;
+    private String roleToken;
+    private final int minValidity = 2 * 60 * 60; // athenz will only give this token if it's at least valid for 2hrs
+    private final int maxValidity = 24 * 60 * 60; // token has upto 24 hours validity
+    private final int cacheDurationInHour = 1; // we will cache role token for an hour then ask athenz lib again
+
+    public AuthenticationAthenz() {
+    }
+
+    @Override
+    public String getAuthMethodName() {
+        return "athenz";
+    }
+
+    @Override
+    synchronized public AuthenticationDataProvider getAuthData() throws PulsarClientException {
+        if (cachedRoleTokenIsValid()) {
+            return new AuthenticationDataAthenz(roleToken, getZtsClient().getHeader());
+        }
+        try {
+            // the following would set up the API call that requests tokens from the server
+            // that can only be used if they are 10 minutes from expiration and last twenty four hours
+            RoleToken token = getZtsClient().getRoleToken(providerDomain, null, minValidity, maxValidity, false);
+            roleToken = token.getToken();
+            cachedRoleTokenTimestamp = System.nanoTime();
+            return new AuthenticationDataAthenz(roleToken, getZtsClient().getHeader());
+        } catch (Throwable t) {
+            throw new GettingAuthenticationDataException(t);
+        }
+    }
+
+    private boolean cachedRoleTokenIsValid() {
+        if (roleToken == null) {
+            return false;
+        }
+        // Ensure we refresh the Athenz role token every hour to avoid using an expired role token
+        return (System.nanoTime() - cachedRoleTokenTimestamp) < TimeUnit.HOURS.toNanos(cacheDurationInHour);
+    }
+
+    @Override
+    public void configure(Map<String, String> authParams) {
+        this.tenantDomain = authParams.get("tenant_domain");
+        this.tenantService = authParams.get("tenant_service");
+        this.providerDomain = authParams.get("provider_domain");
+        this.privateKeyPath = authParams.get("private_key_path");
+        this.keyId = authParams.getOrDefault("key_id", "0");
+    }
+
+    @Override
+    public void start() throws PulsarClientException {
+    }
+
+    @Override
+    public void close() throws IOException {
+    }
+
+    ZTSClient getZtsClient() {
+        if (ztsClient == null) {
+            PrivateKey privateKey = Crypto.loadPrivateKey(new File(privateKeyPath));
+            ServiceIdentityProvider siaProvider = new SimpleServiceIdentityProvider(tenantDomain, tenantService,
+                    privateKey, keyId);
+            ztsClient = new ZTSClient(null, tenantDomain, tenantService, siaProvider);
+        }
+        return ztsClient;
+    }
+}