Implementing authentication for Pulsar Functions

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

@@ -35,7 +35,7 @@
 public class AuthenticationProviderToken implements AuthenticationProvider {
 
     final static String HTTP_HEADER_NAME = "Authorization";
-    final static String HTTP_HEADER_VALUE_PREFIX = "Bearer ";
+    public final static String HTTP_HEADER_VALUE_PREFIX = "Bearer ";
 
     // When symmetric key is configured
     final static String CONF_TOKEN_SECRET_KEY = "tokenSecretKey";

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java

@@ -82,7 +82,7 @@ public void registerFunction(final @PathParam("tenant") String tenant,
                                  final @FormDataParam("functionConfig") String functionConfigJson) {
 
         functions.registerFunction(tenant, namespace, functionName, uploadedInputStream, fileDetail,
-            functionPkgUrl, null, functionConfigJson, clientAppId());
+            functionPkgUrl, null, functionConfigJson, clientAppId(), clientAuthData());
     }
 
     @PUT

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SinkBase.java

@@ -79,7 +79,7 @@ public void registerSink(final @PathParam("tenant") String tenant,
                              final @FormDataParam("sinkConfig") String sinkConfigJson) {
 
         sink.registerFunction(tenant, namespace, sinkName, uploadedInputStream, fileDetail,
-                functionPkgUrl, null, sinkConfigJson, clientAppId());
+                functionPkgUrl, null, sinkConfigJson, clientAppId(), clientAuthData());
     }
 
     @PUT

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SourceBase.java

@@ -79,7 +79,7 @@ public void registerSource(final @PathParam("tenant") String tenant,
                                final @FormDataParam("sourceConfig") String sourceConfigJson) {
 
         source.registerFunction(tenant, namespace, sourceName, uploadedInputStream, fileDetail,
-            functionPkgUrl, null, sourceConfigJson, clientAppId());
+            functionPkgUrl, null, sourceConfigJson, clientAppId(), clientAuthData());
     }
 
     @PUT

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceConfig.java

@@ -23,6 +23,7 @@
 import lombok.Getter;
 import lombok.Setter;
 import lombok.ToString;
+import org.apache.pulsar.functions.proto.Function;
 import org.apache.pulsar.functions.proto.Function.FunctionDetails;
 
 /**
@@ -40,6 +41,7 @@ public class InstanceConfig {
     private String functionVersion;
     private FunctionDetails functionDetails;
     private int maxBufferedTuples;
+    private Function.FunctionAuthenticationSpec functionAuthenticationSpec;
     private int port;
     private String clusterName;
 

pulsar-functions/proto/src/main/proto/Function.proto

@@ -141,6 +141,11 @@ message FunctionMetaData {
     uint64 version = 3;
     uint64 createTime = 4;
     map<int32, FunctionState> instanceStates = 5;
+    FunctionAuthenticationSpec functionAuthSpec = 6;
+}
+
+message FunctionAuthenticationSpec {
+    bytes data = 1;
 }
 
 message Instance {

pulsar-functions/runtime/pom.xml

@@ -63,6 +63,12 @@
       </exclusions>
     </dependency>
 
+      <dependency>
+          <groupId>org.apache.pulsar</groupId>
+          <artifactId>pulsar-broker-common</artifactId>
+          <version>${project.version}</version>
+      </dependency>
+
   </dependencies>
 
 </project>

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/auth/ClearTextFunctionTokenAuthProvider.java

@@ -0,0 +1,86 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.pulsar.functions.auth;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
+import org.apache.pulsar.client.impl.auth.AuthenticationToken;
+import org.apache.pulsar.functions.instance.AuthenticationConfig;
+import org.apache.pulsar.functions.proto.Function;
+
+import javax.naming.AuthenticationException;
+
+import static org.apache.pulsar.broker.authentication.AuthenticationProviderToken.HTTP_HEADER_VALUE_PREFIX;
+import static org.apache.pulsar.client.impl.auth.AuthenticationDataToken.HTTP_HEADER_NAME;
+
+public class ClearTextFunctionTokenAuthProvider implements FunctionAuthProvider {
+    @Override
+    public void configureAuthenticationConfig(AuthenticationConfig authConfig, FunctionAuthData functionAuthData) {
+        authConfig.setClientAuthenticationPlugin(AuthenticationToken.class.getName());
+        authConfig.setClientAuthenticationParameters("token://" + functionAuthData.getData());
+    }
+
+    @Override
+    public FunctionAuthData cacheAuthData(String tenant, String namespace, String name, AuthenticationDataSource authenticationDataSource) throws Exception {
+        String token = null;
+        try {
+            token = getToken(authenticationDataSource);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+
+        if (token != null) {
+            return FunctionAuthData.builder().data(token.getBytes()).build();
+        }
+        return null;
+    }
+
+    @Override
+    public void cleanUpAuthData(String tenant, String namespace, String name, FunctionAuthData functionAuthData) throws Exception {
+        //no-op
+    }
+
+    private String getToken(AuthenticationDataSource authData) throws AuthenticationException {
+        if (authData.hasDataFromCommand()) {
+            // Authenticate Pulsar binary connection
+            return authData.getCommandData();
+        } else if (authData.hasDataFromHttp()) {
+            // Authentication HTTP request. The format here should be compliant to RFC-6750
+            // (https://tools.ietf.org/html/rfc6750#section-2.1). Eg: Authorization: Bearer xxxxxxxxxxxxx
+            String httpHeaderValue = authData.getHttpHeader(HTTP_HEADER_NAME);
+            if (httpHeaderValue == null || !httpHeaderValue.startsWith(HTTP_HEADER_VALUE_PREFIX)) {
+                throw new AuthenticationException("Invalid HTTP Authorization header");
+            }
+
+            // Remove prefix
+            String token = httpHeaderValue.substring(HTTP_HEADER_VALUE_PREFIX.length());
+            return validateToken(token);
+        } else {
+            return null;
+        }
+    }
+
+    private String validateToken(final String token) throws AuthenticationException {
+        if (StringUtils.isNotBlank(token)) {
+            return token;
+        } else {
+            throw new AuthenticationException("Blank token found");
+        }
+    }
+}

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/auth/FunctionAuthData.java

@@ -0,0 +1,31 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.pulsar.functions.auth;
+
+import lombok.Builder;
+import lombok.Data;
+
+@Data
+@Builder
+/**
+ * A wrapper for authentication data for functions
+ */
+public class FunctionAuthData {
+    private byte[] data;
+}

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/auth/FunctionAuthProvider.java

