From e2f62dec099807521fa8075366648158ccc555b2 Mon Sep 17 00:00:00 2001 From: Jiwei Guo Date: Sat, 18 May 2024 22:57:45 +0800 Subject: [PATCH] [improve][admin] Check if the topic existed before the permission operations (#22742) (cherry picked from commit 71640f696fd9109bc677408e3c2cbacb2fb7252b) --- .../broker/admin/impl/PersistentTopicsBase.java | 12 ++++++++---- .../broker/admin/AdminApiSchemaWithAuthTest.java | 1 + .../org/apache/pulsar/broker/admin/AdminApiTest.java | 12 ++++++++++++ .../pulsar/broker/admin/PersistentTopicsTest.java | 10 ++++++++-- .../apache/pulsar/broker/auth/AuthorizationTest.java | 12 +++++++----- .../api/AuthenticatedProducerConsumerTest.java | 5 +++-- .../api/AuthorizationProducerConsumerTest.java | 2 ++ .../websocket/proxy/ProxyAuthorizationTest.java | 8 +++++--- .../org/apache/pulsar/sql/presto/TestPulsarAuth.java | 2 +- .../tests/integration/presto/TestPulsarSQLAuth.java | 7 ++----- 10 files changed, 49 insertions(+), 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 index 9dac4f28c40b1..543d6f5e1bbd3 100644 --- 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> internalGetPartitionedTopicListAsync() protected CompletableFuture>> 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 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 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 index 5159d7b714195..15d6e509ca708 100644 --- 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 index 2426be2bee91b..0dfee193a5727 100644 --- 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 index e0621f364c8c1..9d5bb75efd6d1 100644 --- 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 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 expectActions = new HashSet<>(); expectActions.add(AuthAction.produce); response = mock(AsyncResponse.class); - ArgumentCaptor 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 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 expectActions = new HashSet<>(); expectActions.add(AuthAction.produce); response = mock(AsyncResponse.class); - ArgumentCaptor 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 index 7acd39d741d88..6a75353240f6c 100644 --- 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/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java index 3bd8b920a30fe..44d8549fa6243 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java +++ b/pulsar-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/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 f547052344d72..d7588a8a9e1b4 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 @@ -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 index 8253c39177ceb..750b0d84cba41 100644 --- 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 index 7b550b7270f37..412c41f8b891e 100644 --- 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/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestPulsarSQLAuth.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestPulsarSQLAuth.java index 87db46f2bb625..e3b232021b14c 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestPulsarSQLAuth.java +++ b/tests/integration/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