From 73c95a7b3dff925b45105a281693e074c723731e Mon Sep 17 00:00:00 2001 From: Shen Liu Date: Fri, 23 Apr 2021 10:52:16 +0800 Subject: [PATCH] Fix authorization error if partition number of partitioned topic is updated. (#10300) --- .../PulsarAuthorizationProvider.java | 21 ++++- .../server/ProxyWithJwtAuthorizationTest.java | 91 +++++++++++++++++++ 2 files changed, 110 insertions(+), 2 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 c7dd2f41d0537..f0118b6ac82fa 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 @@ -434,16 +434,33 @@ public CompletableFuture checkPermission(TopicName topicName, String ro return; } } + + // If the partition number of the partitioned topic having topic level policy is updated, + // the new sub partitions may not inherit the policy of the partition topic. + // We can also check the permission of partitioned topic. + // For https://github.com/apache/pulsar/issues/10300 + if (topicName.isPartitioned()) { + topicRoles = policies.get().auth_policies.destination_auth.get(topicName.getPartitionedTopicName()); + if (topicRoles != null) { + // Topic has custom policy + Set topicActions = topicRoles.get(role); + if (topicActions != null && topicActions.contains(action)) { + // The role has topic level permission + permissionFuture.complete(true); + return; + } + } + } } permissionFuture.complete(false); }).exceptionally(ex -> { - log.warn("Client with Role - {} failed to get permissions for topic - {}. {}", role, topicName, + log.warn("Client with Role - {} failed to get permissions for topic - {}. {}", role, topicName, ex.getMessage()); permissionFuture.completeExceptionally(ex); return null; }); } catch (Exception e) { - log.warn("Client with Role - {} failed to get permissions for topic - {}. {}", role, topicName, + log.warn("Client with Role - {} failed to get permissions for topic - {}. {}", role, topicName, e.getMessage()); permissionFuture.completeExceptionally(e); } 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 f683adfb5905a..4a7d0b848d7f2 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 @@ -201,6 +201,97 @@ public void testProxyAuthorization() throws Exception { log.info("-- Exiting {} test --", methodName); } + /** + * 
+     * 1. Create a 2-partition topic and grant produce/consume permission to client role.
+     * 2. Use producer/consumer with client role to process the topic, which is fine.
+     * 2. Update the topic partition number to 4.
+     * 3. Use new producer/consumer with client role to process the topic.
+     * 4. Broker should authorize producer/consumer normally.
+     *
+ */ + @Test + public void testUpdatePartitionNumAndReconnect() throws Exception { + log.info("-- Starting {} test --", methodName); + + startProxy(); + createAdminClient(); + PulsarClient proxyClient = createPulsarClient(proxyService.getServiceUrl(), PulsarClient.builder()); + + String clusterName = "proxy-authorization"; + String namespaceName = "my-property/my-ns"; + String topicName = "persistent://my-property/my-ns/my-topic1"; + String subscriptionName = "my-subscriber-name"; + + admin.clusters().createCluster(clusterName, new ClusterData(brokerUrl.toString())); + + admin.tenants().createTenant("my-property", + new TenantInfo(Sets.newHashSet(), Sets.newHashSet(clusterName))); + admin.namespaces().createNamespace(namespaceName); + admin.topics().createPartitionedTopic(topicName, 2); + admin.topics().grantPermission(topicName, CLIENT_ROLE, + Sets.newHashSet(AuthAction.consume, AuthAction.produce)); + + Consumer consumer = proxyClient.newConsumer() + .topic(topicName) + .subscriptionName(subscriptionName).subscribe(); + + Producer producer = proxyClient.newProducer(Schema.BYTES) + .topic(topicName).create(); + final int MSG_NUM = 10; + Set messageSet = Sets.newHashSet(); + for (int i = 0; i < MSG_NUM; i++) { + String message = "my-message-" + i; + messageSet.add(message); + producer.send(message.getBytes()); + } + + Message msg; + Set receivedMessageSet = Sets.newHashSet(); + for (int i = 0; i < MSG_NUM; i++) { + msg = consumer.receive(5, TimeUnit.SECONDS); + String receivedMessage = new String(msg.getData()); + log.debug("Received message: [{}]", receivedMessage); + String expectedMessage = "my-message-" + i; + receivedMessageSet.add(expectedMessage); + consumer.acknowledgeAsync(msg); + } + Assert.assertEquals(messageSet, receivedMessageSet); + consumer.close(); + producer.close(); + + // update partition num + admin.topics().updatePartitionedTopic(topicName, 4); + + // produce/consume the topic again + consumer = proxyClient.newConsumer() + .topic(topicName) + .subscriptionName(subscriptionName).subscribe(); + producer = proxyClient.newProducer(Schema.BYTES) + .topic(topicName).create(); + + messageSet.clear(); + for (int i = 0; i < MSG_NUM; i++) { + String message = "my-message-" + i; + messageSet.add(message); + producer.send(message.getBytes()); + } + + receivedMessageSet.clear(); + for (int i = 0; i < MSG_NUM; i++) { + msg = consumer.receive(5, TimeUnit.SECONDS); + String receivedMessage = new String(msg.getData()); + log.debug("Received message: [{}]", receivedMessage); + String expectedMessage = "my-message-" + i; + receivedMessageSet.add(expectedMessage); + consumer.acknowledgeAsync(msg); + } + Assert.assertEquals(messageSet, receivedMessageSet); + consumer.close(); + producer.close(); + log.info("-- Exiting {} test --", methodName); + } + /** * 
      * It verifies jwt + Authentication + Authorization (client -> proxy -> broker).