[Issue 5720][authorization provider] Add granularity (#6428)

Fixes #5720

### Motivation

Provide "real" authz abilities to pulsar resources.

### Modifications

Add stuff to `AuthorizationProvider` interface, and use them on every pulsar resource management auth (tenant, namespace, topics, functions, connectors, ...)

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

@@ -28,9 +28,22 @@ public class AuthenticationDataCommand implements AuthenticationDataSource {
     protected final String authData;
     protected final SocketAddress remoteAddress;
     protected final SSLSession sslSession;
+    protected String subscription;
 
     public AuthenticationDataCommand(String authData) {
-        this(authData, null, null);
+        this(authData, null, null, null);
+    }
+
+    public AuthenticationDataCommand(String authData, String subscription) {
+        this(authData, null, null, subscription);
+    }
+
+    public AuthenticationDataCommand(String authData, SocketAddress remoteAddress, SSLSession sslSession,
+                                     String subscription) {
+        this.authData = authData;
+        this.remoteAddress = remoteAddress;
+        this.sslSession = sslSession;
+        this.subscription = subscription;
     }
 
     public AuthenticationDataCommand(String authData, SocketAddress remoteAddress, SSLSession sslSession) {
@@ -85,4 +98,21 @@ public Certificate[] getTlsCertificates() {
         }
     }
 
+    /*
+     * Subscription
+     */
+    @Override
+    public boolean hasSubscription() {
+        return this.subscription != null;
+    }
+
+    @Override
+    public void setSubscription(String subscription) {
+        this.subscription = subscription;
+    }
+
+    @Override
+    public String getSubscription() {
+        return subscription;
+    }
 }

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

@@ -127,4 +127,27 @@ default boolean hasDataFromPeer() {
     default SocketAddress getPeerAddress() {
         return null;
     }
+
+    /**
+     * Check if subscription is defined available.
+     *
+     * @return true if this authentication data contain subscription
+     */
+    default boolean hasSubscription() {
+        return false;
+    }
+
+    /**
+     * Subscription name can be necessary for consumption
+     *
+     * @return a <code>String</code> containing the subscription name
+     */
+    default String getSubscription() { return null; }
+
+    /**
+     * Subscription name can be necessary for consumption
+     *
+     * @return a <code>String</code> containing the subscription name
+     */
+    default void setSubscription(String subscription) { };
 }

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

@@ -22,14 +22,22 @@
 import java.io.IOException;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 
 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.TopicName;
 import org.apache.pulsar.common.naming.NamespaceName;
 import org.apache.pulsar.common.policies.data.AuthAction;
+import org.apache.pulsar.common.policies.data.PolicyName;
+import org.apache.pulsar.common.policies.data.PolicyOperation;
 import org.apache.pulsar.common.policies.data.TenantInfo;
+import org.apache.pulsar.common.policies.data.NamespaceOperation;
+import org.apache.pulsar.common.policies.data.TenantOperation;
+import org.apache.pulsar.common.policies.data.TopicOperation;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.common.util.RestException;
 
 /**
  * Provider of authorization mechanism
@@ -186,4 +194,114 @@ CompletableFuture<Void> revokeSubscriptionPermissionAsync(NamespaceName namespac
     CompletableFuture<Void> grantPermissionAsync(TopicName topicName, Set<AuthAction> actions, String role,
             String authDataJson);
 
+    /**
+     * Grant authorization-action permission on a tenant to the given client
+     * @param tenantName
+     * @param originalRole role not overriden by proxy role if request do pass through proxy
+     * @param role originalRole | proxyRole if the request didn't pass through proxy
+     * @param operation
+     * @param authData
+     * @return CompletableFuture<Boolean>
+     */
+    default CompletableFuture<Boolean> allowTenantOperationAsync(String tenantName, String originalRole, String role,
+                                                            TenantOperation operation,
+                                                            AuthenticationDataSource authData) {
+        return FutureUtil.failedFuture(new IllegalStateException(
+                String.format("allowTenantOperation(%s) on tenant %s is not supported by the Authorization" +
+                                " provider you are using.",
+                        operation.toString(), tenantName)));
+    }
+
+    default Boolean allowTenantOperation(String tenantName, String originalRole, String role, TenantOperation operation,
+                                      AuthenticationDataSource authData) {
+        try {
+            return allowTenantOperationAsync(tenantName, originalRole, role, operation, authData).get();
+        } catch (InterruptedException e) {
+            throw new RestException(e);
+        } catch (ExecutionException e) {
+            throw new RestException(e.getCause());
+        }
+    }
+
+    /**
+     * Grant authorization-action permission on a namespace to the given client
+     * @param namespaceName
+     * @param originalRole role not overriden by proxy role if request do pass through proxy
+     * @param role originalRole | proxyRole if the request didn't pass through proxy
+     * @param operation
+     * @param authData
+     * @return CompletableFuture<Boolean>
+     */
+    default CompletableFuture<Boolean> allowNamespaceOperationAsync(NamespaceName namespaceName, String originalRole,
+                                                                 String role, NamespaceOperation operation,
+                                                                 AuthenticationDataSource authData) {
+        return FutureUtil.failedFuture(
+            new IllegalStateException("NamespaceOperation is not supported by the Authorization provider you are using."));
+    }
+
+    default Boolean allowNamespaceOperation(NamespaceName namespaceName, String originalRole, String role,
+                                         NamespaceOperation operation, AuthenticationDataSource authData) {
+        try {
+            return allowNamespaceOperationAsync(namespaceName, originalRole, role, operation, authData).get();
+        } catch (InterruptedException e) {
+            throw new RestException(e);
+        } catch (ExecutionException e) {
+            throw new RestException(e.getCause());
+        }
+    }
+
+    /**
+     * Grant authorization-action permission on a namespace to the given client
+     * @param namespaceName
+     * @param originalRole role not overriden by proxy role if request do pass through proxy
+     * @param role originalRole | proxyRole if the request didn't pass through proxy
+     * @param operation
+     * @param authData
+     * @return CompletableFuture<Boolean>
+     */
+    default CompletableFuture<Boolean> allowNamespacePolicyOperationAsync(NamespaceName namespaceName, PolicyName policy,
+                                                                          PolicyOperation operation, String originalRole,
+                                                                          String role, AuthenticationDataSource authData) {
+        return FutureUtil.failedFuture(
+                new IllegalStateException("NamespacePolicyOperation is not supported by the Authorization provider you are using."));
+    }
+
+    default Boolean allowNamespacePolicyOperation(NamespaceName namespaceName, PolicyName policy, PolicyOperation operation,
+                                                  String originalRole, String role, AuthenticationDataSource authData) {
+        try {
+            return allowNamespacePolicyOperationAsync(namespaceName, policy, operation, originalRole, role, authData).get();
+        } catch (InterruptedException e) {
+            throw new RestException(e);
+        } catch (ExecutionException e) {
+            throw new RestException(e.getCause());
+        }
+    }
+
+
+    /**
+     * Grant authorization-action permission on a topic to the given client
+     * @param topic
+     * @param originalRole role not overriden by proxy role if request do pass through proxy
+     * @param role originalRole | proxyRole if the request didn't pass through proxy
+     * @param operation
+     * @param authData
+     * @return CompletableFuture<Boolean>
+     */
+    default CompletableFuture<Boolean> allowTopicOperationAsync(TopicName topic, String originalRole, String role,
+                                                             TopicOperation operation,
+                                                             AuthenticationDataSource authData) {
+        return FutureUtil.failedFuture(
+            new IllegalStateException("TopicOperation is not supported by the Authorization provider you are using."));
+    }
+
+    default Boolean allowTopicOperation(TopicName topicName, String originalRole, String role, TopicOperation operation,
+                                     AuthenticationDataSource authData) {
+        try {
+            return allowTopicOperationAsync(topicName, originalRole, role, operation, authData).get();
+        } catch (InterruptedException e) {
+            throw new RestException(e);
+        } catch (ExecutionException e) {
+            throw new RestException(e.getCause());
+        }
+    }
 }