[fix][broker] support missing cluster level fine-granted permissions (#23675)

(cherry picked from commit 1c1a5cc)

Author: mattisonchao

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

@@ -33,6 +33,7 @@
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.policies.data.AuthAction;
 import org.apache.pulsar.common.policies.data.BrokerOperation;
+import org.apache.pulsar.common.policies.data.ClusterOperation;
 import org.apache.pulsar.common.policies.data.NamespaceOperation;
 import org.apache.pulsar.common.policies.data.PolicyName;
 import org.apache.pulsar.common.policies.data.PolicyOperation;
@@ -393,4 +394,23 @@ default CompletableFuture<Boolean> allowBrokerOperationAsync(String clusterName,
         return FutureUtil.failedFuture(
                 new UnsupportedOperationException("allowBrokerOperationAsync is not supported yet."));
     }
+
+
+    default CompletableFuture<Boolean> allowClusterOperationAsync(String clusterName,
+                                                                  ClusterOperation clusterOperation,
+                                                                  String role,
+                                                                  AuthenticationDataSource authData) {
+        return FutureUtil.failedFuture(
+                new UnsupportedOperationException("allowClusterOperationAsync is not supported yet."));
+    }
+
+    default CompletableFuture<Boolean> allowClusterPolicyOperationAsync(String clusterName,
+                                                                        String role,
+                                                                        PolicyName policy,
+                                                                        PolicyOperation operation,
+                                                                        AuthenticationDataSource authData) {
+        return FutureUtil.failedFuture(
+                new IllegalStateException("ClusterPolicyOperation [" + policy.name() + "/" + operation.name() + "] "
+                                          + "is not supported by the Authorization provider you are using."));
+    }
 }

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

@@ -39,6 +39,7 @@
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.policies.data.AuthAction;
 import org.apache.pulsar.common.policies.data.BrokerOperation;
+import org.apache.pulsar.common.policies.data.ClusterOperation;
 import org.apache.pulsar.common.policies.data.NamespaceOperation;
 import org.apache.pulsar.common.policies.data.PolicyName;
 import org.apache.pulsar.common.policies.data.PolicyOperation;
@@ -567,6 +568,50 @@ public CompletableFuture<Boolean> allowBrokerOperationAsync(String clusterName,
         }
     }
 
+    public CompletableFuture<Boolean> allowClusterOperationAsync(String clusterName,
+                                                                ClusterOperation clusterOperation,
+                                                                String originalRole,
+                                                                String role,
+                                                                AuthenticationDataSource authData) {
+        if (!isValidOriginalPrincipal(role, originalRole, authData)) {
+            return CompletableFuture.completedFuture(false);
+        }
+
+        if (isProxyRole(role)) {
+            final var isRoleAuthorizedFuture = provider.allowClusterOperationAsync(clusterName,
+                    clusterOperation, role, authData);
+            final var isOriginalAuthorizedFuture =  provider.allowClusterOperationAsync(clusterName,
+                    clusterOperation, originalRole, authData);
+            return isRoleAuthorizedFuture.thenCombine(isOriginalAuthorizedFuture,
+                    (isRoleAuthorized, isOriginalAuthorized) -> isRoleAuthorized && isOriginalAuthorized);
+        } else {
+            return provider.allowClusterOperationAsync(clusterName, clusterOperation, role, authData);
+        }
+    }
+
+    public CompletableFuture<Boolean> allowClusterPolicyOperationAsync(String clusterName,
+                                                                       PolicyName policy,
+                                                                       PolicyOperation operation,
+                                                                 String originalRole,
+                                                                 String role,
+                                                                 AuthenticationDataSource authData) {
+        if (!isValidOriginalPrincipal(role, originalRole, authData)) {
+            return CompletableFuture.completedFuture(false);
+        }
+
+        if (isProxyRole(role)) {
+            final var isRoleAuthorizedFuture = provider.allowClusterPolicyOperationAsync(clusterName, role,
+                    policy, operation, authData);
+            final var isOriginalAuthorizedFuture =  provider.allowClusterPolicyOperationAsync(clusterName, originalRole,
+                    policy, operation, authData);
+            return isRoleAuthorizedFuture.thenCombine(isOriginalAuthorizedFuture,
+                    (isRoleAuthorized, isOriginalAuthorized) -> isRoleAuthorized && isOriginalAuthorized);
+        } else {
+            return provider.allowClusterPolicyOperationAsync(clusterName, role, policy, operation, authData);
+        }
+    }
+
+
     /**
      * @deprecated - will be removed after 2.12. Use async variant.
      */

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

@@ -41,6 +41,7 @@
 import org.apache.pulsar.common.policies.data.AuthAction;
 import org.apache.pulsar.common.policies.data.AuthPolicies;
 import org.apache.pulsar.common.policies.data.BrokerOperation;
+import org.apache.pulsar.common.policies.data.ClusterOperation;
 import org.apache.pulsar.common.policies.data.NamespaceOperation;
 import org.apache.pulsar.common.policies.data.PolicyName;
 import org.apache.pulsar.common.policies.data.PolicyOperation;
@@ -860,4 +861,18 @@ public CompletableFuture<Map<String, Set<AuthAction>>> getPermissionsAsync(Names
                     });
         });
     }
+
+    @Override
+    public CompletableFuture<Boolean> allowClusterOperationAsync(String clusterName, ClusterOperation clusterOperation,
+                                                                 String role, AuthenticationDataSource authData) {
+        return isSuperUser(role, authData, conf);
+    }
+
+    @Override
+    public CompletableFuture<Boolean> allowClusterPolicyOperationAsync(String clusterName, String role,
+                                                                       PolicyName policy,
+                                                                       PolicyOperation operation,
+                                                                       AuthenticationDataSource authData) {
+        return isSuperUser(role, authData, conf);
+    }
 }