Add pluggable authorization mechanism (#1200)

* Add pluggable authorization service

fix: move FutureUtils to common and fix import

add grantpermission api

take default auth method

pass authData to authorization provider

keep single authorization provider

* fix rebase change

conf/broker.conf

@@ -203,6 +203,9 @@ authenticationProviders=
 # Enforce authorization
 authorizationEnabled=false
 
+# Authorization provider fully qualified class-name
+authorizationProvider=org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider
+
 # Allow wildcard matching in authorization
 # (wildcard matching only applicable if wildcard-char:
 # * presents at first or last position eg: *.pulsar.service, pulsar.service.*)

conf/discovery.conf

@@ -52,6 +52,9 @@ authenticationProviders=
 # Enforce authorization
 authorizationEnabled=false
 
+# Authorization provider name list, which is comma separated list of class names
+authorizationProviders=org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider
+
 # Role names that are treated as "super-user", meaning they will be able to do all admin
 # operations and publish/consume from all topics (comma-separated)
 superUserRoles=

conf/proxy.conf

@@ -49,6 +49,9 @@ authenticationProviders=
 # Enforce authorization
 authorizationEnabled=false
 
+# Authorization provider fully qualified class-name
+authorizationProvider=org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider
+
 # Authentication settings of the proxy itself. Used to connect to brokers
 brokerClientAuthenticationPlugin=
 brokerClientAuthenticationParameters=

conf/standalone.conf

@@ -169,6 +169,9 @@ authenticationProviders=
 # Enforce authorization
 authorizationEnabled=false
 
+# Authorization provider fully qualified class-name
+authorizationProvider=org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider
+
 # Allow wildcard matching in authorization
 # (wildcard matching only applicable if wildcard-char:
 # * presents at first or last position eg: *.pulsar.service, pulsar.service.*)

conf/websocket.conf

@@ -59,6 +59,9 @@ authenticationProviders=
 # Enforce authorization
 authorizationEnabled=false
 
+# Authorization provider fully qualified class-name
+authorizationProvider=org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider
+
 # Allow wildcard matching in authorization
 # (wildcard matching only applicable if wildcard-char:
 # * presents at first or last position eg: *.pulsar.service, pulsar.service.*)

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -24,6 +24,7 @@
 import java.util.Properties;
 import java.util.Set;
 
+import org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider;
 import org.apache.pulsar.common.configuration.FieldContext;
 import org.apache.pulsar.common.configuration.PulsarConfiguration;
 
@@ -188,6 +189,8 @@ public class ServiceConfiguration implements PulsarConfiguration {
 
     // Enforce authorization
     private boolean authorizationEnabled = false;
+    // Authorization provider fully qualified class-name
+    private String authorizationProvider = PulsarAuthorizationProvider.class.getName();
 
     // Role names that are treated as "super-user", meaning they will be able to
     // do all admin operations and publish/consume from all topics
@@ -783,6 +786,14 @@ public void setAuthenticationEnabled(boolean authenticationEnabled) {
         this.authenticationEnabled = authenticationEnabled;
     }
 
+    public String getAuthorizationProvider() {
+        return authorizationProvider;
+    }
+
+    public void setAuthorizationProvider(String authorizationProvider) {
+        this.authorizationProvider = authorizationProvider;
+    }
+
     public void setAuthenticationProviders(Set<String> providersClassNames) {
         authenticationProviders = providersClassNames;
     }

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java

@@ -0,0 +1,120 @@
+/**
+ * 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.broker.authorization;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
+import org.apache.pulsar.broker.cache.ConfigurationCacheService;
+import org.apache.pulsar.common.naming.DestinationName;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.policies.data.AuthAction;
+
+/**
+ * Provider of authorization mechanism
+ */
+public interface AuthorizationProvider extends Closeable {
+
+    /**
+     * Perform initialization for the authorization provider
+     *
+     * @param config
+     *            broker config object
+     * @param configCache
+     *            pulsar zk configuration cache service
+     * @throws IOException
+     *             if the initialization fails
+     */
+    void initialize(ServiceConfiguration conf, ConfigurationCacheService configCache) throws IOException;
+
+    /**
+     * Check if the specified role has permission to send messages to the specified fully qualified destination name.
+     *
+     * @param destination
+     *            the fully qualified destination name associated with the destination.
+     * @param role
+     *            the app id used to send messages to the destination.
+     */
+    CompletableFuture<Boolean> canProduceAsync(DestinationName destination, String role,
+            AuthenticationDataSource authenticationData);
+
+    /**
+     * Check if the specified role has permission to receive messages from the specified fully qualified destination
+     * name.
+     *
+     * @param destination
+     *            the fully qualified destination name associated with the destination.
+     * @param role
+     *            the app id used to receive messages from the destination.
+     * @param subscription
+     *            the subscription name defined by the client
+     */
+    CompletableFuture<Boolean> canConsumeAsync(DestinationName destination, String role,
+            AuthenticationDataSource authenticationData, String subscription);
+
+    /**
+     * Check whether the specified role can perform a lookup for the specified destination.
+     *
+     * For that the caller needs to have producer or consumer permission.
+     *
+     * @param destination
+     * @param role
+     * @return
+     * @throws Exception
+     */
+    CompletableFuture<Boolean> canLookupAsync(DestinationName destination, String role,
+            AuthenticationDataSource authenticationData);
+
+    /**
+     * 
+     * Grant authorization-action permission on a namespace to the given client
+     * 
+     * @param namespace
+     * @param actions
+     * @param role
+     * @param authDataJson
+     *            additional authdata in json format
+     * @return CompletableFuture
+     * @completesWith <br/>
+     *                IllegalArgumentException when namespace not found<br/>
+     *                IllegalStateException when failed to grant permission
+     */
+    CompletableFuture<Void> grantPermissionAsync(NamespaceName namespace, Set<AuthAction> actions, String role,
+            String authDataJson);
+
+    /**
+     * Grant authorization-action permission on a topic to the given client
+     * 
+     * @param topicname
+     * @param role
+     * @param authDataJson
+     *            additional authdata in json format
+     * @return CompletableFuture
+     * @completesWith <br/>
+     *                IllegalArgumentException when namespace not found<br/>
+     *                IllegalStateException when failed to grant permission
+     */
+    CompletableFuture<Void> grantPermissionAsync(DestinationName topicname, Set<AuthAction> actions, String role,
+            String authDataJson);
+
+}