From 897daba523422d8e40714e6ee9b982dfba2f7545 Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Thu, 2 Feb 2023 18:28:07 +0800 Subject: [PATCH] [improve][authorization] Uniformly use allowTopicOperationAsync to check permissions Signed-off-by: Zixuan Liu --- .../authorization/AuthorizationProvider.java | 71 ++++++++++++++++--- .../authorization/AuthorizationService.java | 14 +++- .../MultiRolesTokenAuthorizationProvider.java | 44 ------------ .../PulsarAuthorizationProvider.java | 33 +++++---- .../pulsar/broker/auth/AuthorizationTest.java | 28 ++++---- .../AuthorizationProducerConsumerTest.java | 14 ---- .../proxy/ProxyAuthorizationTest.java | 4 +- 7 files changed, 107 insertions(+), 101 deletions(-) diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java index 67e096bee63e4..27c52e0c75c2f 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java @@ -105,9 +105,15 @@ default void initialize(ServiceConfiguration conf, PulsarResources pulsarResourc * the fully qualified topic name associated with the topic. * @param role * the app id used to send messages to the topic. + * + * @deprecated + * Use {@link #allowTopicOperationAsync(TopicName, String, TopicOperation, AuthenticationDataSource)} instead. */ - CompletableFuture canProduceAsync(TopicName topicName, String role, - AuthenticationDataSource authenticationData); + @Deprecated + default CompletableFuture canProduceAsync(TopicName topicName, String role, + AuthenticationDataSource authenticationData) { + return CompletableFuture.completedFuture(false); + } /** * Check if the specified role has permission to receive messages from the specified fully qualified topic name. @@ -118,9 +124,16 @@ CompletableFuture canProduceAsync(TopicName topicName, String role, * the app id used to receive messages from the topic. * @param subscription * the subscription name defined by the client + * + * @deprecated + * Use {@link #allowTopicOperationAsync(TopicName, String, TopicOperation, AuthenticationDataSource)} instead. */ - CompletableFuture canConsumeAsync(TopicName topicName, String role, - AuthenticationDataSource authenticationData, String subscription); + @Deprecated + default CompletableFuture canConsumeAsync(TopicName topicName, String role, + AuthenticationDataSource authenticationData, + String subscription) { + return CompletableFuture.completedFuture(false); + } /** * Check whether the specified role can perform a lookup for the specified topic. @@ -129,11 +142,15 @@ CompletableFuture canConsumeAsync(TopicName topicName, String role, * * @param topicName * @param role - * @return - * @throws Exception + * + * @deprecated + * Use {@link #allowTopicOperationAsync(TopicName, String, TopicOperation, AuthenticationDataSource)} instead. */ - CompletableFuture canLookupAsync(TopicName topicName, String role, - AuthenticationDataSource authenticationData); + @Deprecated + default CompletableFuture canLookupAsync(TopicName topicName, String role, + AuthenticationDataSource authenticationData) { + return CompletableFuture.completedFuture(false); + } /** * Allow all function operations with in this namespace. @@ -345,9 +362,41 @@ default CompletableFuture allowTopicOperationAsync(TopicName topic, String role, TopicOperation operation, AuthenticationDataSource authData) { - return FutureUtil.failedFuture( - new IllegalStateException("TopicOperation [" + operation.name() + "] is not supported by the Authorization" - + "provider you are using.")); + switch (operation) { + case LOOKUP: + case GET_STATS: + case GET_METADATA: + return canLookupAsync(topic, role, authData); + case PRODUCE: + return canProduceAsync(topic, role, authData); + case GET_SUBSCRIPTIONS: + case CONSUME: + case SUBSCRIBE: + case UNSUBSCRIBE: + case SKIP: + case EXPIRE_MESSAGES: + case PEEK_MESSAGES: + case RESET_CURSOR: + case GET_BACKLOG_SIZE: + case SET_REPLICATED_SUBSCRIPTION_STATUS: + case GET_REPLICATED_SUBSCRIPTION_STATUS: + return canConsumeAsync(topic, role, authData, authData.getSubscription()); + case TERMINATE: + case COMPACT: + case OFFLOAD: + case UNLOAD: + case DELETE_METADATA: + case UPDATE_METADATA: + case ADD_BUNDLE_RANGE: + case GET_BUNDLE_RANGE: + case DELETE_BUNDLE_RANGE: + return CompletableFuture.completedFuture(false); + default: + return FutureUtil.failedFuture( + new IllegalStateException( + "TopicOperation [" + operation.name() + "] is not supported by the Authorization" + + "provider you are using.")); + } } /** diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java index 526e8430d0f1a..048f570ee13b9 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java @@ -27,6 +27,7 @@ import org.apache.pulsar.broker.PulsarServerException; import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.authentication.AuthenticationDataSource; +import org.apache.pulsar.broker.authentication.AuthenticationDataSubscription; import org.apache.pulsar.broker.resources.PulsarResources; import org.apache.pulsar.common.naming.NamespaceName; import org.apache.pulsar.common.naming.TopicName; @@ -166,7 +167,7 @@ public CompletableFuture canProduceAsync(TopicName topicName, String ro if (isSuperUser) { return CompletableFuture.completedFuture(true); } else { - return provider.canProduceAsync(topicName, role, authenticationData); + return provider.allowTopicOperationAsync(topicName, role, TopicOperation.PRODUCE, authenticationData); } }); } @@ -191,7 +192,14 @@ public CompletableFuture canConsumeAsync(TopicName topicName, String ro if (isSuperUser) { return CompletableFuture.completedFuture(true); } else { - return provider.canConsumeAsync(topicName, role, authenticationData, subscription); + AuthenticationDataSource subscriptionDataSource; + if (authenticationData instanceof AuthenticationDataSubscription) { + subscriptionDataSource = authenticationData; + } else { + subscriptionDataSource = new AuthenticationDataSubscription(authenticationData, subscription); + } + return provider.allowTopicOperationAsync(topicName, role, TopicOperation.CONSUME, + subscriptionDataSource); } }); } @@ -273,7 +281,7 @@ public CompletableFuture canLookupAsync(TopicName topicName, String rol if (isSuperUser) { return CompletableFuture.completedFuture(true); } else { - return provider.canLookupAsync(topicName, role, authenticationData); + return provider.allowTopicOperationAsync(topicName, role, TopicOperation.LOOKUP, authenticationData); } }); } diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java index fa613245cfa27..8b8b40e7495e2 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java @@ -200,50 +200,6 @@ public CompletableFuture authorize(AuthenticationDataSource authenticat return FutureUtil.waitForAny(futures, ret -> (boolean) ret).thenApply(v -> v.isPresent()); } - /** - * Check if the specified role has permission to send messages to the specified fully qualified topic name. - * - * @param topicName the fully qualified topic name associated with the topic. - * @param role the app id used to send messages to the topic. - */ - @Override - public CompletableFuture canProduceAsync(TopicName topicName, String role, - AuthenticationDataSource authenticationData) { - return authorize(authenticationData, r -> super.canProduceAsync(topicName, r, authenticationData)); - } - - /** - * Check if the specified role has permission to receive messages from the specified fully qualified topic - * name. - * - * @param topicName the fully qualified topic name associated with the topic. - * @param role the app id used to receive messages from the topic. - * @param subscription the subscription name defined by the client - */ - @Override - public CompletableFuture canConsumeAsync(TopicName topicName, String role, - AuthenticationDataSource authenticationData, - String subscription) { - return authorize(authenticationData, r -> super.canConsumeAsync(topicName, r, authenticationData, - subscription)); - } - - /** - * Check whether the specified role can perform a lookup for the specified topic. - *

- * For that the caller needs to have producer or consumer permission. - * - * @param topicName - * @param role - * @return - * @throws Exception - */ - @Override - public CompletableFuture canLookupAsync(TopicName topicName, String role, - AuthenticationDataSource authenticationData) { - return authorize(authenticationData, r -> super.canLookupAsync(topicName, r, authenticationData)); - } - @Override public CompletableFuture allowFunctionOpsAsync(NamespaceName namespaceName, String role, AuthenticationDataSource authenticationData) { 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 3f6d38194713e..3ea4c8e3c7276 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 @@ -81,9 +81,7 @@ public void initialize(ServiceConfiguration conf, PulsarResources pulsarResource * @param role * the app id used to send messages to the topic. */ - @Override - public CompletableFuture canProduceAsync(TopicName topicName, String role, - AuthenticationDataSource authenticationData) { + private CompletableFuture checkProduceAsync(TopicName topicName, String role) { return checkAuthorization(topicName, role, AuthAction.produce); } @@ -95,12 +93,18 @@ public CompletableFuture canProduceAsync(TopicName topicName, String ro * the fully qualified topic name associated with the topic. * @param role * the app id used to receive messages from the topic. - * @param subscription - * the subscription name defined by the client + * @param authData + * is used for get the subscription name defined by the client */ - @Override - public CompletableFuture canConsumeAsync(TopicName topicName, String role, - AuthenticationDataSource authenticationData, String subscription) { + private CompletableFuture checkConsumeAsync(TopicName topicName, String role, + AuthenticationDataSource authData) { + String subscription; + if (authData != null) { + subscription = authData.getSubscription(); + } else { + subscription = null; + } + return pulsarResources.getNamespaceResources().getPoliciesAsync(topicName.getNamespaceObject()) .thenCompose(policies -> { if (!policies.isPresent()) { @@ -150,15 +154,14 @@ public CompletableFuture canConsumeAsync(TopicName topicName, String ro * @return * @throws Exception */ - @Override - public CompletableFuture canLookupAsync(TopicName topicName, String role, + private CompletableFuture checkLookupAsync(TopicName topicName, String role, AuthenticationDataSource authenticationData) { - return canProduceAsync(topicName, role, authenticationData) + return checkProduceAsync(topicName, role) .thenCompose(canProduce -> { if (canProduce) { return CompletableFuture.completedFuture(true); } - return canConsumeAsync(topicName, role, authenticationData, null); + return checkConsumeAsync(topicName, role, authenticationData); }); } @@ -520,9 +523,9 @@ public CompletableFuture allowTopicOperationAsync(TopicName topicName, case LOOKUP: case GET_STATS: case GET_METADATA: - return canLookupAsync(topicName, role, authData); + return checkLookupAsync(topicName, role, authData); case PRODUCE: - return canProduceAsync(topicName, role, authData); + return checkProduceAsync(topicName, role); case GET_SUBSCRIPTIONS: case CONSUME: case SUBSCRIBE: @@ -534,7 +537,7 @@ public CompletableFuture allowTopicOperationAsync(TopicName topicName, case GET_BACKLOG_SIZE: case SET_REPLICATED_SUBSCRIPTION_STATUS: case GET_REPLICATED_SUBSCRIPTION_STATUS: - return canConsumeAsync(topicName, role, authData, authData.getSubscription()); + return checkConsumeAsync(topicName, role, authData); case TERMINATE: case COMPACT: case OFFLOAD: 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 11afe889ee935..6145c033e3260 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 @@ -18,10 +18,11 @@ */ package org.apache.pulsar.broker.auth; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.Mockito.when; import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertThrows; import static org.testng.Assert.assertTrue; -import static org.testng.Assert.fail; import com.google.common.collect.Sets; import java.util.EnumSet; import org.apache.pulsar.broker.authorization.AuthorizationService; @@ -73,7 +74,9 @@ public void cleanup() throws Exception { public void simple() throws Exception { AuthorizationService auth = pulsar.getBrokerService().getAuthorizationService(); - assertFalse(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null)); + assertThatThrownBy( + () -> auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null)).hasMessageContaining( + "404"); admin.clusters().createCluster("c1", ClusterData.builder().build()); admin.tenants().createTenant("p1", new TenantInfoImpl(Sets.newHashSet("role1"), Sets.newHashSet("c1"))); @@ -209,20 +212,19 @@ public void simple() throws Exception { SubscriptionAuthMode.Prefix); waitForChange(); + // role1 is tenant's superuser 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 { - assertFalse(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds1"), "role1", null, "sub1")); - fail(); - } catch (Exception ignored) {} - try { - assertFalse(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds1"), "role2", null, "sub2")); - fail(); - } catch (Exception ignored) {} - + assertTrue(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds1"), "role1", null, "sub1")); assertTrue(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds1"), "role1", null, "role1-sub1")); + + // role2 + assertTrue(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds1"), "role2", null)); 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")); + assertTrue( + auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds1"), "pulsar.super_user", null, "role3-sub1")); + + assertThrows(() -> auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds1"), "role2", null, "sub2")); + assertThrows(() -> auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds1"), "role2", null, "sub1")); admin.namespaces().deleteNamespace("p1/c1/ns1"); admin.tenants().deleteTenant("p1"); 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 0ce3b7df07d1f..2f5875417e3b3 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 @@ -935,20 +935,6 @@ public Boolean allowNamespaceOperation( return null; } - @Override - public CompletableFuture allowTopicOperationAsync( - TopicName topic, String role, TopicOperation operation, AuthenticationDataSource authData) { - CompletableFuture isAuthorizedFuture; - - if (role.equals("plugbleRole")) { - isAuthorizedFuture = CompletableFuture.completedFuture(true); - } else { - isAuthorizedFuture = CompletableFuture.completedFuture(false); - } - - return isAuthorizedFuture; - } - @Override public Boolean allowTopicOperation( TopicName topicName, String role, TopicOperation operation, AuthenticationDataSource authData) { 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 a3b26a4a9d122..07a83f7090584 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 @@ -19,6 +19,7 @@ package org.apache.pulsar.websocket.proxy; import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgs; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyString; @@ -83,7 +84,8 @@ protected void cleanup() throws Exception { public void test() throws Exception { AuthorizationService auth = service.getAuthorizationService(); - assertFalse(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null)); + assertThatThrownBy(() -> auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", + null)).hasMessageContaining("404"); admin.clusters().createCluster(configClusterName, ClusterData.builder().build()); admin.tenants().createTenant("p1", new TenantInfoImpl(Sets.newHashSet("role1"), Sets.newHashSet("c1")));