Skip to content

Commit

Permalink
[improve][broker] Add fine-grain authorization to retention admin API (
Browse files Browse the repository at this point in the history
  • Loading branch information
mattisonchao authored Feb 29, 2024
1 parent 72cedb7 commit 6ec473e
Show file tree
Hide file tree
Showing 5 changed files with 246 additions and 10 deletions.
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
Original file line number Diff line number Diff line change
@@ -0,0 +1,175 @@
/*
* 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 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;

import static org.awaitility.Awaitility.await;


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() {
configureTokenAuthentication();
configureDefaultAuthorization();
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);
}
}

}
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 configureTokenAuthentication() {
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 configureDefaultAuthorization() {
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();
}
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
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

0 comments on commit 6ec473e

Please sign in to comment.