From 0f155b664d15013f19215bde40821de6401157b4 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Thu, 25 Apr 2024 11:54:00 +0800 Subject: [PATCH 1/9] change permission control for sub. --- .../broker/authorization/PulsarAuthorizationProvider.java | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) 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 a39c3d0560760..b6dc8945257c3 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 @@ -111,11 +111,10 @@ public CompletableFuture canConsumeAsync(TopicName topicName, String ro } } else { if (isNotBlank(subscription)) { - // validate if role is authorized to access subscription. (skip validation if authorization - // list is empty) + // validate if role is authorized to access subscription. Set roles = policies.get().auth_policies .getSubscriptionAuthentication().get(subscription); - if (roles != null && !roles.isEmpty() && !roles.contains(role)) { + if (roles == null || roles.isEmpty() || !roles.contains(role)) { log.warn("[{}] is not authorized to subscribe on {}-{}", role, topicName, subscription); return CompletableFuture.completedFuture(false); } From ff2ed045224286675131ea948dd8da47136f78c1 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Thu, 25 Apr 2024 15:33:29 +0800 Subject: [PATCH 2/9] add test code to reproduce. --- .../AuthorizationProducerConsumerTest.java | 57 +++++++++++++++++++ 1 file changed, 57 insertions(+) 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 769486054ab04..2ad74907b57c9 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 @@ -174,6 +174,63 @@ public void testProducerAndConsumerAuthorization() throws Exception { log.info("-- Exiting {} test --", methodName); } + @Test + public void testSubscriptionPermission() throws Exception { + log.info("-- Starting {} test --", methodName); + cleanup(); + conf.setTopicLevelPoliciesEnabled(false); + conf.setAuthorizationProvider(PulsarAuthorizationProvider.class.getName()); + setup(); + + final String jackRole = "jack"; + final String tomRole = "tom"; + final String namespace = "my-property/my-ns-sub-auth"; + final String topicName = "persistent://" + namespace + "/my-topic"; + + Authentication adminAuthentication = new ClientAuthentication("superUser"); + @Cleanup + PulsarAdmin superAdmin = spy( + PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString()).authentication(adminAuthentication).build()); + + Authentication jackAuthentication = new ClientAuthentication(jackRole); + @Cleanup + PulsarClient jackClient = PulsarClient.builder().serviceUrl(pulsar.getBrokerServiceUrl()) + .authentication(jackAuthentication).build(); + + Authentication tomAuthentication = new ClientAuthentication(tomRole); + @Cleanup + PulsarClient tomClient = PulsarClient.builder().serviceUrl(pulsar.getBrokerServiceUrl()) + .authentication(tomAuthentication).build(); + + superAdmin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build()); + superAdmin.tenants().createTenant("my-property", + new TenantInfoImpl(Sets.newHashSet(), Sets.newHashSet("test"))); + superAdmin.namespaces().createNamespace(namespace, Sets.newHashSet("test")); + superAdmin.topics().createPartitionedTopic(topicName, 1); + + // jack apply for new subscription `sub`, so admin grant consume permission and subscription permission to subject1 + superAdmin.topics().grantPermission(topicName, jackRole, Sets.newHashSet(AuthAction.consume)); + superAdmin.namespaces().grantPermissionOnSubscription(namespace, "sub", + Sets.newHashSet(jackRole)); + + // as jack get the consume permission, he can create any new subscription `sub1`, `sub2` and so on + Consumer consumer = jackClient.newConsumer().topic(topicName).subscriptionName("sub1").subscribe(); + consumer.close(); + + // tom apply for new subscription `sub1`, so admin grant consume permission and subscription permission to subject2 + superAdmin.topics().grantPermission(topicName, tomRole, Sets.newHashSet(AuthAction.consume)); + superAdmin.namespaces().grantPermissionOnSubscription(namespace, "sub1", + Sets.newHashSet(tomRole)); + + // at this time, consumer can't receive any message, because jack don't have the subscription permission for `sub1` + try { + consumer = jackClient.newConsumer().topic(topicName).subscriptionName("sub1").subscribe(); + consumer.receive(); + } catch (PulsarClientException.AuthorizationException e) { + fail("reproduce issue with authorization exception: " + e.getMessage()); + } + } + @Test public void testSubscriberPermission() throws Exception { log.info("-- Starting {} test --", methodName); From a24e36089f87136a195d7d4f12bc7cfc5693f3f6 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 26 Apr 2024 11:16:24 +0800 Subject: [PATCH 3/9] fix test code. --- .../AuthorizationProducerConsumerTest.java | 31 ++++++++++++++----- 1 file changed, 24 insertions(+), 7 deletions(-) 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 2ad74907b57c9..af61b74463685 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 @@ -174,6 +174,10 @@ public void testProducerAndConsumerAuthorization() throws Exception { log.info("-- Exiting {} test --", methodName); } + /** + * verify new permission check for subscription + * @throws Exception + */ @Test public void testSubscriptionPermission() throws Exception { log.info("-- Starting {} test --", methodName); @@ -204,7 +208,7 @@ public void testSubscriptionPermission() throws Exception { superAdmin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build()); superAdmin.tenants().createTenant("my-property", - new TenantInfoImpl(Sets.newHashSet(), Sets.newHashSet("test"))); + new TenantInfoImpl(new HashSet<>(), Sets.newHashSet("test"))); superAdmin.namespaces().createNamespace(namespace, Sets.newHashSet("test")); superAdmin.topics().createPartitionedTopic(topicName, 1); @@ -213,19 +217,32 @@ public void testSubscriptionPermission() throws Exception { superAdmin.namespaces().grantPermissionOnSubscription(namespace, "sub", Sets.newHashSet(jackRole)); - // as jack get the consume permission, he can create any new subscription `sub1`, `sub2` and so on - Consumer consumer = jackClient.newConsumer().topic(topicName).subscriptionName("sub1").subscribe(); - consumer.close(); + // jack can create new subscription `sub` with subscription permission + Consumer consumer; + try { + consumer = jackClient.newConsumer().topic(topicName).subscriptionName("sub").subscribe(); + consumer.close(); + } catch (PulsarClientException.AuthorizationException e) { + fail("reproduce issue with authorization exception: " + e.getMessage()); + } + + // though jack get the consume permission, he can't create any new subscription `sub1` without subscription permission + try { + consumer = jackClient.newConsumer().topic(topicName).subscriptionName("sub1").subscribe(); + consumer.close(); + } catch (PulsarClientException.AuthorizationException e) { + // we do not allow to create new subscription without subscription permission + } // tom apply for new subscription `sub1`, so admin grant consume permission and subscription permission to subject2 superAdmin.topics().grantPermission(topicName, tomRole, Sets.newHashSet(AuthAction.consume)); superAdmin.namespaces().grantPermissionOnSubscription(namespace, "sub1", Sets.newHashSet(tomRole)); - // at this time, consumer can't receive any message, because jack don't have the subscription permission for `sub1` + // tom can create new subscription `sub1` with subscription permission try { - consumer = jackClient.newConsumer().topic(topicName).subscriptionName("sub1").subscribe(); - consumer.receive(); + consumer = tomClient.newConsumer().topic(topicName).subscriptionName("sub1").subscribe(); + consumer.close(); } catch (PulsarClientException.AuthorizationException e) { fail("reproduce issue with authorization exception: " + e.getMessage()); } From 98e66d9dbcd59439efa449cfb3a609c604717fa8 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 26 Apr 2024 11:54:53 +0800 Subject: [PATCH 4/9] fix test code. --- .../api/AuthorizationProducerConsumerTest.java | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) 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 af61b74463685..72b7fd41afd42 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 @@ -310,6 +310,11 @@ public void testSubscriberPermission() throws Exception { // grant topic consume authorization to the subscriptionRole tenantAdmin.topics().grantPermission(topicName, subscriptionRole, Collections.singleton(AuthAction.consume)); + // grant subscription permission to the subscriptionRole + tenantAdmin.namespaces().grantPermissionOnSubscription(namespace, subscriptionName, + Collections.singleton(subscriptionRole)); + tenantAdmin.namespaces().grantPermissionOnSubscription(namespace, subscriptionName2, + Collections.singleton(subscriptionRole)); replacePulsarClient(PulsarClient.builder() .serviceUrl(pulsar.getBrokerServiceUrl()) @@ -391,6 +396,8 @@ public void testSubscriberPermission() throws Exception { Collections.singleton(otherPrincipal)); TreeMap> permissionOnSubscription = new TreeMap<>(); permissionOnSubscription.put(subscriptionName, Collections.singleton(otherPrincipal)); + // we have granted subscription permission of subscriptionName2 to subscriptionRole before. + permissionOnSubscription.put(subscriptionName2, Collections.singleton(subscriptionRole)); Assert.assertEquals(tenantAdmin.namespaces().getPermissionOnSubscription(namespace), permissionOnSubscription); // now, subscriptionRole doesn't have subscription level access so, it will fail to access subscription @@ -414,6 +421,8 @@ public void testSubscriberPermission() throws Exception { Sets.newHashSet(otherPrincipal, subscriptionRole)); TreeMap> permissionOnSubscription1 = new TreeMap<>(); permissionOnSubscription1.put(subscriptionName, Sets.newHashSet(otherPrincipal, subscriptionRole)); + // we have granted subscription permission of subscriptionName2 to subscriptionRole before. + permissionOnSubscription1.put(subscriptionName2, Collections.singleton(subscriptionRole)); Assert.assertEquals(tenantAdmin.namespaces().getPermissionOnSubscription(namespace), permissionOnSubscription1); sub1Admin.topics().skipAllMessages(topicName, subscriptionName); @@ -479,9 +488,12 @@ public void testClearBacklogPermission() throws Exception { assertEquals(tenantAdmin.topics().getPartitionedTopicList(namespace), Lists.newArrayList(topicName)); - // grant topic consume&produce authorization to the subscriptionRole + // grant topic consume&produce authorization and subscription permission + // to the subscriptionRole superAdmin.topics().grantPermission(topicName, subscriptionRole, Sets.newHashSet(AuthAction.produce, AuthAction.consume)); + superAdmin.namespaces().grantPermissionOnSubscription(namespace, subscriptionName, + Sets.newHashSet(subscriptionRole)); replacePulsarClient(PulsarClient.builder() .serviceUrl(pulsar.getBrokerServiceUrl()) .authentication(subAdminAuthentication)); @@ -778,6 +790,7 @@ public void testRevokePermission() throws Exception { Assert.assertFalse(authorizationService.canConsume(topicName, role, null, "sub1")); authorizationService.grantPermissionAsync(topicName, actions, role, "auth-json").get(); authorizationService.grantPermissionAsync(topicName, actions, role2, "auth-json").get(); + authorizationService.grantSubscriptionPermissionAsync(namespaceName, "sub1", Sets.newHashSet(role, role2), "auth-json").get(); Assert.assertTrue(authorizationService.canProduce(topicName, role, null)); Assert.assertTrue(authorizationService.canConsume(topicName, role, null, "sub1")); @@ -872,6 +885,9 @@ public void testPermissionForProducerCreateInitialSubscription() throws Exceptio PulsarClient pulsarClientProducerRole = PulsarClient.builder().serviceUrl(lookupUrl) .authentication(authenticationProducerRole).build(); + // as producer will create an initial subscription, it should have permission to create the subscription + admin.namespaces().grantPermissionOnSubscription(tn.getNamespace(), initialSubscriptionName, + Sets.newHashSet(producerRole)); Producer producer = ((ProducerBuilderImpl) pulsarClientProducerRole.newProducer()) .initialSubscriptionName(initialSubscriptionName) .topic(topic) From f4b926264efb15ea11cf86dbbfa5e777d7b3bbf1 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 26 Apr 2024 15:08:38 +0800 Subject: [PATCH 5/9] add code. --- .../server/ProxyWithAuthorizationTest.java | 23 +++++++++++++------ 1 file changed, 16 insertions(+), 7 deletions(-) diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java index bc96c7ea51041..473c24296fa68 100644 --- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java +++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java @@ -278,9 +278,11 @@ public void testProxyAuthorization() throws Exception { initializeCluster(admin, namespaceName); + String subscriptionName = "my-subscriber-name"; + admin.namespaces().grantPermissionOnSubscription(namespaceName, subscriptionName, Sets.newHashSet("Client")); Consumer consumer = proxyClient.newConsumer() .topic("persistent://my-tenant/my-ns/my-topic1") - .subscriptionName("my-subscriber-name").subscribe(); + .subscriptionName(subscriptionName).subscribe(); Producer producer = proxyClient.newProducer(Schema.BYTES) .topic("persistent://my-tenant/my-ns/my-topic1").create(); @@ -339,8 +341,10 @@ public void testTlsHostVerificationProxyToClient(boolean hostnameVerificationEna } try { + String subscriptionName = "my-subscriber-name"; + admin.namespaces().grantPermissionOnSubscription(namespaceName, subscriptionName, Sets.newHashSet("Client")); proxyClient.newConsumer().topic("persistent://my-tenant/my-ns/my-topic1") - .subscriptionName("my-subscriber-name").subscribe(); + .subscriptionName(subscriptionName).subscribe(); if (hostnameVerificationEnabled) { Assert.fail("Connection should be failed due to hostnameVerification enabled"); } @@ -398,8 +402,10 @@ public void testTlsHostVerificationProxyToBroker(boolean hostnameVerificationEna } try { + String subscriptionName = "my-subscriber-name"; + admin.namespaces().grantPermissionOnSubscription(namespaceName, subscriptionName, Sets.newHashSet("Client")); proxyClient.newConsumer().topic("persistent://my-tenant/my-ns/my-topic1") - .subscriptionName("my-subscriber-name").subscribe(); + .subscriptionName(subscriptionName).subscribe(); if (hostnameVerificationEnabled) { Assert.fail("Connection should be failed due to hostnameVerification enabled"); } @@ -418,7 +424,7 @@ public void testTlsHostVerificationProxyToBroker(boolean hostnameVerificationEna * This test verifies whether the Client and Proxy honor the protocols and ciphers specified. Details description of * test cases can be found in protocolsCiphersProviderCodecProvider */ - @Test(dataProvider = "protocolsCiphersProvider", timeOut = 5000) + @Test(dataProvider = "protocolsCiphersProvider", timeOut = 10000) public void tlsCiphersAndProtocols(Set tlsCiphers, Set tlsProtocols, boolean expectFailure) throws Exception { log.info("-- Starting {} test --", methodName); @@ -482,9 +488,11 @@ public void tlsCiphersAndProtocols(Set tlsCiphers, Set tlsProtoc try { @Cleanup PulsarClient proxyClient = createPulsarClient("pulsar://localhost:" + proxyService.getListenPortTls().get(), PulsarClient.builder()); + String subscriptionName = "my-subscriber-name"; + admin.namespaces().grantPermissionOnSubscription(namespaceName, subscriptionName, Sets.newHashSet("Client")); Consumer consumer = proxyClient.newConsumer() .topic("persistent://my-tenant/my-ns/my-topic1") - .subscriptionName("my-subscriber-name").subscribe(); + .subscriptionName(subscriptionName).subscribe(); if (expectFailure) { Assert.fail("Failure expected for this test case"); @@ -532,10 +540,11 @@ public void testProxyTlsTransportWithAuth(Authentication auth) throws Exception String namespaceName = "my-tenant/my-ns"; initializeCluster(admin, namespaceName); - + String subscriptionName = "my-subscriber-name"; + admin.namespaces().grantPermissionOnSubscription(namespaceName, subscriptionName, Sets.newHashSet("Client")); Consumer consumer = proxyClient.newConsumer() .topic("persistent://my-tenant/my-ns/my-topic1") - .subscriptionName("my-subscriber-name").subscribe(); + .subscriptionName(subscriptionName).subscribe(); Producer producer = proxyClient.newProducer(Schema.BYTES) .topic("persistent://my-tenant/my-ns/my-topic1").create(); From 54a13f4246a3dd6555acb631948971a67d25983b Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 26 Apr 2024 15:28:50 +0800 Subject: [PATCH 6/9] fix test code. --- .../pulsar/broker/admin/TopicAuthZTest.java | 28 +++++++++++++++++-- 1 file changed, 25 insertions(+), 3 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAuthZTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAuthZTest.java index ad47ac74a8980..db4cc7f59a251 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAuthZTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAuthZTest.java @@ -298,6 +298,7 @@ public void testGetPartitionedStatsAndInternalStats() { @SneakyThrows public void testCreateSubscriptionAndUpdateSubscriptionPropertiesAndAnalyzeSubscriptionBacklog() { final String random = UUID.randomUUID().toString(); + final String namespace = "public/default"; final String topic = "persistent://public/default/" + random; final String subject = UUID.randomUUID().toString(); final String token = Jwts.builder() @@ -322,7 +323,9 @@ public void testCreateSubscriptionAndUpdateSubscriptionPropertiesAndAnalyzeSubsc for (AuthAction action : AuthAction.values()) { superUserAdmin.topics().grantPermission(topic, subject, Set.of(action)); if (action == AuthAction.consume) { - subAdmin.topics().createSubscription(topic, "test-sub" + suffix.incrementAndGet(), MessageId.earliest); + String subscriptionName = "test-sub" + suffix.incrementAndGet(); + superUserAdmin.namespaces().grantPermissionOnSubscription(namespace, subscriptionName, Sets.newHashSet(subject)); + subAdmin.topics().createSubscription(topic, subscriptionName, MessageId.earliest); } else { Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, () -> subAdmin.topics().createSubscription(topic, "test-sub" + suffix.incrementAndGet(), MessageId.earliest)); @@ -361,8 +364,10 @@ public void testCreateSubscriptionAndUpdateSubscriptionPropertiesAndAnalyzeSubsc for (AuthAction action : AuthAction.values()) { superUserAdmin.topics().grantPermission(topic, subject, Set.of(action)); if (action == AuthAction.consume) { - subAdmin.topics().updateSubscriptionProperties(topic, "test-sub", properties); - subAdmin.topics().getSubscriptionProperties(topic, "test-sub"); + String subscriptionName = "test-sub"; + superUserAdmin.namespaces().grantPermissionOnSubscription(namespace, subscriptionName, Sets.newHashSet(subject)); + subAdmin.topics().updateSubscriptionProperties(topic, subscriptionName, properties); + subAdmin.topics().getSubscriptionProperties(topic, subscriptionName); subAdmin.topics().analyzeSubscriptionBacklog(TopicName.get(topic).getPartition(0).getLocalName(), "test-sub", Optional.empty()); } else { Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, @@ -706,6 +711,7 @@ public void testGetInternalStats(boolean partitioned) { @SneakyThrows public void testDeleteSubscription(boolean partitioned) { final String random = UUID.randomUUID().toString(); + final String namespace = "public/default"; final String topic = "persistent://public/default/" + random; final String subject = UUID.randomUUID().toString(); final String token = Jwts.builder() @@ -734,6 +740,8 @@ public void testDeleteSubscription(boolean partitioned) { for (AuthAction action : AuthAction.values()) { superUserAdmin.topics().grantPermission(topic, subject, Set.of(action)); if (AuthAction.consume == action) { + String subscriptionName = "test-sub"; + superUserAdmin.namespaces().grantPermissionOnSubscription(namespace, subscriptionName, Sets.newHashSet(subject)); subAdmin.topics().deleteSubscription(topic, "test-sub"); } else { Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, @@ -748,6 +756,7 @@ public void testDeleteSubscription(boolean partitioned) { @SneakyThrows public void testSkipAllMessage(boolean partitioned) { final String random = UUID.randomUUID().toString(); + final String namespace = "public/default"; final String topic = "persistent://public/default/" + random; final String subject = UUID.randomUUID().toString(); final String token = Jwts.builder() @@ -773,6 +782,7 @@ public void testSkipAllMessage(boolean partitioned) { for (AuthAction action : AuthAction.values()) { superUserAdmin.topics().grantPermission(topic, subject, Set.of(action)); if (AuthAction.consume == action) { + superUserAdmin.namespaces().grantPermissionOnSubscription(namespace, subName, Sets.newHashSet(subject)); subAdmin.topics().skipAllMessages(topic,subName); } else { Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, @@ -787,6 +797,7 @@ public void testSkipAllMessage(boolean partitioned) { @SneakyThrows public void testSkipMessage() { final String random = UUID.randomUUID().toString(); + final String namespace = "public/default"; final String topic = "persistent://public/default/" + random; final String subject = UUID.randomUUID().toString(); final String token = Jwts.builder() @@ -811,6 +822,7 @@ public void testSkipMessage() { for (AuthAction action : AuthAction.values()) { superUserAdmin.topics().grantPermission(topic, subject, Set.of(action)); if (AuthAction.consume == action) { + superUserAdmin.namespaces().grantPermissionOnSubscription(namespace, subName, Sets.newHashSet(subject)); subAdmin.topics().skipMessages(topic, subName, 1); } else { Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, @@ -861,6 +873,7 @@ public void testExpireMessagesForAllSubscriptions(boolean partitioned) { @SneakyThrows public void testResetCursor(boolean partitioned) { final String random = UUID.randomUUID().toString(); + final String namespace = "public/default"; final String topic = "persistent://public/default/" + random; final String subject = UUID.randomUUID().toString(); final String token = Jwts.builder() @@ -885,6 +898,7 @@ public void testResetCursor(boolean partitioned) { for (AuthAction action : AuthAction.values()) { superUserAdmin.topics().grantPermission(topic, subject, Set.of(action)); if (AuthAction.consume == action) { + superUserAdmin.namespaces().grantPermissionOnSubscription(namespace, subName, Sets.newHashSet(subject)); subAdmin.topics().resetCursor(topic, subName, System.currentTimeMillis()); } else { Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, @@ -899,6 +913,7 @@ public void testResetCursor(boolean partitioned) { @SneakyThrows public void testResetCursorOnPosition() { final String random = UUID.randomUUID().toString(); + final String namespace = "public/default"; final String topic = "persistent://public/default/" + random; final String subject = UUID.randomUUID().toString(); final String token = Jwts.builder() @@ -923,6 +938,7 @@ public void testResetCursorOnPosition() { for (AuthAction action : AuthAction.values()) { superUserAdmin.topics().grantPermission(topic, subject, Set.of(action)); if (AuthAction.consume == action) { + superUserAdmin.namespaces().grantPermissionOnSubscription(namespace, subName, Sets.newHashSet(subject)); subAdmin.topics().resetCursor(topic, subName, MessageId.latest); } else { Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, @@ -983,6 +999,7 @@ public void testGetMessageById() { @SneakyThrows public void testPeekNthMessage() { final String random = UUID.randomUUID().toString(); + final String namespace = "public/default"; final String topic = "persistent://public/default/" + random; final String subject = UUID.randomUUID().toString(); final String token = Jwts.builder() @@ -1015,6 +1032,7 @@ public void testPeekNthMessage() { for (AuthAction action : AuthAction.values()) { superUserAdmin.topics().grantPermission(topic, subject, Set.of(action)); if (AuthAction.consume == action) { + superUserAdmin.namespaces().grantPermissionOnSubscription(namespace, subName, Sets.newHashSet(subject)); subAdmin.topics().peekMessages(topic, subName, 1); } else { Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, @@ -1075,6 +1093,7 @@ public void testExamineMessage() { @SneakyThrows public void testExpireMessage() { final String random = UUID.randomUUID().toString(); + final String namespace = "public/default"; final String topic = "persistent://public/default/" + random; final String subject = UUID.randomUUID().toString(); final String token = Jwts.builder() @@ -1112,6 +1131,7 @@ public void testExpireMessage() { for (AuthAction action : AuthAction.values()) { superUserAdmin.topics().grantPermission(topic, subject, Set.of(action)); if (AuthAction.consume == action) { + superUserAdmin.namespaces().grantPermissionOnSubscription(namespace, subName, Sets.newHashSet(subject)); subAdmin.topics().expireMessages(topic, subName, 1); } else { Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, @@ -1127,6 +1147,7 @@ public void testExpireMessage() { public void testExpireMessageByPosition() { final String random = UUID.randomUUID().toString(); final String topic = "persistent://public/default/" + random; + final String namespace = "public/default"; final String subject = UUID.randomUUID().toString(); final String token = Jwts.builder() .claim("sub", subject).signWith(SECRET_KEY).compact(); @@ -1163,6 +1184,7 @@ public void testExpireMessageByPosition() { for (AuthAction action : AuthAction.values()) { superUserAdmin.topics().grantPermission(topic, subject, Set.of(action)); if (AuthAction.consume == action) { + superUserAdmin.namespaces().grantPermissionOnSubscription(namespace, subName, Sets.newHashSet(subject)); subAdmin.topics().expireMessages(topic, subName, MessageId.earliest, false); } else { Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, From ce3206a5d4c9df12332dcc248bfcb5d65b45a72d Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 26 Apr 2024 16:03:50 +0800 Subject: [PATCH 7/9] fix test code. --- .../org/apache/pulsar/broker/auth/AuthorizationTest.java | 5 +++++ 1 file changed, 5 insertions(+) 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 f59f9d480b8c2..df8a53bfdbf98 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 @@ -230,14 +230,19 @@ public void simple() throws Exception { assertTrue(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds1"), "role1", null)); assertTrue(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds1"), "role2", null)); try { + admin.namespaces().grantPermissionOnSubscription("p1/c1/ns1", "sub1", Sets.newHashSet("role1")); assertFalse(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds1"), "role1", null, "sub1")); fail(); } catch (Exception ignored) {} try { + admin.namespaces().grantPermissionOnSubscription("p1/c1/ns1", "sub2", Sets.newHashSet("role2")); assertFalse(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds1"), "role2", null, "sub2")); fail(); } catch (Exception ignored) {} + // though we use prefix subscription mode, we still require that the role own the permission. + admin.namespaces().grantPermissionOnSubscription("p1/c1/ns1", "role1-sub1", Sets.newHashSet("role1")); + admin.namespaces().grantPermissionOnSubscription("p1/c1/ns1", "role2-sub2", Sets.newHashSet("role2")); assertTrue(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds1"), "role1", null, "role1-sub1")); 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")); From 0200f39fb15cf6917c2f0f5860134aec78bcf287 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 26 Apr 2024 16:19:17 +0800 Subject: [PATCH 8/9] fix test code. --- .../org/apache/pulsar/proxy/server/ProxyForwardAuthDataTest.java | 1 + .../pulsar/proxy/server/ProxyWithAuthorizationNegTest.java | 1 + 2 files changed, 2 insertions(+) diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyForwardAuthDataTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyForwardAuthDataTest.java index 5e969ca26e4fd..d037fbfdef833 100644 --- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyForwardAuthDataTest.java +++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyForwardAuthDataTest.java @@ -99,6 +99,7 @@ public void testForwardAuthData() throws Exception { Sets.newHashSet(AuthAction.consume, AuthAction.produce)); admin.namespaces().grantPermissionOnNamespace(namespaceName, "client", Sets.newHashSet(AuthAction.consume, AuthAction.produce)); + admin.namespaces().grantPermissionOnSubscription(namespaceName, subscriptionName, Sets.newHashSet("client", "proxy")); // Step 2: Run Pulsar Proxy without forwarding authData - expect Exception ProxyConfiguration proxyConfig = new ProxyConfiguration(); diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationNegTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationNegTest.java index cf9ad5831ec0a..bab0f4a305c6a 100644 --- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationNegTest.java +++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationNegTest.java @@ -184,6 +184,7 @@ public void testProxyAuthorization() throws Exception { admin.namespaces().grantPermissionOnNamespace(namespaceName, "Proxy", Sets.newHashSet(AuthAction.produce)); admin.namespaces().grantPermissionOnNamespace(namespaceName, "Client", Sets.newHashSet(AuthAction.consume, AuthAction.produce)); + admin.namespaces().grantPermissionOnSubscription(namespaceName, "my-subscriber-name", Sets.newHashSet("Client", "Proxy")); Consumer consumer; try { From 8305fa623a978425c4b630e60ade5556413c9862 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 26 Apr 2024 17:22:59 +0800 Subject: [PATCH 9/9] fix test code. --- .../pulsar/proxy/server/ProxyWithJwtAuthorizationTest.java | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithJwtAuthorizationTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithJwtAuthorizationTest.java index 5fb3e04682421..e809457e6a1d8 100644 --- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithJwtAuthorizationTest.java +++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithJwtAuthorizationTest.java @@ -187,6 +187,8 @@ public void testProxyAuthorization() throws Exception { // excepted admin.namespaces().grantPermissionOnNamespace(namespaceName, CLIENT_ROLE, Sets.newHashSet(AuthAction.consume)); + admin.namespaces().grantPermissionOnSubscription(namespaceName, "my-subscriber-name", + Sets.newHashSet(CLIENT_ROLE)); log.info("-- Admin permissions {} ---", admin.namespaces().getPermissions(namespaceName)); consumer = proxyClient.newConsumer() .topic("persistent://my-property/proxy-authorization/my-ns/my-topic1") @@ -263,6 +265,7 @@ public void testUpdatePartitionNumAndReconnect() throws Exception { admin.topics().createPartitionedTopic(topicName, 2); admin.topics().grantPermission(topicName, CLIENT_ROLE, Sets.newHashSet(AuthAction.consume, AuthAction.produce)); + admin.namespaces().grantPermissionOnSubscription(namespaceName, subscriptionName, Sets.newHashSet(CLIENT_ROLE)); Consumer consumer = proxyClient.newConsumer() .topic(topicName) @@ -385,6 +388,8 @@ public void testProxyAuthorizationWithPrefixSubscriptionAuthMode() throws Except Assert.fail("should have failed with authorization error"); } catch (Exception ex) { // excepted + admin.namespaces().grantPermissionOnSubscription(namespaceName, CLIENT_ROLE + "-sub1", + Sets.newHashSet(CLIENT_ROLE)); consumer = proxyClient.newConsumer() .topic("persistent://my-property/proxy-authorization/my-ns/my-topic1") .subscriptionName(CLIENT_ROLE + "-sub1").subscribe();