Skip to content
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

[improve][admin] Opt-out of topic-existence check #354

Merged
merged 4 commits into from
Dec 30, 2024
Merged

[improve][admin] Opt-out of topic-existence check #354

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
e26395b
[fix][admin] Fix can't delete tenant for v1 (#22550)
Technoboy- Apr 23, 2024
799230c
[improve][admin] Check if the topic existed before the permission ope…
Technoboy- May 18, 2024
c067284
[improve][admin] Opt-out of topic-existence check (#23709)
omarkj Dec 11, 2024
7ef66a1
fix topic not found issue in ProxyAuthorizationWithoutImplicitPermiss…
nikhil-ctds Dec 27, 2024
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
6 changes: 6 additions & 0 deletions pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3312,6 +3312,12 @@ The delayed message index time step(in seconds) in per bucket snapshot segment,
)
private int transactionPendingAckBatchedWriteMaxDelayInMillis = 1;

@FieldContext(
category = CATEGORY_SERVER,
doc = "Opt-out of topic-existence check when setting permissions"
)
private boolean allowAclChangesOnNonExistentTopics = false;

@FieldContext(
category = CATEGORY_SERVER,
doc = "The class name of the factory that implements the topic compaction service."
Expand Down
10 changes: 8 additions & 2 deletions pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicResources.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,8 +100,14 @@ public CompletableFuture<Void> clearDomainPersistence(NamespaceName ns) {

public CompletableFuture<Void> clearTenantPersistence(String tenant) {
String path = MANAGED_LEDGER_PATH + "/" + tenant;
log.info("Clearing tenant persistence for tenant: {}, path {}", tenant, path);
return store.deleteIfExists(path, Optional.empty());
return store.exists(path)
.thenCompose(exists -> {
if (exists) {
return store.deleteRecursive(path);
} else {
return CompletableFuture.completedFuture(null);
}
});
}

void handleNotification(Notification notification) {
Expand Down
41 changes: 34 additions & 7 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -221,7 +221,16 @@ protected CompletableFuture<List<String>> internalGetPartitionedTopicListAsync()

protected CompletableFuture<Map<String, Set<AuthAction>>> internalGetPermissionsOnTopic() {
// This operation should be reading from zookeeper and it should be allowed without having admin privileges
return validateAdminAccessForTenantAsync(namespaceName.getTenant())
CompletableFuture<Void> validateAccessForTenantCf =
validateAdminAccessForTenantAsync(namespaceName.getTenant());

var checkIfTopicExists = !pulsar().getConfiguration().isAllowAclChangesOnNonExistentTopics();
if (checkIfTopicExists) {
validateAccessForTenantCf = validateAccessForTenantCf
.thenCompose(__ -> internalCheckTopicExists(topicName));
}

return validateAccessForTenantCf
.thenCompose(__ -> getAuthorizationService().getPermissionsAsync(topicName));
}

Expand Down Expand Up @@ -272,10 +281,18 @@ private CompletableFuture<Void> grantPermissionsAsync(TopicName topicUri, String
protected void internalGrantPermissionsOnTopic(final AsyncResponse asyncResponse, String role,
Set<AuthAction> actions) {
// This operation should be reading from zookeeper and it should be allowed without having admin privileges
validateAdminAccessForTenantAsync(namespaceName.getTenant())
.thenCompose(__ -> validatePoliciesReadOnlyAccessAsync().thenCompose(unused1 ->
grantPermissionsAsync(topicName, role, actions)
.thenAccept(unused -> asyncResponse.resume(Response.noContent().build()))))
CompletableFuture<Void> validateAccessForTenantCf = validateAdminAccessForTenantAsync(namespaceName.getTenant())
.thenCompose(__ -> validatePoliciesReadOnlyAccessAsync());

var checkIfTopicExists = !pulsar().getConfiguration().isAllowAclChangesOnNonExistentTopics();
if (checkIfTopicExists) {
validateAccessForTenantCf = validateAccessForTenantCf
.thenCompose(__ -> internalCheckTopicExists(topicName));
}

validateAccessForTenantCf
.thenCompose(unused1 -> grantPermissionsAsync(topicName, role, actions))
.thenAccept(unused -> asyncResponse.resume(Response.noContent().build()))
.exceptionally(ex -> {
Throwable realCause = FutureUtil.unwrapCompletionException(ex);
log.error("[{}] Failed to get permissions for topic {}", clientAppId(), topicName, realCause);
Expand All @@ -286,8 +303,18 @@ protected void internalGrantPermissionsOnTopic(final AsyncResponse asyncResponse

protected void internalRevokePermissionsOnTopic(AsyncResponse asyncResponse, String role) {
// This operation should be reading from zookeeper and it should be allowed without having admin privileges
validateAdminAccessForTenantAsync(namespaceName.getTenant())
.thenCompose(__ -> getPartitionedTopicMetadataAsync(topicName, true, false)
CompletableFuture<Void> validateAccessForTenantCf =
validateAdminAccessForTenantAsync(namespaceName.getTenant())
.thenCompose(__ -> validatePoliciesReadOnlyAccessAsync());

var checkIfTopicExists = !pulsar().getConfiguration().isAllowAclChangesOnNonExistentTopics();
if (checkIfTopicExists) {
validateAccessForTenantCf = validateAccessForTenantCf
.thenCompose(__ -> internalCheckTopicExists(topicName));
}

validateAccessForTenantCf
.thenCompose(unused1 -> getPartitionedTopicMetadataAsync(topicName, true, false)
.thenCompose(metadata -> {
int numPartitions = metadata.partitions;
CompletableFuture<Void> future = CompletableFuture.completedFuture(null);
Expand Down
1 change: 1 addition & 0 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaWithAuthTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,7 @@ public void testGetCreateDeleteSchema() throws Exception {
.serviceHttpUrl(brokerUrl != null ? brokerUrl.toString() : brokerUrlTls.toString())
.authentication(AuthenticationToken.class.getName(), PRODUCE_TOKEN)
.build();
admin.topics().createNonPartitionedTopic(topicName);
admin.topics().grantPermission(topicName, "consumer", EnumSet.of(AuthAction.consume));
admin.topics().grantPermission(topicName, "producer", EnumSet.of(AuthAction.produce));

Expand Down
32 changes: 32 additions & 0 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.admin;

import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
Expand Down Expand Up @@ -3612,4 +3613,35 @@ public void testRetentionAndBacklogQuotaCheck() throws PulsarAdminException {
});

}

@Test
@SneakyThrows
public void testPermissions() {
String namespace = "prop-xyz/ns1/";
final String random = UUID.randomUUID().toString();
final String topic = "persistent://" + namespace + random;
final String subject = UUID.randomUUID().toString();
assertThrows(NotFoundException.class, () -> admin.topics().getPermissions(topic));
assertThrows(NotFoundException.class, () -> admin.topics().grantPermission(topic, subject, Set.of(AuthAction.produce)));
assertThrows(NotFoundException.class, () -> admin.topics().revokePermissions(topic, subject));
}

@Test
@SneakyThrows
public void testPermissionsAllowAclChangesOnNonExistentTopics() {
pulsar.getConfiguration().setAllowAclChangesOnNonExistentTopics(true);
try {
String namespace = "prop-xyz/ns1/";
final String random = UUID.randomUUID().toString();
final String topic = "persistent://" + namespace + random;
final String subject = UUID.randomUUID().toString();
admin.topics().grantPermission(topic, subject, Set.of(AuthAction.produce));
assertThat(admin.topics().getPermissions(topic).get(subject)).containsExactly(AuthAction.produce);
admin.topics().revokePermissions(topic, subject);
assertThat(admin.topics().getPermissions(topic).get(subject)).isNullOrEmpty();
} finally {
// reset config
pulsar.getConfiguration().setAllowAclChangesOnNonExistentTopics(false);
}
}
}
10 changes: 8 additions & 2 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -888,12 +888,15 @@ public void testGetList() throws Exception {
public void testGrantNonPartitionedTopic() {
final String topicName = "non-partitioned-topic";
AsyncResponse response = mock(AsyncResponse.class);
ArgumentCaptor<Response> responseCaptor = ArgumentCaptor.forClass(Response.class);
persistentTopics.createNonPartitionedTopic(response, testTenant, testNamespace, topicName, true, null);
verify(response, timeout(5000).times(1)).resume(responseCaptor.capture());
Assert.assertEquals(responseCaptor.getValue().getStatus(), Response.Status.NO_CONTENT.getStatusCode());
String role = "role";
Set<AuthAction> expectActions = new HashSet<>();
expectActions.add(AuthAction.produce);
response = mock(AsyncResponse.class);
ArgumentCaptor<Response> responseCaptor = ArgumentCaptor.forClass(Response.class);
responseCaptor = ArgumentCaptor.forClass(Response.class);
persistentTopics.grantPermissionsOnTopic(response, testTenant, testNamespace, topicName, role, expectActions);
verify(response, timeout(5000).times(1)).resume(responseCaptor.capture());
Assert.assertEquals(responseCaptor.getValue().getStatus(), Response.Status.NO_CONTENT.getStatusCode());
Expand Down Expand Up @@ -951,12 +954,15 @@ public void testGrantPartitionedTopic() {
public void testRevokeNonPartitionedTopic() {
final String topicName = "non-partitioned-topic";
AsyncResponse response = mock(AsyncResponse.class);
ArgumentCaptor<Response> responseCaptor = ArgumentCaptor.forClass(Response.class);
persistentTopics.createNonPartitionedTopic(response, testTenant, testNamespace, topicName, true, null);
verify(response, timeout(5000).times(1)).resume(responseCaptor.capture());
Assert.assertEquals(responseCaptor.getValue().getStatus(), Response.Status.NO_CONTENT.getStatusCode());
String role = "role";
Set<AuthAction> expectActions = new HashSet<>();
expectActions.add(AuthAction.produce);
response = mock(AsyncResponse.class);
ArgumentCaptor<Response> responseCaptor = ArgumentCaptor.forClass(Response.class);
responseCaptor = ArgumentCaptor.forClass(Response.class);
persistentTopics.grantPermissionsOnTopic(response, testTenant, testNamespace, topicName, role, expectActions);
verify(response, timeout(5000).times(1)).resume(responseCaptor.capture());
Assert.assertEquals(responseCaptor.getValue().getStatus(), Response.Status.NO_CONTENT.getStatusCode());
Expand Down
41 changes: 36 additions & 5 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 @@ -32,6 +32,7 @@
import org.apache.pulsar.broker.resources.PulsarResources;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminBuilder;
import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.AuthAction;
Expand All @@ -55,12 +56,17 @@ public AuthorizationTest() {
@Override
public void setup() throws Exception {
conf.setClusterName("c1");
conf.setSystemTopicEnabled(false);
conf.setAuthenticationEnabled(true);
conf.setForceDeleteNamespaceAllowed(true);
conf.setForceDeleteTenantAllowed(true);
conf.setAuthenticationProviders(
Sets.newHashSet("org.apache.pulsar.broker.auth.MockAuthenticationProvider"));
conf.setAuthorizationEnabled(true);
conf.setAuthorizationAllowWildcardsMatching(true);
conf.setSuperUserRoles(Sets.newHashSet("pulsar.super_user", "pass.pass"));
conf.setBrokerClientAuthenticationPlugin(MockAuthentication.class.getName());
conf.setBrokerClientAuthenticationParameters("user:pass.pass");
internalSetup();
}

Expand All @@ -69,6 +75,11 @@ protected void customizeNewPulsarAdminBuilder(PulsarAdminBuilder pulsarAdminBuil
pulsarAdminBuilder.authentication(new MockAuthentication("pass.pass"));
}

@Override
protected void customizeNewPulsarClientBuilder(ClientBuilder clientBuilder) {
clientBuilder.authentication(new MockAuthentication("pass.pass"));
}

@AfterClass(alwaysRun = true)
@Override
public void cleanup() throws Exception {
Expand All @@ -95,8 +106,9 @@ public void simple() throws Exception {
assertTrue(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null));
assertTrue(auth.canProduce(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null));

admin.topics().grantPermission("persistent://p1/c1/ns1/ds2", "other-role",
EnumSet.of(AuthAction.consume));
String topic = "persistent://p1/c1/ns1/ds2";
admin.topics().createNonPartitionedTopic(topic);
admin.topics().grantPermission(topic, "other-role", EnumSet.of(AuthAction.consume));
waitForChange();

assertTrue(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds2"), "other-role", null));
Expand Down Expand Up @@ -166,8 +178,9 @@ public void simple() throws Exception {
assertFalse(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds2"), "my.role.1", null));
assertFalse(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds2"), "my.role.2", null));

admin.topics().grantPermission("persistent://p1/c1/ns1/ds1", "my.*",
EnumSet.of(AuthAction.produce));
String topic1 = "persistent://p1/c1/ns1/ds1";
admin.topics().createNonPartitionedTopic(topic1);
admin.topics().grantPermission(topic1, "my.*", EnumSet.of(AuthAction.produce));
waitForChange();

assertTrue(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds1"), "my.role.1", null));
Expand Down Expand Up @@ -230,8 +243,26 @@ public void simple() throws Exception {
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"));

admin.namespaces().deleteNamespace("p1/c1/ns1");
admin.namespaces().deleteNamespace("p1/c1/ns1", true);
admin.tenants().deleteTenant("p1");

admin.clusters().deleteCluster("c1");
}

@Test
public void testDeleteV1Tenant() throws Exception {
admin.clusters().createCluster("c1", ClusterData.builder().build());
admin.tenants().createTenant("p1", new TenantInfoImpl(Sets.newHashSet("role1"), Sets.newHashSet("c1")));
waitForChange();
admin.namespaces().createNamespace("p1/c1/ns1");
waitForChange();


String topic = "persistent://p1/c1/ns1/ds2";
admin.topics().createNonPartitionedTopic(topic);

admin.namespaces().deleteNamespace("p1/c1/ns1", true);
admin.tenants().deleteTenant("p1", true);
admin.clusters().deleteCluster("c1");
}

Expand Down
5 changes: 3 additions & 2 deletions ...-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -262,8 +262,9 @@ public void testAnonymousSyncProducerAndConsumer(int batchMessageDelayMs) throws
admin.close();
admin = spy(PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString()).build());
admin.namespaces().createNamespace("my-property/my-ns", Sets.newHashSet("test"));
admin.topics().grantPermission("persistent://my-property/my-ns/my-topic", "anonymousUser",
EnumSet.allOf(AuthAction.class));
String topic = "persistent://my-property/my-ns/my-topic";
admin.topics().createNonPartitionedTopic(topic);
admin.topics().grantPermission(topic, "anonymousUser", EnumSet.allOf(AuthAction.class));

// setup the client
replacePulsarClient(PulsarClient.builder().serviceUrl(pulsar.getBrokerServiceUrl())
Expand Down
2 changes: 2 additions & 0 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 @@ -236,6 +236,7 @@ public void testSubscriberPermission() throws Exception {
}

// grant topic consume authorization to the subscriptionRole
tenantAdmin.topics().createNonPartitionedTopic(topicName);
tenantAdmin.topics().grantPermission(topicName, subscriptionRole,
Collections.singleton(AuthAction.consume));

Expand Down Expand Up @@ -852,6 +853,7 @@ public void testPermissionForProducerCreateInitialSubscription() throws Exceptio
admin.tenants().createTenant("my-property",
new TenantInfoImpl(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
admin.namespaces().createNamespace("my-property/my-ns", Sets.newHashSet("test"));
admin.topics().createNonPartitionedTopic(topic);
admin.topics().grantPermission(topic, invalidRole, Collections.singleton(AuthAction.produce));
admin.topics().grantPermission(topic, producerRole, Sets.newHashSet(AuthAction.produce, AuthAction.consume));

Expand Down
8 changes: 5 additions & 3 deletions 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 @@ -55,6 +55,7 @@ public ProxyAuthorizationTest() {
@Override
protected void setup() throws Exception {
conf.setClusterName(configClusterName);
conf.setForceDeleteNamespaceAllowed(true);
internalSetup();

WebSocketProxyConfiguration config = new WebSocketProxyConfiguration();
Expand Down Expand Up @@ -99,8 +100,9 @@ public void test() throws Exception {
assertTrue(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null));
assertTrue(auth.canProduce(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null));

admin.topics().grantPermission("persistent://p1/c1/ns1/ds2", "other-role",
EnumSet.of(AuthAction.consume));
String topic = "persistent://p1/c1/ns1/ds2";
admin.topics().createNonPartitionedTopic(topic);
admin.topics().grantPermission(topic, "other-role", EnumSet.of(AuthAction.consume));
waitForChange();

assertTrue(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds2"), "other-role", null));
Expand All @@ -127,7 +129,7 @@ public void test() throws Exception {
assertTrue(auth.canProduce(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null));
assertTrue(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null, null));

admin.namespaces().deleteNamespace("p1/c1/ns1");
admin.namespaces().deleteNamespace("p1/c1/ns1", true);
admin.tenants().deleteTenant("p1");
admin.clusters().deleteCluster("c1");
}
Expand Down
3 changes: 3 additions & 0 deletions ...pulsar/websocket/proxy/ProxyAuthorizationWithoutImplicitPermissionOnSubscriptionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,8 @@ public void testAuthorizationServiceDirectly() throws Exception {
assertTrue(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null));
assertTrue(auth.canProduce(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null));

String topic = "persistent://p1/c1/ns1/ds2";
admin.topics().createNonPartitionedTopic(topic);
admin.topics().grantPermission("persistent://p1/c1/ns1/ds2", "other-role",
EnumSet.of(AuthAction.consume));
waitForChange();
Expand Down Expand Up @@ -134,6 +136,7 @@ public void testAuthorizationServiceDirectly() throws Exception {
assertTrue(auth.canProduce(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null));
assertTrue(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null, null));

admin.topics().delete(topic);
admin.namespaces().deleteNamespace("p1/c1/ns1");
admin.tenants().deleteTenant("p1");
admin.clusters().deleteCluster("c1");
Expand Down
2 changes: 1 addition & 1 deletion pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarAuth.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -154,7 +154,7 @@ public void testPulsarSqlAuth() throws PulsarAdminException {
String partitionedTopic = "persistent://p1/c1/ns1/" + RandomStringUtils.randomAlphabetic(4);
String passToken = AuthTokenUtils.createToken(secretKey, passRole, Optional.empty());
String deniedToken = AuthTokenUtils.createToken(secretKey, deniedRole, Optional.empty());

admin.topics().createNonPartitionedTopic(topic);
admin.topics().grantPermission(topic, passRole, EnumSet.of(AuthAction.consume));
admin.topics().createPartitionedTopic(partitionedTopic, 2);
admin.topics().grantPermission(partitionedTopic, passRole, EnumSet.of(AuthAction.consume));
Expand Down
Loading
Loading