Add authorization support on function apis (apache#2213)

* Add authorization support on function apis

* fix authorization enable check

conf/functions_worker.yml

@@ -45,3 +45,11 @@ initialBrokerReconnectMaxRetries: 60
 assignmentWriteMaxRetries: 60
 instanceLivenessCheckFreqMs: 30000
 metricsSamplingPeriodSec: 60
+# Enforce authentication
+authenticationEnabled: false
+# Enforce authorization on accessing functions api
+authorizationEnabled: false
+# Set of autentication provider name list, which is a list of class names
+authenticationProviders: 
+# Set of role names that are treated as "super-user", meaning they will be able to access any admin-api
+superUserRoles: 

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

@@ -82,8 +82,8 @@ public Response registerFunction(final @PathParam("tenant") String tenant,
                                      final @FormDataParam("url") String functionPkgUrl,
                                      final @FormDataParam("functionDetails") String functionDetailsJson) {
 
-        return functions.registerFunction(
-            tenant, namespace, functionName, uploadedInputStream, fileDetail, functionPkgUrl, functionDetailsJson);
+        return functions.registerFunction(tenant, namespace, functionName, uploadedInputStream, fileDetail,
+                functionPkgUrl, functionDetailsJson, clientAppId());
     }
 
     @PUT
@@ -103,8 +103,8 @@ public Response updateFunction(final @PathParam("tenant") String tenant,
                                    final @FormDataParam("url") String functionPkgUrl,
                                    final @FormDataParam("functionDetails") String functionDetailsJson) {
 
-        return functions.updateFunction(
-            tenant, namespace, functionName, uploadedInputStream, fileDetail, functionPkgUrl, functionDetailsJson);
+        return functions.updateFunction(tenant, namespace, functionName, uploadedInputStream, fileDetail,
+                functionPkgUrl, functionDetailsJson, clientAppId());
 
     }
 
@@ -122,8 +122,7 @@ public Response updateFunction(final @PathParam("tenant") String tenant,
     public Response deregisterFunction(final @PathParam("tenant") String tenant,
                                        final @PathParam("namespace") String namespace,
                                        final @PathParam("functionName") String functionName) {
-        return functions.deregisterFunction(
-            tenant, namespace, functionName);
+        return functions.deregisterFunction(tenant, namespace, functionName, clientAppId());
     }
 
     @GET

pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java

@@ -77,6 +77,7 @@
 import org.testng.Assert;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
 import com.google.common.collect.Maps;
@@ -99,7 +100,6 @@ public class PulsarSinkE2ETest {
     PulsarAdmin admin;
     PulsarClient pulsarClient;
     BrokerStats brokerStatsClient;
-    WorkerServer functionsWorkerServer;
     WorkerService functionsWorkerService;
     final String tenant = "external-repl-prop";
     String pulsarFunctionsNamespace = tenant + "/use/pulsar-function-admin";
@@ -120,6 +120,11 @@ public class PulsarSinkE2ETest {
 
     private static final Logger log = LoggerFactory.getLogger(PulsarSinkE2ETest.class);
 
+    @DataProvider(name = "validRoleName")
+    public Object[][] validRoleName() {
+        return new Object[][] { { Boolean.TRUE }, { Boolean.FALSE } };
+    }
+
     @BeforeMethod
     void setup(Method method) throws Exception {
 
@@ -187,6 +192,7 @@ && isNotBlank(workerConfig.getClientAuthenticationParameters())) {
         pulsarClient = clientBuilder.build();
 
         TenantInfo propAdmin = new TenantInfo();
+        propAdmin.getAdminRoles().add("superUser");
         propAdmin.setAllowedClusters(Sets.newHashSet(Lists.newArrayList("use")));
         admin.tenants().updateTenant(tenant, propAdmin);
 
@@ -231,6 +237,9 @@ private WorkerService createPulsarFunctionWorker(ServiceConfiguration config) {
         workerConfig.setUseTls(true);
         workerConfig.setTlsAllowInsecureConnection(true);
         workerConfig.setTlsTrustCertsFilePath(TLS_CLIENT_CERT_FILE_PATH);
+
+        workerConfig.setAuthenticationEnabled(true);
+        workerConfig.setAuthorizationEnabled(true);
 
         return new WorkerService(workerConfig);
     }
@@ -416,4 +425,34 @@ protected FunctionDetails createSinkConfig(String jarFile, String tenant, String
 
         return functionDetailsBuilder.build();
     }
+
+    @Test(dataProvider = "validRoleName")
+    public void testAuthorization(boolean validRoleName) throws Exception {
+
+        final String namespacePortion = "io";
+        final String replNamespace = tenant + "/" + namespacePortion;
+        final String sinkTopic = "persistent://" + replNamespace + "/output";
+        final String functionName = "PulsarSink-test";
+        final String subscriptionName = "test-sub";
+        admin.namespaces().createNamespace(replNamespace);
+        Set<String> clusters = Sets.newHashSet(Lists.newArrayList("use"));
+        admin.namespaces().setNamespaceReplicationClusters(replNamespace, clusters);
+
+        String roleName = validRoleName ? "superUser" : "invalid";
+        TenantInfo propAdmin = new TenantInfo();
+        propAdmin.getAdminRoles().add(roleName);
+        propAdmin.setAllowedClusters(Sets.newHashSet(Lists.newArrayList("use")));
+        admin.tenants().updateTenant(tenant, propAdmin);
+
+        String jarFilePathUrl = Utils.FILE + ":"
+                + PulsarSink.class.getProtectionDomain().getCodeSource().getLocation().getPath();
+        FunctionDetails functionDetails = createSinkConfig(jarFilePathUrl, tenant, namespacePortion, functionName,
+                sinkTopic, subscriptionName);
+        try {
+            admin.functions().createFunctionWithUrl(functionDetails, jarFilePathUrl);
+            Assert.assertTrue(validRoleName);
+        } catch (org.apache.pulsar.client.admin.PulsarAdminException.NotAuthorizedException ne) {
+            Assert.assertFalse(validRoleName);
+        }
+    }
 }

pulsar-functions/worker/pom.xml

@@ -34,6 +34,12 @@
 
   <dependencies>
 
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>pulsar-broker-common</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
     <dependency>
       <groupId>${project.groupId}</groupId>
       <artifactId>pulsar-functions-runtime</artifactId>

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java

@@ -20,14 +20,18 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import com.google.common.collect.Sets;
 
 import java.io.File;
 import java.io.IOException;
 import java.io.Serializable;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.util.Properties;
+import java.util.Set;
 
 import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.common.configuration.PulsarConfiguration;
 
 import lombok.Data;
 import lombok.EqualsAndHashCode;
@@ -42,7 +46,7 @@
 @EqualsAndHashCode
 @ToString
 @Accessors(chain = true)
-public class WorkerConfig implements Serializable {
+public class WorkerConfig implements Serializable, PulsarConfiguration {
 
     private static final long serialVersionUID = 1L;
 
@@ -74,6 +78,17 @@ public class WorkerConfig implements Serializable {
     private boolean tlsAllowInsecureConnection = false;
     private boolean tlsHostnameVerificationEnable = false;
     private int metricsSamplingPeriodSec = 60;
+    // Enforce authentication
+    private boolean authenticationEnabled = false;
+    // Autentication provider name list, which is a list of class names
+    private Set<String> authenticationProviders = Sets.newTreeSet();
+    // Enforce authorization on accessing functions admin-api
+    private boolean authorizationEnabled = false;
+    // Role names that are treated as "super-user", meaning they will be able to access any admin-api
+    private Set<String> superUserRoles = Sets.newTreeSet();
+
+    private Properties properties = new Properties();
+
 
     @Data
     @Setter
@@ -135,4 +150,9 @@ public static String unsafeLocalhostResolve() {
             throw new IllegalStateException("Failed to resolve localhost name.", ex);
         }
     }
+
+    @Override
+    public void setProperties(Properties properties) {
+        this.properties = properties;
+    }
 }

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java

@@ -35,9 +35,12 @@
 import org.apache.distributedlog.DistributedLogConfiguration;
 import org.apache.distributedlog.api.namespace.Namespace;
 import org.apache.distributedlog.api.namespace.NamespaceBuilder;
+import org.apache.pulsar.broker.authentication.AuthenticationService;
+import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.api.ClientBuilder;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.common.configuration.PulsarConfigurationLoader;
 
 /**
  * A service component contains everything to run a worker except rest server.
@@ -57,8 +60,9 @@ public class WorkerService {
     private SchedulerManager schedulerManager;
     private boolean isInitialized = false;
     private final ScheduledExecutorService statsUpdater;
-
+    private AuthenticationService authenticationService;
     private ConnectorsManager connectorsManager;
+    private PulsarAdmin admin;
 
     public WorkerService(WorkerConfig workerConfig) {
         this.workerConfig = workerConfig;
@@ -68,6 +72,11 @@ public WorkerService(WorkerConfig workerConfig) {
 
     public void start(URI dlogUri) throws InterruptedException {
         log.info("Starting worker {}...", workerConfig.getWorkerId());
+
+        this.admin = Utils.getPulsarAdminClient(workerConfig.getPulsarWebServiceUrl(),
+                workerConfig.getClientAuthenticationPlugin(), workerConfig.getClientAuthenticationParameters(),
+                workerConfig.getTlsTrustCertsFilePath(), workerConfig.isTlsAllowInsecureConnection());
+
         try {
             log.info("Worker Configs: {}", new ObjectMapper().writerWithDefaultPrettyPrinter()
                     .writeValueAsString(workerConfig));
@@ -128,6 +137,8 @@ && isNotBlank(workerConfig.getClientAuthenticationParameters())) {
 
             // initialize function metadata manager
             this.functionMetaDataManager.initialize();
+
+            authenticationService = new AuthenticationService(PulsarConfigurationLoader.convertFrom(workerConfig));
 
             // Starting cluster services
             log.info("Start cluster services...");
@@ -200,6 +211,10 @@ public void stop() {
         if (null != schedulerManager) {
             schedulerManager.close();
         }
+
+        if (null != this.admin) {
+            this.admin.close();
+        }
     }
 
 }

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/FunctionApiResource.java

@@ -18,9 +18,13 @@
  */
 package org.apache.pulsar.functions.worker.rest;
 
+import java.util.Optional;
 import java.util.function.Supplier;
 import javax.servlet.ServletContext;
+import javax.servlet.http.HttpServletRequest;
 import javax.ws.rs.core.Context;
+
+import org.apache.pulsar.broker.web.AuthenticationFilter;
 import org.apache.pulsar.functions.worker.WorkerService;
 import org.apache.pulsar.functions.worker.rest.api.FunctionsImpl;
 
@@ -32,6 +36,8 @@ public class FunctionApiResource implements Supplier<WorkerService> {
     private WorkerService workerService;
     @Context
     protected ServletContext servletContext;
+    @Context
+    protected HttpServletRequest httpRequest;
 
     public FunctionApiResource() {
         this.functions = new FunctionsImpl(this);
@@ -44,4 +50,10 @@ public synchronized WorkerService get() {
         }
         return this.workerService;
     }
+
+    public String clientAppId() {
+        return httpRequest != null
+                ? (String) httpRequest.getAttribute(AuthenticationFilter.AuthenticatedRoleAttributeName)
+                : null;
+    }
 }

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/WorkerServer.java

@@ -19,18 +19,27 @@
 package org.apache.pulsar.functions.worker.rest;
 
 import java.util.ArrayList;
+import java.util.EnumSet;
 import java.util.List;
+import java.util.Map;
+
+import javax.servlet.DispatcherType;
+
 import lombok.extern.slf4j.Slf4j;
+
+import org.apache.pulsar.broker.web.AuthenticationFilter;
 import org.apache.pulsar.functions.worker.WorkerConfig;
 import org.apache.pulsar.functions.worker.WorkerService;
 import org.eclipse.jetty.server.Handler;
 import org.eclipse.jetty.server.Server;
 
 import java.net.BindException;
 import java.net.URI;
+
 import org.eclipse.jetty.server.handler.ContextHandlerCollection;
 import org.eclipse.jetty.server.handler.DefaultHandler;
 import org.eclipse.jetty.server.handler.HandlerCollection;
+import org.eclipse.jetty.servlet.FilterHolder;
 import org.eclipse.jetty.servlet.ServletContextHandler;
 import org.eclipse.jetty.servlet.ServletHolder;
 import org.glassfish.jersey.server.ResourceConfig;
@@ -41,6 +50,7 @@ public class WorkerServer implements Runnable {
 
     private final WorkerConfig workerConfig;
     private final WorkerService workerService;
+    private static final String MATCH_ALL = "/*";
 
     private static String getErrorMessage(Server server, int port, Exception ex) {
         if (ex instanceof BindException) {
@@ -106,7 +116,12 @@ public static ServletContextHandler newServletContextHandler(String contextPath,
         final ServletHolder apiServlet =
                 new ServletHolder(new ServletContainer(config));
         contextHandler.addServlet(apiServlet, "/*");
+        if (workerService.getWorkerConfig().isAuthenticationEnabled()) {
+            FilterHolder filter = new FilterHolder(new AuthenticationFilter(workerService.getAuthenticationService()));
+            contextHandler.addFilter(filter, MATCH_ALL, EnumSet.allOf(DispatcherType.class));
+        }
 
         return contextHandler;
     }
+
 }