Secretprovider interfaces and some default implementations (apache#2855)

* Added SecretProvider and SecretProviderConfigurator interfaces and some implementations

* Added appropriate documentation

* Added validation logic to config provider configurator

* Added more comments

* Took feedback into account

pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionConfig.java

@@ -82,7 +82,8 @@ public enum Runtime {
     // This is a map of secretName(aka how the secret is going to be
     // accessed in the function via context) to an object that
     // encapsulates how the secret is fetched by the underlying
-    // secrets provider
+    // secrets provider. The type of an value here can be found by the
+    // SecretProviderConfigurator.getSecretObjectType() method.
     private Map<String, Object> secrets;
     private Runtime runtime;
     private boolean autoAck;

pulsar-common/src/main/java/org/apache/pulsar/common/io/SinkConfig.java

@@ -58,7 +58,8 @@ public class SinkConfig {
     // This is a map of secretName(aka how the secret is going to be
     // accessed in the function via context) to an object that
     // encapsulates how the secret is fetched by the underlying
-    // secrets provider
+    // secrets provider. The type of an value here can be found by the
+    // SecretProviderConfigurator.getSecretObjectType() method.
     private Map<String, Object> secrets;
     private int parallelism = 1;
     private FunctionConfig.ProcessingGuarantees processingGuarantees;

pulsar-common/src/main/java/org/apache/pulsar/common/io/SourceConfig.java

@@ -49,7 +49,8 @@ public class SourceConfig {
     // This is a map of secretName(aka how the secret is going to be
     // accessed in the function via context) to an object that
     // encapsulates how the secret is fetched by the underlying
-    // secrets provider
+    // secrets provider. The type of an value here can be found by the
+    // SecretProviderConfigurator.getSecretObjectType() method.
     private Map<String, Object> secrets;
     private int parallelism = 1;
     private FunctionConfig.ProcessingGuarantees processingGuarantees;

pulsar-functions/pom.xml

@@ -43,6 +43,7 @@
     <module>runtime-shaded</module>
     <module>runtime-all</module>
     <module>worker</module>
+    <module>secrets</module>
   </modules>
 
   <dependencyManagement>

pulsar-functions/secrets/pom.xml

@@ -0,0 +1,54 @@
+<!--
+
+    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.
+
+-->
+<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>org.apache.pulsar</groupId>
+    <artifactId>pulsar-functions</artifactId>
+    <version>2.3.0-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>pulsar-functions-secrets</artifactId>
+  <name>Pulsar Functions :: Secrets</name>
+
+  <dependencies>
+  <dependency>
+    <groupId>io.kubernetes</groupId>
+    <artifactId>client-java</artifactId>
+    <version>2.0.0</version>
+    <scope>compile</scope>
+    <exclusions>
+      <exclusion>
+        <groupId>ch.qos.logback</groupId>
+        <artifactId>logback-classic</artifactId>
+      </exclusion>
+    </exclusions>
+  </dependency>
+
+    <dependency>
+      <groupId>org.apache.pulsar</groupId>
+      <artifactId>pulsar-functions-proto</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+  </dependencies>
+
+</project>

pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsprovider/ClearTextSecretsProvider.java

@@ -0,0 +1,38 @@
+/**
+ * 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.secretsprovider;
+
+/**
+ * This file defines a very basic clear text secrets provider which treats
+ * the secrets as being passed in cleartext.
+ */
+public class ClearTextSecretsProvider implements SecretsProvider {
+        /**
+     * Fetches a secret
+     * @return The actual secret
+     */
+    @Override
+    public String provideSecret(String secretName, Object pathToSecret) {
+        if (pathToSecret != null) {
+            return pathToSecret.toString();
+        } else {
+            return null;
+        }
+    }
+}

pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsprovider/EnvironmentBasedSecretsProvider.java

@@ -0,0 +1,35 @@
+/**
+ * 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.secretsprovider;
+
+/**
+ * This defines a very simple Secrets Provider that looks up environment variable
+ * thats named the same as secretName and fetches it.
+ */
+public class EnvironmentBasedSecretsProvider implements SecretsProvider {
+
+    /**
+     * Fetches a secret
+     * @return The actual secret
+     */
+    @Override
+    public String provideSecret(String secretName, Object pathToSecret) {
+        return System.getenv(secretName);
+    }
+}

pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsprovider/SecretsProvider.java

@@ -0,0 +1,40 @@
+/**
+ * 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.secretsprovider;
+
+import java.util.Map;
+
+/**
+ * This file defines the SecretsProvider interface. This interface is used by the function
+ * instances/containers to actually fetch the secrets. What SecretsProvider to use is
+ * decided by the SecretsProviderConfigurator
+ */
+public interface SecretsProvider {
+    /**
+     * Initialize the SecretsProvider
+     * @return
+     */
+    default void init(Map<String, String> config) {}
+
+    /**
+     * Fetches a secret
+     * @return The actual secret
+     */
+    String provideSecret(String secretName, Object pathToSecret);
+}

pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsproviderconfigurator/DefaultSecretsProviderConfigurator.java

@@ -0,0 +1,67 @@
+/**
+ * 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.secretsproviderconfigurator;
+
+import com.google.gson.reflect.TypeToken;
+import io.kubernetes.client.models.V1Container;
+import org.apache.pulsar.functions.proto.Function;
+import org.apache.pulsar.functions.secretsprovider.ClearTextSecretsProvider;
+
+import java.lang.reflect.Type;
+import java.util.Map;
+
+/**
+ * This is a barebones version of a secrets provider which wires in ClearTextSecretsProvider
+ * to the function instances/containers.
+ * While this is the default configurator, it is highly recommended that for real-security
+ * you use some alternate provider.
+ */
+public class DefaultSecretsProviderConfigurator implements SecretsProviderConfigurator {
+    @Override
+    public String getSecretsProviderClassName(Function.FunctionDetails functionDetails) {
+        switch (functionDetails.getRuntime()) {
+            case JAVA:
+                return ClearTextSecretsProvider.class.getName();
+            case PYTHON:
+                return "secretsprovider.ClearTextSecretsProvider";
+            default:
+                throw new RuntimeException("Unknwon runtime " + functionDetails.getRuntime());
+        }
+    }
+
+    @Override
+    public Map<String, String> getSecretsProviderConfig(Function.FunctionDetails functionDetails) {
+        return null;
+    }
+
+    @Override
+    public void configureKubernetesRuntimeSecretsProvider(V1Container container, Function.FunctionDetails functionDetails) {
+        // noop
+    }
+
+    @Override
+    public void configureProcessRuntimeSecretsProvider(ProcessBuilder processBuilder, Function.FunctionDetails functionDetails) {
+        // noop
+    }
+
+    @Override
+    public Type getSecretObjectType() {
+        return new TypeToken<String>() {}.getType();
+    }
+}

pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsproviderconfigurator/KubernetesSecretsProviderConfigurator.java

@@ -0,0 +1,113 @@
+/**
+ * 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.secretsproviderconfigurator;
+
+import com.google.gson.Gson;
+import com.google.gson.reflect.TypeToken;
+import io.kubernetes.client.models.V1Container;
+import io.kubernetes.client.models.V1EnvVar;
+import io.kubernetes.client.models.V1EnvVarSource;
+import io.kubernetes.client.models.V1SecretKeySelector;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.functions.proto.Function;
+import org.apache.pulsar.functions.secretsprovider.EnvironmentBasedSecretsProvider;
+
+import java.lang.reflect.Type;
+import java.util.Map;
+
+/**
+ * This file defines the SecretsProviderConfigurator that will be used by default for running in Kubernetes.
+ * As such this implementation is strictly when workers are configured to use kubernetes runtime.
+ * We use kubernetes in built secrets and bind them as environment variables within the function container
+ * to ensure that the secrets are availble to the function at runtime. Then we plug in the
+ * EnvironmentBasedSecretsConfig as the secrets provider who knows how to read these environment variables
+ */
+public class KubernetesSecretsProviderConfigurator implements SecretsProviderConfigurator {
+    private static String ID_KEY = "id";
+    private static String KEY_KEY = "key";
+    @Override
+    public String getSecretsProviderClassName(Function.FunctionDetails functionDetails) {
+        switch (functionDetails.getRuntime()) {
+            case JAVA:
+                return EnvironmentBasedSecretsProvider.class.getName();
+            case PYTHON:
+                return "secretsprovider.EnvironmentBasedSecretsProvider";
+            default:
+                throw new RuntimeException("Unknown function runtime " + functionDetails.getRuntime());
+        }
+    }
+
+    @Override
+    public Map<String, String> getSecretsProviderConfig(Function.FunctionDetails functionDetails) {
+        return null;
+    }
+
+    // Kubernetes secrets can be exposed as volume mounts or as environment variables in the pods. We are currently using the
+    // environment variables way. Essentially the secretName/secretPath is attached as secretRef to the environment variables
+    // of a pod and kubernetes magically makes the secret pointed to by this combination available as a env variable.
+    @Override
+    public void configureKubernetesRuntimeSecretsProvider(V1Container container, Function.FunctionDetails functionDetails) {
+        if (!StringUtils.isEmpty(functionDetails.getSecretsMap())) {
+            Type type = new TypeToken<Map<String, Object>>() {
+            }.getType();
+            Map<String, Object> secretsMap = new Gson().fromJson(functionDetails.getSecretsMap(), type);
+            for (Map.Entry<String, Object> entry : secretsMap.entrySet()) {
+                final V1EnvVar secretEnv = new V1EnvVar();
+                Map<String, String> kv = (Map<String, String>) entry.getValue();
+                secretEnv.name(entry.getKey())
+                        .valueFrom(new V1EnvVarSource()
+                                .secretKeyRef(new V1SecretKeySelector()
+                                        .name(kv.get(ID_KEY))
+                                        .key(kv.get(KEY_KEY))));
+                container.addEnvItem(secretEnv);
+            }
+        }
+    }
+
+    @Override
+    public void configureProcessRuntimeSecretsProvider(ProcessBuilder processBuilder, Function.FunctionDetails functionDetails) {
+        throw new RuntimeException("KubernetesSecretsProviderConfigurator should only be setup for Kubernetes Runtime");
+    }
+
+    @Override
+    public Type getSecretObjectType() {
+        return new TypeToken<Map<String, String>>() {}.getType();
+    }
+
+    // The secret object should be of type Map<String, String> and it should contain "id" and "key"
+    @Override
+    public void validateSecretMap(Map<String, Object> secretMap) {
+        for (Object object : secretMap.values()) {
+            if (object instanceof Map) {
+                Map<String, String> kubernetesSecret = (Map<String, String>) object;
+                if (kubernetesSecret.size() < 2) {
+                    throw new IllegalArgumentException("Kubernetes Secret should contain id and key");
+                }
+                if (!kubernetesSecret.containsKey(ID_KEY)) {
+                    throw new IllegalArgumentException("Kubernetes Secret should contain id information");
+                }
+                if (!kubernetesSecret.containsKey(KEY_KEY)) {
+                    throw new IllegalArgumentException("Kubernetes Secret should contain key information");
+                }
+            } else {
+                throw new IllegalArgumentException("Kubernetes Secret should be a Map containing id/key pairs");
+            }
+        }
+    }
+}