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); } 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, 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")); 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..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 @@ -174,6 +174,80 @@ 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); + 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(new HashSet<>(), 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)); + + // 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)); + + // tom can create new subscription `sub1` with subscription permission + try { + consumer = tomClient.newConsumer().topic(topicName).subscriptionName("sub1").subscribe(); + consumer.close(); + } catch (PulsarClientException.AuthorizationException e) { + fail("reproduce issue with authorization exception: " + e.getMessage()); + } + } + @Test public void testSubscriberPermission() throws Exception { log.info("-- Starting {} test --", methodName); @@ -236,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()) @@ -317,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 @@ -340,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); @@ -405,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)); @@ -704,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")); @@ -798,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) 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 { 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(); 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();