[improve][admin] Check if the topic existed before the permission ope…

…rations (apache#22742) (cherry picked from commit 71640f6)
datastax · Dec 30, 2024 · e2f62de · e2f62de
1 parent f964d29
commit e2f62de
Show file tree

Hide file tree

Showing 10 changed files with 49 additions and 22 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -222,6 +222,7 @@ protected CompletableFuture<List<String>> internalGetPartitionedTopicListAsync()
     protected CompletableFuture<Map<String, Set<AuthAction>>> internalGetPermissionsOnTopic() {
         // This operation should be reading from zookeeper and it should be allowed without having admin privileges
         return validateAdminAccessForTenantAsync(namespaceName.getTenant())
+                .thenCompose(__ -> internalCheckTopicExists(topicName))
                 .thenCompose(__ -> getAuthorizationService().getPermissionsAsync(topicName));
     }
 
@@ -273,9 +274,10 @@ protected void internalGrantPermissionsOnTopic(final AsyncResponse asyncResponse
                                                    Set<AuthAction> actions) {
         // This operation should be reading from zookeeper and it should be allowed without having admin privileges
         validateAdminAccessForTenantAsync(namespaceName.getTenant())
-                .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync().thenCompose(unused1 ->
-                        grantPermissionsAsync(topicName, role, actions)
-                                .thenAccept(unused -> asyncResponse.resume(Response.noContent().build()))))
+                .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync())
+                .thenCompose(__ -> internalCheckTopicExists(topicName))
+                .thenCompose(unused1 -> grantPermissionsAsync(topicName, role, actions))
+                .thenAccept(unused -> asyncResponse.resume(Response.noContent().build()))
                 .exceptionally(ex -> {
                     Throwable realCause = FutureUtil.unwrapCompletionException(ex);
                     log.error("[{}] Failed to get permissions for topic {}", clientAppId(), topicName, realCause);
@@ -287,7 +289,9 @@ protected void internalGrantPermissionsOnTopic(final AsyncResponse asyncResponse
     protected void internalRevokePermissionsOnTopic(AsyncResponse asyncResponse, String role) {
         // This operation should be reading from zookeeper and it should be allowed without having admin privileges
         validateAdminAccessForTenantAsync(namespaceName.getTenant())
-                .thenCompose(__ -> getPartitionedTopicMetadataAsync(topicName, true, false)
+                .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync())
+                .thenCompose(__ -> internalCheckTopicExists(topicName))
+                .thenCompose(unused1 -> getPartitionedTopicMetadataAsync(topicName, true, false)
                 .thenCompose(metadata -> {
                     int numPartitions = metadata.partitions;
                     CompletableFuture<Void> future = CompletableFuture.completedFuture(null);

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaWithAuthTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaWithAuthTest.java
@@ -115,6 +115,7 @@ public void testGetCreateDeleteSchema() throws Exception {
                 .serviceHttpUrl(brokerUrl != null ? brokerUrl.toString() : brokerUrlTls.toString())
                 .authentication(AuthenticationToken.class.getName(), PRODUCE_TOKEN)
                 .build();
+        admin.topics().createNonPartitionedTopic(topicName);
         admin.topics().grantPermission(topicName, "consumer", EnumSet.of(AuthAction.consume));
         admin.topics().grantPermission(topicName, "producer", EnumSet.of(AuthAction.produce));
 

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
@@ -3612,4 +3612,16 @@ public void testRetentionAndBacklogQuotaCheck() throws PulsarAdminException {
         });
 
     }
+
+    @Test
+    @SneakyThrows
+    public void testPermissions() {
+        String namespace = "prop-xyz/ns1/";
+        final String random = UUID.randomUUID().toString();
+        final String topic = "persistent://" + namespace + random;
+        final String subject = UUID.randomUUID().toString();
+        assertThrows(NotFoundException.class, () -> admin.topics().getPermissions(topic));
+        assertThrows(NotFoundException.class, () -> admin.topics().grantPermission(topic, subject, Set.of(AuthAction.produce)));
+        assertThrows(NotFoundException.class, () -> admin.topics().revokePermissions(topic, subject));
+    }
 }
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
@@ -888,12 +888,15 @@ public void testGetList() throws Exception {
     public void testGrantNonPartitionedTopic() {
         final String topicName = "non-partitioned-topic";
         AsyncResponse response = mock(AsyncResponse.class);
+        ArgumentCaptor<Response> responseCaptor = ArgumentCaptor.forClass(Response.class);
         persistentTopics.createNonPartitionedTopic(response, testTenant, testNamespace, topicName, true, null);
+        verify(response, timeout(5000).times(1)).resume(responseCaptor.capture());
+        Assert.assertEquals(responseCaptor.getValue().getStatus(), Response.Status.NO_CONTENT.getStatusCode());
         String role = "role";
         Set<AuthAction> expectActions = new HashSet<>();
         expectActions.add(AuthAction.produce);
         response = mock(AsyncResponse.class);
-        ArgumentCaptor<Response> responseCaptor = ArgumentCaptor.forClass(Response.class);
+        responseCaptor = ArgumentCaptor.forClass(Response.class);
         persistentTopics.grantPermissionsOnTopic(response, testTenant, testNamespace, topicName, role, expectActions);
         verify(response, timeout(5000).times(1)).resume(responseCaptor.capture());
         Assert.assertEquals(responseCaptor.getValue().getStatus(), Response.Status.NO_CONTENT.getStatusCode());
@@ -951,12 +954,15 @@ public void testGrantPartitionedTopic() {
     public void testRevokeNonPartitionedTopic() {
         final String topicName = "non-partitioned-topic";
         AsyncResponse response = mock(AsyncResponse.class);
+        ArgumentCaptor<Response> responseCaptor = ArgumentCaptor.forClass(Response.class);
         persistentTopics.createNonPartitionedTopic(response, testTenant, testNamespace, topicName, true, null);
+        verify(response, timeout(5000).times(1)).resume(responseCaptor.capture());
+        Assert.assertEquals(responseCaptor.getValue().getStatus(), Response.Status.NO_CONTENT.getStatusCode());
         String role = "role";
         Set<AuthAction> expectActions = new HashSet<>();
         expectActions.add(AuthAction.produce);
         response = mock(AsyncResponse.class);
-        ArgumentCaptor<Response> responseCaptor = ArgumentCaptor.forClass(Response.class);
+        responseCaptor = ArgumentCaptor.forClass(Response.class);
         persistentTopics.grantPermissionsOnTopic(response, testTenant, testNamespace, topicName, role, expectActions);
         verify(response, timeout(5000).times(1)).resume(responseCaptor.capture());
         Assert.assertEquals(responseCaptor.getValue().getStatus(), Response.Status.NO_CONTENT.getStatusCode());

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationTest.java
@@ -106,8 +106,9 @@ public void simple() throws Exception {
         assertTrue(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null));
         assertTrue(auth.canProduce(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null));
 
-        admin.topics().grantPermission("persistent://p1/c1/ns1/ds2", "other-role",
-                EnumSet.of(AuthAction.consume));
+        String topic = "persistent://p1/c1/ns1/ds2";
+        admin.topics().createNonPartitionedTopic(topic);
+        admin.topics().grantPermission(topic, "other-role", EnumSet.of(AuthAction.consume));
         waitForChange();
 
         assertTrue(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds2"), "other-role", null));
@@ -177,8 +178,9 @@ public void simple() throws Exception {
         assertFalse(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds2"), "my.role.1", null));
         assertFalse(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds2"), "my.role.2", null));
 
-        admin.topics().grantPermission("persistent://p1/c1/ns1/ds1", "my.*",
-                EnumSet.of(AuthAction.produce));
+        String topic1 = "persistent://p1/c1/ns1/ds1";
+        admin.topics().createNonPartitionedTopic(topic1);
+        admin.topics().grantPermission(topic1, "my.*", EnumSet.of(AuthAction.produce));
         waitForChange();
 
         assertTrue(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds1"), "my.role.1", null));
@@ -241,7 +243,7 @@ public void simple() throws Exception {
         assertTrue(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds1"), "role2", null, "role2-sub2"));
         assertTrue(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds1"), "pulsar.super_user", null, "role3-sub1"));
 
-        admin.namespaces().deleteNamespace("p1/c1/ns1");
+        admin.namespaces().deleteNamespace("p1/c1/ns1", true);
         admin.tenants().deleteTenant("p1");
 
         admin.clusters().deleteCluster("c1");

diff --git a/...-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java b/...-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java
@@ -262,8 +262,9 @@ public void testAnonymousSyncProducerAndConsumer(int batchMessageDelayMs) throws
         admin.close();
         admin = spy(PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString()).build());
         admin.namespaces().createNamespace("my-property/my-ns", Sets.newHashSet("test"));
-        admin.topics().grantPermission("persistent://my-property/my-ns/my-topic", "anonymousUser",
-                EnumSet.allOf(AuthAction.class));
+        String topic = "persistent://my-property/my-ns/my-topic";
+        admin.topics().createNonPartitionedTopic(topic);
+        admin.topics().grantPermission(topic, "anonymousUser", EnumSet.allOf(AuthAction.class));
 
         // setup the client
         replacePulsarClient(PulsarClient.builder().serviceUrl(pulsar.getBrokerServiceUrl())

diff --git a/...-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java b/...-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java
@@ -236,6 +236,7 @@ public void testSubscriberPermission() throws Exception {
         }
 
         // grant topic consume authorization to the subscriptionRole
+        tenantAdmin.topics().createNonPartitionedTopic(topicName);
         tenantAdmin.topics().grantPermission(topicName, subscriptionRole,
                 Collections.singleton(AuthAction.consume));
 
@@ -852,6 +853,7 @@ public void testPermissionForProducerCreateInitialSubscription() throws Exceptio
         admin.tenants().createTenant("my-property",
                 new TenantInfoImpl(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
         admin.namespaces().createNamespace("my-property/my-ns", Sets.newHashSet("test"));
+        admin.topics().createNonPartitionedTopic(topic);
         admin.topics().grantPermission(topic, invalidRole, Collections.singleton(AuthAction.produce));
         admin.topics().grantPermission(topic, producerRole, Sets.newHashSet(AuthAction.produce, AuthAction.consume));
 

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthorizationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthorizationTest.java
@@ -55,6 +55,7 @@ public ProxyAuthorizationTest() {
     @Override
     protected void setup() throws Exception {
         conf.setClusterName(configClusterName);
+        conf.setForceDeleteNamespaceAllowed(true);
         internalSetup();
 
         WebSocketProxyConfiguration config = new WebSocketProxyConfiguration();
@@ -99,8 +100,9 @@ public void test() throws Exception {
         assertTrue(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null));
         assertTrue(auth.canProduce(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null));
 
-        admin.topics().grantPermission("persistent://p1/c1/ns1/ds2", "other-role",
-                EnumSet.of(AuthAction.consume));
+        String topic = "persistent://p1/c1/ns1/ds2";
+        admin.topics().createNonPartitionedTopic(topic);
+        admin.topics().grantPermission(topic, "other-role", EnumSet.of(AuthAction.consume));
         waitForChange();
 
         assertTrue(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds2"), "other-role", null));
@@ -127,7 +129,7 @@ public void test() throws Exception {
         assertTrue(auth.canProduce(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null));
         assertTrue(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null, null));
 
-        admin.namespaces().deleteNamespace("p1/c1/ns1");
+        admin.namespaces().deleteNamespace("p1/c1/ns1", true);
         admin.tenants().deleteTenant("p1");
         admin.clusters().deleteCluster("c1");
     }

diff --git a/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarAuth.java b/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarAuth.java
@@ -154,7 +154,7 @@ public void testPulsarSqlAuth() throws PulsarAdminException {
         String partitionedTopic = "persistent://p1/c1/ns1/" + RandomStringUtils.randomAlphabetic(4);
         String passToken = AuthTokenUtils.createToken(secretKey, passRole, Optional.empty());
         String deniedToken = AuthTokenUtils.createToken(secretKey, deniedRole, Optional.empty());
-
+        admin.topics().createNonPartitionedTopic(topic);
         admin.topics().grantPermission(topic, passRole, EnumSet.of(AuthAction.consume));
         admin.topics().createPartitionedTopic(partitionedTopic, 2);
         admin.topics().grantPermission(partitionedTopic, passRole, EnumSet.of(AuthAction.consume));

diff --git a/...tegration/src/test/java/org/apache/pulsar/tests/integration/presto/TestPulsarSQLAuth.java b/...tegration/src/test/java/org/apache/pulsar/tests/integration/presto/TestPulsarSQLAuth.java
@@ -106,10 +106,8 @@ public void testPulsarSQLAuthCheck() throws PulsarAdminException {
         String passToken = AuthTokenUtils.createToken(secretKey, passRole, Optional.empty());
         String deniedToken = AuthTokenUtils.createToken(secretKey, deniedRole, Optional.empty());
         String topic = "testPulsarSQLAuthCheck";
-
-        admin.topics().grantPermission(topic, passRole, EnumSet.of(AuthAction.consume));
-
         admin.topics().createNonPartitionedTopic(topic);
+        admin.topics().grantPermission(topic, passRole, EnumSet.of(AuthAction.consume));
 
         String queryAllDataSql = String.format("select * from pulsar.\"%s\".\"%s\";", "public/default", topic);
 
@@ -173,9 +171,8 @@ public void testCheckAuthForMultipleTopics() throws PulsarAdminException {
         String topic1 = "testCheckAuthForMultipleTopics1";
         String topic2 = "testCheckAuthForMultipleTopics2";
 
-        admin.topics().grantPermission(topic1, testRole, EnumSet.of(AuthAction.consume));
-
         admin.topics().createNonPartitionedTopic(topic1);
+        admin.topics().grantPermission(topic1, testRole, EnumSet.of(AuthAction.consume));
 
         admin.topics().createPartitionedTopic(topic2, 2); // Test for partitioned topic