From 3d8b4ca5c296a69f4d25688bbafc7a8f07543068 Mon Sep 17 00:00:00 2001
From: mattison chao <mattisonchao@gmail.com>
Date: Tue, 24 May 2022 15:03:20 +0800
Subject: [PATCH 1/2] [branch-2.9] [Authorization] Role with namespace produce
 authz can also get topics #13773

---
 .../PulsarAuthorizationProvider.java          | 34 +++++++++++++++++++
 .../AuthorizationProducerConsumerTest.java    |  5 +++
 2 files changed, 39 insertions(+)

diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
index d0884da7dc08f..95e2ce41dc9af 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
@@ -563,6 +563,40 @@ public CompletableFuture<Boolean> allowNamespaceOperationAsync(NamespaceName nam
                 });
     }
 
+    private CompletableFuture<Boolean> allowConsumeOrProduceOpsAsync(NamespaceName namespaceName,
+                                                                     String role,
+                                                                     AuthenticationDataSource authenticationData) {
+        CompletableFuture<Boolean> finalResult = new CompletableFuture<>();
+        allowTheSpecifiedActionOpsAsync(namespaceName, role, authenticationData, AuthAction.consume)
+                .whenComplete((consumeAuthorized, e) -> {
+                    if (e == null) {
+                        if (consumeAuthorized) {
+                            finalResult.complete(consumeAuthorized);
+                            return;
+                        }
+                    } else {
+                        if (log.isDebugEnabled()) {
+                            log.debug("Namespace [{}] Role [{}] exception occurred while trying to check Consume "
+                                    + "permission. {}", namespaceName, role, e.getCause());
+                        }
+                    }
+                    allowTheSpecifiedActionOpsAsync(namespaceName, role, authenticationData, AuthAction.produce)
+                            .whenComplete((produceAuthorized, ex) -> {
+                                if (ex == null) {
+                                    finalResult.complete(produceAuthorized);
+                                } else {
+                                    if (log.isDebugEnabled()) {
+                                        log.debug("Namespace [{}] Role [{}] exception occurred while trying to check "
+                                                + "Produce permission. {}", namespaceName, role, ex.getCause());
+                                    }
+                                    finalResult.completeExceptionally(ex.getCause());
+                                }
+                            });
+                });
+
+        return finalResult;
+    }
+
     @Override
     public CompletableFuture<Boolean> allowNamespacePolicyOperationAsync(NamespaceName namespaceName,
                                                                          PolicyName policy,
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java
index 62aa429436d80..dcfb16c92de61 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java
@@ -432,6 +432,11 @@ public void testClearBacklogPermission() throws Exception {
         assertEquals(sub1Admin.topics().getStats(topicName + "-partition-0").getSubscriptions()
                 .get(subscriptionName).getMsgBacklog(), 0);
 
+        superAdmin.namespaces().revokePermissionsOnNamespace(namespace, subscriptionRole);
+        superAdmin.namespaces().grantPermissionOnNamespace(namespace, subscriptionRole,
+                Sets.newHashSet(AuthAction.produce));
+        assertEquals(sub1Admin.topics().getPartitionedTopicList(namespace),
+                Lists.newArrayList(topicName));
         log.info("-- Exiting {} test --", methodName);
     }
 

From 0e81d46a9e3d8779ff086a965ff59d9f9b32e6fd Mon Sep 17 00:00:00 2001
From: mattison chao <mattisonchao@gmail.com>
Date: Tue, 24 May 2022 15:32:32 +0800
Subject: [PATCH 2/2] Add logic

---
 .../pulsar/broker/authorization/PulsarAuthorizationProvider.java | 1 +
 1 file changed, 1 insertion(+)

diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
index 95e2ce41dc9af..9aea1261cf216 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
@@ -542,6 +542,7 @@ public CompletableFuture<Boolean> allowNamespaceOperationAsync(NamespaceName nam
                                         namespaceName, role, authData, AuthAction.packages);
                             case GET_TOPIC:
                             case GET_TOPICS:
+                                return allowConsumeOrProduceOpsAsync(namespaceName, role, authData);
                             case UNSUBSCRIBE:
                             case CLEAR_BACKLOG:
                                 return allowTheSpecifiedActionOpsAsync(