@@ -0,0 +1,58 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.pulsar.functions.auth;
+
+import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
+import org.apache.pulsar.functions.instance.AuthenticationConfig;
+import org.apache.pulsar.functions.proto.Function;
+
+/**
+ * This is a generic interface that functions can use to cache and distribute appropriate authentication
+ * data that is needed to configure the runtime of functions to support appropriate authentication of function instances
+ */
+public interface FunctionAuthProvider {
+
+    /**
+     * Set authentication configs for function instance based on the data in FunctionAuthenticationSpec
+     * @param authConfig authentication configs passed to the function instance
+     * @param functionAuthData function authentication data that is provider specific
+     */
+    void configureAuthenticationConfig(AuthenticationConfig authConfig, FunctionAuthData functionAuthData);
+
+    /**
+     * Cache auth data in as part of function metadata for function that runtime may need to configure authentication
+     * @param tenant tenant that the function is running under
+     * @param namespace namespace that is the function is running under
+     * @param name name of the function
+     * @param authenticationDataSource auth data
+     * @return
+     * @throws Exception
+     */
+    FunctionAuthData cacheAuthData(String tenant, String namespace, String name, AuthenticationDataSource authenticationDataSource) throws Exception;
+
+    /**
+     * Clean up operation for auth when function is terminated
+     * @param tenant tenant that the function is running under
+     * @param namespace namespace that is the function is running under
+     * @param name name of the function
+     * @param functionAuthData function auth data
+     * @throws Exception
+     */
+    void cleanUpAuthData(String tenant, String namespace, String name, FunctionAuthData functionAuthData) throws Exception;
+}

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/auth/FunctionAuthUtils.java

@@ -0,0 +1,28 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.pulsar.functions.auth;
+
+import org.apache.pulsar.functions.proto.Function;
+
+public final class FunctionAuthUtils {
+
+    public static final FunctionAuthData getFunctionAuthData(Function.FunctionAuthenticationSpec functionAuthenticationSpec) {
+        return FunctionAuthData.builder().data(functionAuthenticationSpec.getData().toByteArray()).build();
+    }
+}

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/auth/KubernetesFunctionAuthProvider.java

@@ -0,0 +1,42 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.pulsar.functions.auth;
+
+import io.kubernetes.client.models.V1ServiceAccount;
+import io.kubernetes.client.models.V1StatefulSet;
+
+/**
+ * Kubernetes runtime specific functions authentication provider
+ */
+public interface KubernetesFunctionAuthProvider extends FunctionAuthProvider {
+
+    /**
+     * Configure service account spec generated for function based on function auth data
+     * @param serviceAccount service account spec for the function
+     * @param functionAuthData function auth data
+     */
+    void configureAuthDataKubernetesServiceAccount(V1ServiceAccount serviceAccount, FunctionAuthData functionAuthData);
+
+    /**
+     * Configure function statefulset spec based on function auth data
+     * @param statefulSet statefulset spec for function
+     * @param functionAuthData function auth data
+     */
+    void configureAuthDataStatefulSet(V1StatefulSet statefulSet, FunctionAuthData functionAuthData);
+}