Skip to content

[improve][authorization] Uniformly use allowTopicOperationAsync to check permissions #19461

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
nodece wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[improve][authorization] Uniformly use allowTopicOperationAsync to check permissions #19461

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 60 additions & 11 deletions ...er-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Boolean> canProduceAsync(TopicName topicName, String role,
AuthenticationDataSource authenticationData);
@Deprecated
default CompletableFuture<Boolean> 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.
Expand All @@ -118,9 +124,16 @@ CompletableFuture<Boolean> 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<Boolean> canConsumeAsync(TopicName topicName, String role,
AuthenticationDataSource authenticationData, String subscription);
@Deprecated
default CompletableFuture<Boolean> 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.
Expand All @@ -129,11 +142,15 @@ CompletableFuture<Boolean> canConsumeAsync(TopicName topicName, String role,
*
* @param topicName
* @param role
* @return
* @throws Exception
*
* @deprecated
* Use {@link #allowTopicOperationAsync(TopicName, String, TopicOperation, AuthenticationDataSource)} instead.
*/
CompletableFuture<Boolean> canLookupAsync(TopicName topicName, String role,
AuthenticationDataSource authenticationData);
@Deprecated
default CompletableFuture<Boolean> canLookupAsync(TopicName topicName, String role,
AuthenticationDataSource authenticationData) {
return CompletableFuture.completedFuture(false);
}
Comment on lines +150 to +153

@michaeljmarshall michaeljmarshall Apr 14, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Instead of providing a default implementation, I think we should just deprecate these methods. Then, in the PulsarAuthorizationProvider class, we can remove the @Override annotation and those methods become internal implementation details instead of top level methods on the interface.


/**
* Allow all function operations with in this namespace.
Expand Down Expand Up @@ -345,9 +362,41 @@ default CompletableFuture<Boolean> 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."));
Comment on lines -348 to -350

@michaeljmarshall michaeljmarshall Apr 14, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Are you able to explain a bit more why this is required for backwards compatibility? I think we should be able to leave this here and leave the switch statement in the PulsarAuthorizationProvider.

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."));
}
}

/**
Expand Down
14 changes: 11 additions & 3 deletions ...ker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -166,7 +167,7 @@ public CompletableFuture<Boolean> 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);
}
});
}
Expand All @@ -191,7 +192,14 @@ public CompletableFuture<Boolean> 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);
Comment on lines -194 to +202

@michaeljmarshall michaeljmarshall Apr 14, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we deprecate this canConsumeAsync method? It seems like the same kind of logic that we're deprecating in the AuthorizationProvider.

}
});
}
Expand Down Expand Up @@ -273,7 +281,7 @@ public CompletableFuture<Boolean> 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);
Comment on lines 281 to +284

@michaeljmarshall michaeljmarshall Apr 14, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is only tangentially related to your PR, but it seems like a broken abstraction that the AuthorizationService decides that the superuser privilege supersedes the AuthenticationProvider implementaiton. Instead, I think we could just call provider.allowTopicOperationAsync(topicName, role, TopicOperation.LOOKUP, authenticationData);, as you've done here.

My goal is to move that logic into the provider so that plugin implementations do not need to deal with the complexity of when to check for super user and when not to. All of this logic should be delegated to the provider, in my opinion.

That change is probably worth its own PR, though.

}
});
}
Expand Down
44 changes: 0 additions & 44 deletions ...ain/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,50 +200,6 @@ public CompletableFuture<Boolean> 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<Boolean> 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<Boolean> 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.
* <p>
* For that the caller needs to have producer or consumer permission.
*
* @param topicName
* @param role
* @return
* @throws Exception
*/
@Override
public CompletableFuture<Boolean> canLookupAsync(TopicName topicName, String role,
AuthenticationDataSource authenticationData) {
return authorize(authenticationData, r -> super.canLookupAsync(topicName, r, authenticationData));
}

@Override
public CompletableFuture<Boolean> allowFunctionOpsAsync(NamespaceName namespaceName, String role,
AuthenticationDataSource authenticationData) {
Expand Down
33 changes: 18 additions & 15 deletions ...mon/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Boolean> canProduceAsync(TopicName topicName, String role,
AuthenticationDataSource authenticationData) {
private CompletableFuture<Boolean> checkProduceAsync(TopicName topicName, String role) {
return checkAuthorization(topicName, role, AuthAction.produce);
}

Expand All @@ -95,12 +93,18 @@ public CompletableFuture<Boolean> 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<Boolean> canConsumeAsync(TopicName topicName, String role,
AuthenticationDataSource authenticationData, String subscription) {
private CompletableFuture<Boolean> 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()) {
Expand Down Expand Up @@ -150,15 +154,14 @@ public CompletableFuture<Boolean> canConsumeAsync(TopicName topicName, String ro
* @return
* @throws Exception
*/
@Override
public CompletableFuture<Boolean> canLookupAsync(TopicName topicName, String role,
private CompletableFuture<Boolean> 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);
});
}

Expand Down Expand Up @@ -520,9 +523,9 @@ public CompletableFuture<Boolean> 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:
Expand All @@ -534,7 +537,7 @@ public CompletableFuture<Boolean> 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:
Expand Down
28 changes: 15 additions & 13 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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")));
Expand Down Expand Up @@ -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");
Expand Down
14 changes: 0 additions & 14 deletions ...-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -935,20 +935,6 @@ public Boolean allowNamespaceOperation(
return null;
}

@Override
public CompletableFuture<Boolean> allowTopicOperationAsync(
TopicName topic, String role, TopicOperation operation, AuthenticationDataSource authData) {
CompletableFuture<Boolean> 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) {
Expand Down
4 changes: 3 additions & 1 deletion pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthorizationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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")));
Expand Down