Skip to content
This repository has been archived by the owner on Apr 1, 2024. It is now read-only.

[improve][broker] Add fine-grain authorization to retention admin API #6039

Closed
wants to merge 1 commit into from
Closed

[improve][broker] Add fine-grain authorization to retention admin API #6039

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
9 changes: 6 additions & 3 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2435,7 +2435,8 @@ public void getRetention(@Suspended final AsyncResponse asyncResponse,
@ApiParam(value = "Whether leader broker redirected this call to this broker. For internal use.")
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
preValidation(authoritative)
validateTopicPolicyOperationAsync(topicName, PolicyName.RETENTION, PolicyOperation.READ)
.thenCompose(__ -> preValidation(authoritative))
.thenCompose(__ -> internalGetRetention(applied, isGlobal))
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
Expand All @@ -2462,7 +2463,8 @@ public void setRetention(@Suspended final AsyncResponse asyncResponse,
@QueryParam("isGlobal") @DefaultValue("false") boolean isGlobal,
@ApiParam(value = "Retention policies for the specified topic") RetentionPolicies retention) {
validateTopicName(tenant, namespace, encodedTopic);
preValidation(authoritative)
validateTopicPolicyOperationAsync(topicName, PolicyName.RETENTION, PolicyOperation.WRITE)
.thenCompose(__ -> preValidation(authoritative))
.thenCompose(__ -> internalSetRetention(retention, isGlobal))
.thenRun(() -> {
try {
Expand Down Expand Up @@ -2498,7 +2500,8 @@ public void removeRetention(@Suspended final AsyncResponse asyncResponse,
@ApiParam(value = "Whether leader broker redirected this call to this broker. For internal use.")
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
preValidation(authoritative)
validateTopicPolicyOperationAsync(topicName, PolicyName.RETENTION, PolicyOperation.WRITE)
.thenCompose(__ -> preValidation(authoritative))
.thenCompose(__ -> internalRemoveRetention(isGlobal))
.thenRun(() -> {
log.info("[{}] Successfully remove retention: namespace={}, topic={}",
Expand Down
174 changes: 174 additions & 0 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesAuthZTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,174 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.pulsar.broker.admin;

import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
import io.jsonwebtoken.Jwts;
import java.util.Set;
import java.util.UUID;
import lombok.Cleanup;
import lombok.SneakyThrows;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.impl.auth.AuthenticationToken;
import org.apache.pulsar.common.policies.data.AuthAction;
import org.apache.pulsar.common.policies.data.RetentionPolicies;
import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.security.MockedPulsarStandalone;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;


public final class TopicPoliciesAuthZTest extends MockedPulsarStandalone {

private PulsarAdmin superUserAdmin;

private PulsarAdmin tenantManagerAdmin;

private static final String TENANT_ADMIN_SUBJECT = UUID.randomUUID().toString();
private static final String TENANT_ADMIN_TOKEN = Jwts.builder()
.claim("sub", TENANT_ADMIN_SUBJECT).signWith(SECRET_KEY).compact();

@SneakyThrows
@BeforeClass
public void before() {
loadTokenAuthentication();
loadDefaultAuthorization();
start();
this.superUserAdmin =PulsarAdmin.builder()
.serviceHttpUrl(getPulsarService().getWebServiceAddress())
.authentication(new AuthenticationToken(SUPER_USER_TOKEN))
.build();
final TenantInfo tenantInfo = superUserAdmin.tenants().getTenantInfo("public");
tenantInfo.getAdminRoles().add(TENANT_ADMIN_SUBJECT);
superUserAdmin.tenants().updateTenant("public", tenantInfo);
this.tenantManagerAdmin = PulsarAdmin.builder()
.serviceHttpUrl(getPulsarService().getWebServiceAddress())
.authentication(new AuthenticationToken(TENANT_ADMIN_TOKEN))
.build();
}


@SneakyThrows
@AfterClass
public void after() {
close();
}


@SneakyThrows
@Test
public void testRetention() {
final String random = UUID.randomUUID().toString();
final String topic = "persistent://public/default/" + random;
final String subject = UUID.randomUUID().toString();
final String token = Jwts.builder()
.claim("sub", subject).signWith(SECRET_KEY).compact();
superUserAdmin.topics().createNonPartitionedTopic(topic);

@Cleanup
final PulsarAdmin subAdmin = PulsarAdmin.builder()
.serviceHttpUrl(getPulsarService().getWebServiceAddress())
.authentication(new AuthenticationToken(token))
.build();
final RetentionPolicies definedRetentionPolicy = new RetentionPolicies(1, 1);
// test superuser
superUserAdmin.topicPolicies().setRetention(topic, definedRetentionPolicy);

// because the topic policies is eventual consistency, we should wait here
await().untilAsserted(() -> {
final RetentionPolicies receivedRetentionPolicy = superUserAdmin.topicPolicies().getRetention(topic);
Assert.assertEquals(receivedRetentionPolicy, definedRetentionPolicy);
});
superUserAdmin.topicPolicies().removeRetention(topic);

await().untilAsserted(() -> {
final RetentionPolicies retention = superUserAdmin.topicPolicies().getRetention(topic);
Assert.assertNull(retention);
});

// test tenant manager

tenantManagerAdmin.topicPolicies().setRetention(topic, definedRetentionPolicy);
await().untilAsserted(() -> {
final RetentionPolicies receivedRetentionPolicy = tenantManagerAdmin.topicPolicies().getRetention(topic);
Assert.assertEquals(receivedRetentionPolicy, definedRetentionPolicy);
});
tenantManagerAdmin.topicPolicies().removeRetention(topic);
await().untilAsserted(() -> {
final RetentionPolicies retention = tenantManagerAdmin.topicPolicies().getRetention(topic);
Assert.assertNull(retention);
});

// test nobody

try {
subAdmin.topicPolicies().getRetention(topic);
Assert.fail("unexpected behaviour");
} catch (PulsarAdminException ex) {
Assert.assertTrue(ex instanceof PulsarAdminException.NotAuthorizedException);
}

try {

subAdmin.topicPolicies().setRetention(topic, definedRetentionPolicy);
Assert.fail("unexpected behaviour");
} catch (PulsarAdminException ex) {
Assert.assertTrue(ex instanceof PulsarAdminException.NotAuthorizedException);
}

try {
subAdmin.topicPolicies().removeRetention(topic);
Assert.fail("unexpected behaviour");
} catch (PulsarAdminException ex) {
Assert.assertTrue(ex instanceof PulsarAdminException.NotAuthorizedException);
}

// test sub user with permissions
for (AuthAction action : AuthAction.values()) {
superUserAdmin.namespaces().grantPermissionOnNamespace("public/default",
subject, Set.of(action));
try {
subAdmin.topicPolicies().getRetention(topic);
Assert.fail("unexpected behaviour");
} catch (PulsarAdminException ex) {
Assert.assertTrue(ex instanceof PulsarAdminException.NotAuthorizedException);
}

try {

subAdmin.topicPolicies().setRetention(topic, definedRetentionPolicy);
Assert.fail("unexpected behaviour");
} catch (PulsarAdminException ex) {
Assert.assertTrue(ex instanceof PulsarAdminException.NotAuthorizedException);
}

try {
subAdmin.topicPolicies().removeRetention(topic);
Assert.fail("unexpected behaviour");
} catch (PulsarAdminException ex) {
Assert.assertTrue(ex instanceof PulsarAdminException.NotAuthorizedException);
}
superUserAdmin.namespaces().revokePermissionsOnNamespace("public/default", subject);
}
}

}
60 changes: 59 additions & 1 deletion .../security/tls/MockedPulsarStandalone.java → ...lsar/security/MockedPulsarStandalone.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,25 +16,36 @@
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.pulsar.security.tls;
package org.apache.pulsar.security;

import static org.apache.pulsar.utils.ResourceUtils.getAbsolutePath;
import io.jsonwebtoken.Jwts;
import io.jsonwebtoken.SignatureAlgorithm;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Sets;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import javax.crypto.SecretKey;
import lombok.Getter;
import lombok.SneakyThrows;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.authentication.AuthenticationProviderTls;
import org.apache.pulsar.broker.authentication.AuthenticationProviderToken;
import org.apache.pulsar.broker.authentication.utils.AuthTokenUtils;
import org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider;
import org.apache.pulsar.broker.testcontext.PulsarTestContext;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.impl.auth.AuthenticationKeyStoreTls;
import org.apache.pulsar.client.impl.auth.AuthenticationTls;
import org.apache.pulsar.client.impl.auth.AuthenticationToken;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.common.util.ObjectMapperFactory;



public abstract class MockedPulsarStandalone implements AutoCloseable {
Expand All @@ -60,6 +71,50 @@ public abstract class MockedPulsarStandalone implements AutoCloseable {
serviceConfiguration.setExposeBundlesMetricsInPrometheus(true);
}


protected static final SecretKey SECRET_KEY = AuthTokenUtils.createSecretKey(SignatureAlgorithm.HS256);

private static final String BROKER_INTERNAL_CLIENT_SUBJECT = "broker_internal";
private static final String BROKER_INTERNAL_CLIENT_TOKEN = Jwts.builder()
.claim("sub", BROKER_INTERNAL_CLIENT_SUBJECT).signWith(SECRET_KEY).compact();
protected static final String SUPER_USER_SUBJECT = "super-user";
protected static final String SUPER_USER_TOKEN = Jwts.builder()
.claim("sub", SUPER_USER_SUBJECT).signWith(SECRET_KEY).compact();
protected static final String NOBODY_SUBJECT = "nobody";
protected static final String NOBODY_TOKEN = Jwts.builder()
.claim("sub", NOBODY_SUBJECT).signWith(SECRET_KEY).compact();


@SneakyThrows
protected void loadTokenAuthentication() {
serviceConfiguration.setAuthenticationEnabled(true);
serviceConfiguration.setAuthenticationProviders(Set.of(AuthenticationProviderToken.class.getName()));
// internal client
serviceConfiguration.setBrokerClientAuthenticationPlugin(AuthenticationToken.class.getName());
final Map<String, String> brokerClientAuthParams = new HashMap<>();
brokerClientAuthParams.put("token", BROKER_INTERNAL_CLIENT_TOKEN);
final String brokerClientAuthParamStr = MAPPER.writeValueAsString(brokerClientAuthParams);
serviceConfiguration.setBrokerClientAuthenticationParameters(brokerClientAuthParamStr);

Properties properties = serviceConfiguration.getProperties();
if (properties == null) {
properties = new Properties();
serviceConfiguration.setProperties(properties);
}
properties.put("tokenSecretKey", AuthTokenUtils.encodeKeyBase64(SECRET_KEY));

}



protected void loadDefaultAuthorization() {
serviceConfiguration.setAuthorizationEnabled(true);
serviceConfiguration.setAuthorizationProvider(PulsarAuthorizationProvider.class.getName());
serviceConfiguration.setSuperUserRoles(Set.of(SUPER_USER_SUBJECT, BROKER_INTERNAL_CLIENT_SUBJECT));
}



@SneakyThrows
protected void loadECTlsCertificateWithFile() {
serviceConfiguration.setTlsEnabled(true);
Expand Down Expand Up @@ -176,4 +231,7 @@ public void close() throws Exception {
protected static final String TLS_EC_KS_TRUSTED_STORE =
getAbsolutePath("certificate-authority/ec/jks/ca.truststore.jks");
protected static final String TLS_EC_KS_TRUSTED_STORE_PASS = "rootpw";


private static final ObjectMapper MAPPER = ObjectMapperFactory.getMapper().getObjectMapper();
}
2 changes: 1 addition & 1 deletion ...-broker/src/test/java/org/apache/pulsar/security/tls/ec/TlsWithECCertificateFileTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@
import java.util.UUID;
import lombok.Cleanup;
import lombok.SneakyThrows;
import org.apache.pulsar.security.tls.MockedPulsarStandalone;
import org.apache.pulsar.security.MockedPulsarStandalone;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
Expand Down
10 changes: 5 additions & 5 deletions pulsar-broker/src/test/java/org/apache/pulsar/security/tls/ec/TlsWithECKeyStoreTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,9 +21,13 @@

import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import lombok.Cleanup;
import lombok.SneakyThrows;
import org.apache.pulsar.security.tls.MockedPulsarStandalone;
import org.apache.pulsar.security.MockedPulsarStandalone;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
Expand All @@ -35,10 +39,6 @@
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;


@Test
Expand Down
Loading