You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by ma...@apache.org on 2024/11/29 08:30:54 UTC
(pulsar) branch master updated: [fix][broker]: support missing broker level fine-granted permissions (#23637)
This is an automated email from the ASF dual-hosted git repository.
mattisonchao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new eb60d0ab202 [fix][broker]: support missing broker level fine-granted permissions (#23637)
eb60d0ab202 is described below
commit eb60d0ab2022f11ebd2217a32ef60886f757cddf
Author: Qiang Zhao <ma...@apache.org>
AuthorDate: Fri Nov 29 16:30:48 2024 +0800
[fix][broker]: support missing broker level fine-granted permissions (#23637)
---
.../authorization/AuthorizationProvider.java | 10 +
.../broker/authorization/AuthorizationService.java | 23 ++
.../pulsar/broker/admin/impl/BrokersBase.java | 100 +++++++-
.../admin/BrokerEndpointsAuthorizationTest.java | 277 +++++++++++++++++++++
.../pulsar/client/admin/internal/BrokersImpl.java | 2 +-
.../common/policies/data/BrokerOperation.java | 39 +++
6 files changed, 437 insertions(+), 14 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 ffb38f770a9..48386265940 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
@@ -32,6 +32,7 @@ import org.apache.pulsar.client.admin.RevokeTopicPermissionOptions;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.AuthAction;
+import org.apache.pulsar.common.policies.data.BrokerOperation;
import org.apache.pulsar.common.policies.data.NamespaceOperation;
import org.apache.pulsar.common.policies.data.PolicyName;
import org.apache.pulsar.common.policies.data.PolicyOperation;
@@ -383,4 +384,13 @@ public interface AuthorizationProvider extends Closeable {
String.format("getPermissionsAsync on namespaceName %s is not supported by the Authorization",
namespaceName)));
}
+
+ default CompletableFuture<Boolean> allowBrokerOperationAsync(String clusterName,
+ String brokerId,
+ BrokerOperation brokerOperation,
+ String role,
+ AuthenticationDataSource authData) {
+ return FutureUtil.failedFuture(
+ new UnsupportedOperationException("allowBrokerOperationAsync is not supported yet."));
+ }
}
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 2951eb1f297..1348a405b0d 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
@@ -38,6 +38,7 @@ import org.apache.pulsar.client.admin.RevokeTopicPermissionOptions;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.AuthAction;
+import org.apache.pulsar.common.policies.data.BrokerOperation;
import org.apache.pulsar.common.policies.data.NamespaceOperation;
import org.apache.pulsar.common.policies.data.PolicyName;
import org.apache.pulsar.common.policies.data.PolicyOperation;
@@ -544,6 +545,28 @@ public class AuthorizationService {
}
}
+ public CompletableFuture<Boolean> allowBrokerOperationAsync(String clusterName,
+ String brokerId,
+ BrokerOperation brokerOperation,
+ String originalRole,
+ String role,
+ AuthenticationDataSource authData) {
+ if (!isValidOriginalPrincipal(role, originalRole, authData)) {
+ return CompletableFuture.completedFuture(false);
+ }
+
+ if (isProxyRole(role)) {
+ final var isRoleAuthorizedFuture = provider.allowBrokerOperationAsync(clusterName, brokerId,
+ brokerOperation, role, authData);
+ final var isOriginalAuthorizedFuture = provider.allowBrokerOperationAsync(clusterName, brokerId,
+ brokerOperation, originalRole, authData);
+ return isRoleAuthorizedFuture.thenCombine(isOriginalAuthorizedFuture,
+ (isRoleAuthorized, isOriginalAuthorized) -> isRoleAuthorized && isOriginalAuthorized);
+ } else {
+ return provider.allowBrokerOperationAsync(clusterName, brokerId, brokerOperation, role, authData);
+ }
+ }
+
/**
* @deprecated - will be removed after 2.12. Use async variant.
*/
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
index e397dbb64a0..a24a78d8e31 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
@@ -69,6 +69,7 @@ import org.apache.pulsar.common.conf.InternalConfigurationData;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicVersion;
import org.apache.pulsar.common.policies.data.BrokerInfo;
+import org.apache.pulsar.common.policies.data.BrokerOperation;
import org.apache.pulsar.common.policies.data.NamespaceOwnershipStatus;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.ThreadDumpUtil;
@@ -107,7 +108,8 @@ public class BrokersBase extends AdminResource {
@ApiResponse(code = 404, message = "Cluster does not exist: cluster={clustername}") })
public void getActiveBrokers(@Suspended final AsyncResponse asyncResponse,
@PathParam("cluster") String cluster) {
- validateSuperUserAccessAsync()
+ validateBothSuperuserAndBrokerOperation(cluster == null ? pulsar().getConfiguration().getClusterName()
+ : cluster, pulsar().getBrokerId(), BrokerOperation.LIST_BROKERS)
.thenCompose(__ -> validateClusterOwnershipAsync(cluster))
.thenCompose(__ -> pulsar().getLoadManager().get().getAvailableBrokersAsync())
.thenAccept(activeBrokers -> {
@@ -148,7 +150,9 @@ public class BrokersBase extends AdminResource {
@ApiResponse(code = 403, message = "This operation requires super-user access"),
@ApiResponse(code = 404, message = "Leader broker not found") })
public void getLeaderBroker(@Suspended final AsyncResponse asyncResponse) {
- validateSuperUserAccessAsync().thenAccept(__ -> {
+ validateBothSuperuserAndBrokerOperation(pulsar().getConfig().getClusterName(),
+ pulsar().getBrokerId(), BrokerOperation.GET_LEADER_BROKER)
+ .thenAccept(__ -> {
LeaderBroker leaderBroker = pulsar().getLeaderElectionService().getCurrentLeader()
.orElseThrow(() -> new RestException(Status.NOT_FOUND, "Couldn't find leader broker"));
BrokerInfo brokerInfo = BrokerInfo.builder()
@@ -175,7 +179,8 @@ public class BrokersBase extends AdminResource {
public void getOwnedNamespaces(@Suspended final AsyncResponse asyncResponse,
@PathParam("clusterName") String cluster,
@PathParam("brokerId") String brokerId) {
- validateSuperUserAccessAsync()
+ validateBothSuperuserAndBrokerOperation(pulsar().getConfig().getClusterName(),
+ pulsar().getBrokerId(), BrokerOperation.LIST_OWNED_NAMESPACES)
.thenCompose(__ -> maybeRedirectToBroker(brokerId))
.thenCompose(__ -> validateClusterOwnershipAsync(cluster))
.thenCompose(__ -> pulsar().getNamespaceService().getOwnedNameSpacesStatusAsync())
@@ -204,7 +209,8 @@ public class BrokersBase extends AdminResource {
public void updateDynamicConfiguration(@Suspended AsyncResponse asyncResponse,
@PathParam("configName") String configName,
@PathParam("configValue") String configValue) {
- validateSuperUserAccessAsync()
+ validateBothSuperuserAndBrokerOperation(pulsar().getConfig().getClusterName(), pulsar().getBrokerId(),
+ BrokerOperation.UPDATE_DYNAMIC_CONFIGURATION)
.thenCompose(__ -> persistDynamicConfigurationAsync(configName, configValue))
.thenAccept(__ -> {
LOG.info("[{}] Updated Service configuration {}/{}", clientAppId(), configName, configValue);
@@ -228,7 +234,8 @@ public class BrokersBase extends AdminResource {
public void deleteDynamicConfiguration(
@Suspended AsyncResponse asyncResponse,
@PathParam("configName") String configName) {
- validateSuperUserAccessAsync()
+ validateBothSuperuserAndBrokerOperation(pulsar().getConfig().getClusterName(), pulsar().getBrokerId(),
+ BrokerOperation.DELETE_DYNAMIC_CONFIGURATION)
.thenCompose(__ -> internalDeleteDynamicConfigurationOnMetadataAsync(configName))
.thenAccept(__ -> {
LOG.info("[{}] Successfully to delete dynamic configuration {}", clientAppId(), configName);
@@ -249,7 +256,8 @@ public class BrokersBase extends AdminResource {
@ApiResponse(code = 404, message = "Configuration not found"),
@ApiResponse(code = 500, message = "Internal server error")})
public void getAllDynamicConfigurations(@Suspended AsyncResponse asyncResponse) {
- validateSuperUserAccessAsync()
+ validateBothSuperuserAndBrokerOperation(pulsar().getConfig().getClusterName(), pulsar().getBrokerId(),
+ BrokerOperation.LIST_DYNAMIC_CONFIGURATIONS)
.thenCompose(__ -> dynamicConfigurationResources().getDynamicConfigurationAsync())
.thenAccept(configOpt -> asyncResponse.resume(configOpt.orElseGet(Collections::emptyMap)))
.exceptionally(ex -> {
@@ -266,7 +274,8 @@ public class BrokersBase extends AdminResource {
@ApiResponses(value = {
@ApiResponse(code = 403, message = "You don't have admin permission to get configuration")})
public void getDynamicConfigurationName(@Suspended AsyncResponse asyncResponse) {
- validateSuperUserAccessAsync()
+ validateBothSuperuserAndBrokerOperation(pulsar().getConfig().getClusterName(), pulsar().getBrokerId(),
+ BrokerOperation.LIST_DYNAMIC_CONFIGURATIONS)
.thenAccept(__ -> asyncResponse.resume(pulsar().getBrokerService().getDynamicConfiguration()))
.exceptionally(ex -> {
LOG.error("[{}] Failed to get all dynamic configuration names.", clientAppId(), ex);
@@ -281,7 +290,8 @@ public class BrokersBase extends AdminResource {
response = String.class, responseContainer = "Map")
@ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission") })
public void getRuntimeConfiguration(@Suspended AsyncResponse asyncResponse) {
- validateSuperUserAccessAsync()
+ validateBothSuperuserAndBrokerOperation(pulsar().getConfig().getClusterName(), pulsar().getBrokerId(),
+ BrokerOperation.LIST_RUNTIME_CONFIGURATIONS)
.thenAccept(__ -> asyncResponse.resume(pulsar().getBrokerService().getRuntimeConfiguration()))
.exceptionally(ex -> {
LOG.error("[{}] Failed to get runtime configuration.", clientAppId(), ex);
@@ -322,7 +332,8 @@ public class BrokersBase extends AdminResource {
@ApiOperation(value = "Get the internal configuration data", response = InternalConfigurationData.class)
@ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission") })
public void getInternalConfigurationData(@Suspended AsyncResponse asyncResponse) {
- validateSuperUserAccessAsync()
+ validateBothSuperuserAndBrokerOperation(pulsar().getConfig().getClusterName(), pulsar().getBrokerId(),
+ BrokerOperation.GET_INTERNAL_CONFIGURATION_DATA)
.thenAccept(__ -> asyncResponse.resume(pulsar().getInternalConfigurationData()))
.exceptionally(ex -> {
LOG.error("[{}] Failed to get internal configuration data.", clientAppId(), ex);
@@ -339,7 +350,8 @@ public class BrokersBase extends AdminResource {
@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 500, message = "Internal server error")})
public void backlogQuotaCheck(@Suspended AsyncResponse asyncResponse) {
- validateSuperUserAccessAsync()
+ validateBothSuperuserAndBrokerOperation(pulsar().getConfig().getClusterName(), pulsar().getBrokerId(),
+ BrokerOperation.CHECK_BACKLOG_QUOTA)
.thenAcceptAsync(__ -> {
pulsar().getBrokerService().monitorBacklogQuota();
asyncResponse.resume(Response.noContent().build());
@@ -378,7 +390,8 @@ public class BrokersBase extends AdminResource {
@ApiParam(value = "Topic Version")
@QueryParam("topicVersion") TopicVersion topicVersion,
@QueryParam("brokerId") String brokerId) {
- validateSuperUserAccessAsync()
+ validateBothSuperuserAndBrokerOperation(pulsar().getConfig().getClusterName(), StringUtils.isBlank(brokerId)
+ ? pulsar().getBrokerId() : brokerId, BrokerOperation.HEALTH_CHECK)
.thenAccept(__ -> checkDeadlockedThreads())
.thenCompose(__ -> maybeRedirectToBroker(
StringUtils.isBlank(brokerId) ? pulsar().getBrokerId() : brokerId))
@@ -596,8 +609,9 @@ public class BrokersBase extends AdminResource {
@QueryParam("forcedTerminateTopic") @DefaultValue("true") boolean forcedTerminateTopic,
@Suspended final AsyncResponse asyncResponse
) {
- validateSuperUserAccess();
- doShutDownBrokerGracefullyAsync(maxConcurrentUnloadPerSec, forcedTerminateTopic)
+ validateBothSuperuserAndBrokerOperation(pulsar().getConfig().getClusterName(), pulsar().getBrokerId(),
+ BrokerOperation.SHUTDOWN)
+ .thenCompose(__ -> doShutDownBrokerGracefullyAsync(maxConcurrentUnloadPerSec, forcedTerminateTopic))
.thenAccept(__ -> {
LOG.info("[{}] Successfully shutdown broker gracefully", clientAppId());
asyncResponse.resume(Response.noContent().build());
@@ -614,5 +628,65 @@ public class BrokersBase extends AdminResource {
pulsar().getBrokerService().unloadNamespaceBundlesGracefully(maxConcurrentUnloadPerSec, forcedTerminateTopic);
return pulsar().closeAsync();
}
+
+
+ private CompletableFuture<Void> validateBothSuperuserAndBrokerOperation(String cluster, String brokerId,
+ BrokerOperation operation) {
+ final var superUserAccessValidation = validateSuperUserAccessAsync();
+ final var brokerOperationValidation = validateBrokerOperationAsync(cluster, brokerId, operation);
+ return FutureUtil.waitForAll(List.of(superUserAccessValidation, brokerOperationValidation))
+ .handle((result, err) -> {
+ if (!superUserAccessValidation.isCompletedExceptionally()
+ || !brokerOperationValidation.isCompletedExceptionally()) {
+ return null;
+ }
+ if (LOG.isDebugEnabled()) {
+ Throwable superUserValidationException = null;
+ try {
+ superUserAccessValidation.join();
+ } catch (Throwable ex) {
+ superUserValidationException = FutureUtil.unwrapCompletionException(ex);
+ }
+ Throwable brokerOperationValidationException = null;
+ try {
+ brokerOperationValidation.join();
+ } catch (Throwable ex) {
+ brokerOperationValidationException = FutureUtil.unwrapCompletionException(ex);
+ }
+ LOG.debug("validateBothSuperuserAndBrokerOperation failed."
+ + " originalPrincipal={} clientAppId={} operation={} broker={} "
+ + "superuserValidationError={} brokerOperationValidationError={}",
+ originalPrincipal(), clientAppId(), operation.toString(), brokerId,
+ superUserValidationException, brokerOperationValidationException);
+ }
+ throw new RestException(Status.UNAUTHORIZED,
+ String.format("Unauthorized to validateBothSuperuserAndBrokerOperation for"
+ + " originalPrincipal [%s] and clientAppId [%s] "
+ + "about operation [%s] on broker [%s]",
+ originalPrincipal(), clientAppId(), operation.toString(), brokerId));
+ });
+ }
+
+
+ private CompletableFuture<Void> validateBrokerOperationAsync(String cluster, String brokerId,
+ BrokerOperation operation) {
+ final var pulsar = pulsar();
+ if (pulsar.getBrokerService().isAuthenticationEnabled()
+ && pulsar.getBrokerService().isAuthorizationEnabled()) {
+ return pulsar.getBrokerService().getAuthorizationService()
+ .allowBrokerOperationAsync(cluster, brokerId, operation, originalPrincipal(),
+ clientAppId(), clientAuthData())
+ .thenAccept(isAuthorized -> {
+ if (!isAuthorized) {
+ throw new RestException(Status.UNAUTHORIZED,
+ String.format("Unauthorized to validateBrokerOperation for"
+ + " originalPrincipal [%s] and clientAppId [%s] "
+ + "about operation [%s] on broker [%s]",
+ originalPrincipal(), clientAppId(), operation.toString(), brokerId));
+ }
+ });
+ }
+ return CompletableFuture.completedFuture(null);
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/BrokerEndpointsAuthorizationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/BrokerEndpointsAuthorizationTest.java
new file mode 100644
index 00000000000..ef66f005b3c
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/BrokerEndpointsAuthorizationTest.java
@@ -0,0 +1,277 @@
+/*
+ * 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.mockito.Mockito.any;
+import static org.mockito.Mockito.eq;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import lombok.SneakyThrows;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.pulsar.broker.authorization.AuthorizationService;
+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.naming.TopicVersion;
+import org.apache.pulsar.common.policies.data.BrokerOperation;
+import org.apache.pulsar.security.MockedPulsarStandalone;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-admin")
+public class BrokerEndpointsAuthorizationTest extends MockedPulsarStandalone {
+ private AuthorizationService orignalAuthorizationService;
+ private AuthorizationService spyAuthorizationService;
+
+ private PulsarAdmin superUserAdmin;
+ private PulsarAdmin nobodyAdmin;
+
+ @SneakyThrows
+ @BeforeClass(alwaysRun = true)
+ public void setup() {
+ configureTokenAuthentication();
+ configureDefaultAuthorization();
+ start();
+ this.superUserAdmin = PulsarAdmin.builder()
+ .serviceHttpUrl(getPulsarService().getWebServiceAddress())
+ .authentication(new AuthenticationToken(SUPER_USER_TOKEN))
+ .build();
+ this.nobodyAdmin = PulsarAdmin.builder()
+ .serviceHttpUrl(getPulsarService().getWebServiceAddress())
+ .authentication(new AuthenticationToken(NOBODY_TOKEN))
+ .build();
+ }
+
+ @BeforeMethod(alwaysRun = true)
+ public void before() throws IllegalAccessException {
+ orignalAuthorizationService = getPulsarService().getBrokerService().getAuthorizationService();
+ spyAuthorizationService = spy(orignalAuthorizationService);
+ FieldUtils.writeField(getPulsarService().getBrokerService(), "authorizationService",
+ spyAuthorizationService, true);
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void after() throws IllegalAccessException {
+ if (orignalAuthorizationService != null) {
+ FieldUtils.writeField(getPulsarService().getBrokerService(), "authorizationService", orignalAuthorizationService, true);
+ }
+ }
+
+ @SneakyThrows
+ @AfterClass(alwaysRun = true)
+ public void cleanup() {
+ if (superUserAdmin != null) {
+ superUserAdmin.close();
+ superUserAdmin = null;
+ }
+ spyAuthorizationService = null;
+ orignalAuthorizationService = null;
+ super.close();
+ }
+
+ @Test
+ public void testGetActiveBroker() throws PulsarAdminException {
+ superUserAdmin.brokers().getActiveBrokers();
+ final String brokerId = getPulsarService().getBrokerId();
+ final String clusterName = getPulsarService().getConfiguration().getClusterName();
+ // test allow broker operation
+ verify(spyAuthorizationService)
+ .allowBrokerOperationAsync(eq(clusterName), eq(brokerId), eq(BrokerOperation.LIST_BROKERS), any(), any(), any());
+ // fallback to superuser
+ verify(spyAuthorizationService).isSuperUser(any(), any());
+
+ // ---- test nobody
+ Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, () -> nobodyAdmin.brokers().getActiveBrokers());
+ }
+
+ @Test
+ public void testGetActiveBrokerWithCluster() throws PulsarAdminException {
+ final String clusterName = getPulsarService().getConfiguration().getClusterName();
+ superUserAdmin.brokers().getActiveBrokers(clusterName);
+ final String brokerId = getPulsarService().getBrokerId();
+ // test allow broker operation
+ verify(spyAuthorizationService)
+ .allowBrokerOperationAsync(eq(clusterName), eq(brokerId), eq(BrokerOperation.LIST_BROKERS), any(), any(), any());
+ // fallback to superuser
+ verify(spyAuthorizationService).isSuperUser(any(), any());
+
+ // ---- test nobody
+ Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, () -> nobodyAdmin.brokers().getActiveBrokers(clusterName));
+ }
+
+ @Test
+ public void testGetLeaderBroker() throws PulsarAdminException {
+ superUserAdmin.brokers().getLeaderBroker();
+ final String clusterName = getPulsarService().getConfiguration().getClusterName();
+ final String brokerId = getPulsarService().getBrokerId();
+ // test allow broker operation
+ verify(spyAuthorizationService)
+ .allowBrokerOperationAsync(eq(clusterName), eq(brokerId), eq(BrokerOperation.GET_LEADER_BROKER), any(), any(), any());
+ // fallback to superuser
+ verify(spyAuthorizationService).isSuperUser(any(), any());
+
+ // ---- test nobody
+ Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, () -> nobodyAdmin.brokers().getLeaderBroker());
+ }
+
+ @Test
+ public void testGetOwnedNamespaces() throws PulsarAdminException {
+ final String clusterName = getPulsarService().getConfiguration().getClusterName();
+ final String brokerId = getPulsarService().getBrokerId();
+ superUserAdmin.brokers().getOwnedNamespaces(clusterName, brokerId);
+ // test allow broker operation
+ verify(spyAuthorizationService)
+ .allowBrokerOperationAsync(eq(clusterName), eq(brokerId), eq(BrokerOperation.LIST_OWNED_NAMESPACES), any(), any(), any());
+ // fallback to superuser
+ verify(spyAuthorizationService).isSuperUser(any(), any());
+
+ // ---- test nobody
+ Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, () -> nobodyAdmin.brokers().getOwnedNamespaces(clusterName, brokerId));
+ }
+
+ @Test
+ public void testUpdateDynamicConfiguration() throws PulsarAdminException {
+ final String clusterName = getPulsarService().getConfiguration().getClusterName();
+ final String brokerId = getPulsarService().getBrokerId();
+ superUserAdmin.brokers().updateDynamicConfiguration("maxTenants", "10");
+ // test allow broker operation
+ verify(spyAuthorizationService)
+ .allowBrokerOperationAsync(eq(clusterName), eq(brokerId), eq(BrokerOperation.UPDATE_DYNAMIC_CONFIGURATION), any(), any(), any());
+ // fallback to superuser
+ verify(spyAuthorizationService).isSuperUser(any(), any());
+
+ // ---- test nobody
+ Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, () -> nobodyAdmin.brokers().updateDynamicConfiguration("maxTenants", "10"));
+ }
+
+ @Test
+ public void testDeleteDynamicConfiguration() throws PulsarAdminException {
+ final String clusterName = getPulsarService().getConfiguration().getClusterName();
+ final String brokerId = getPulsarService().getBrokerId();
+ superUserAdmin.brokers().deleteDynamicConfiguration("maxTenants");
+ // test allow broker operation
+ verify(spyAuthorizationService)
+ .allowBrokerOperationAsync(eq(clusterName), eq(brokerId), eq(BrokerOperation.DELETE_DYNAMIC_CONFIGURATION), any(), any(), any());
+ // fallback to superuser
+ verify(spyAuthorizationService).isSuperUser(any(), any());
+
+ // ---- test nobody
+ Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, () -> nobodyAdmin.brokers().deleteDynamicConfiguration("maxTenants"));
+ }
+
+
+ @Test
+ public void testGetAllDynamicConfiguration() throws PulsarAdminException {
+ final String clusterName = getPulsarService().getConfiguration().getClusterName();
+ final String brokerId = getPulsarService().getBrokerId();
+ superUserAdmin.brokers().getAllDynamicConfigurations();
+ // test allow broker operation
+ verify(spyAuthorizationService)
+ .allowBrokerOperationAsync(eq(clusterName), eq(brokerId), eq(BrokerOperation.LIST_DYNAMIC_CONFIGURATIONS), any(), any(), any());
+ // fallback to superuser
+ verify(spyAuthorizationService).isSuperUser(any(), any());
+
+ // ---- test nobody
+ Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, () -> nobodyAdmin.brokers().getAllDynamicConfigurations());
+ }
+
+
+ @Test
+ public void testGetDynamicConfigurationName() throws PulsarAdminException {
+ final String clusterName = getPulsarService().getConfiguration().getClusterName();
+ final String brokerId = getPulsarService().getBrokerId();
+ superUserAdmin.brokers().getDynamicConfigurationNames();
+ // test allow broker operation
+ verify(spyAuthorizationService)
+ .allowBrokerOperationAsync(eq(clusterName), eq(brokerId), eq(BrokerOperation.LIST_DYNAMIC_CONFIGURATIONS), any(), any(), any());
+ // fallback to superuser
+ verify(spyAuthorizationService).isSuperUser(any(), any());
+
+ // ---- test nobody
+ Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, () -> nobodyAdmin.brokers().getDynamicConfigurationNames());
+ }
+
+
+ @Test
+ public void testGetRuntimeConfiguration() throws PulsarAdminException {
+ final String clusterName = getPulsarService().getConfiguration().getClusterName();
+ final String brokerId = getPulsarService().getBrokerId();
+ superUserAdmin.brokers().getRuntimeConfigurations();
+ // test allow broker operation
+ verify(spyAuthorizationService)
+ .allowBrokerOperationAsync(eq(clusterName), eq(brokerId), eq(BrokerOperation.LIST_RUNTIME_CONFIGURATIONS), any(), any(), any());
+ // fallback to superuser
+ verify(spyAuthorizationService).isSuperUser(any(), any());
+
+ // ---- test nobody
+ Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, () -> nobodyAdmin.brokers().getRuntimeConfigurations());
+ }
+
+
+ @Test
+ public void testGetInternalConfigurationData() throws PulsarAdminException {
+ final String clusterName = getPulsarService().getConfiguration().getClusterName();
+ final String brokerId = getPulsarService().getBrokerId();
+ superUserAdmin.brokers().getInternalConfigurationData();
+ // test allow broker operation
+ verify(spyAuthorizationService)
+ .allowBrokerOperationAsync(eq(clusterName), eq(brokerId), eq(BrokerOperation.GET_INTERNAL_CONFIGURATION_DATA), any(), any(), any());
+ // fallback to superuser
+ verify(spyAuthorizationService).isSuperUser(any(), any());
+
+ // ---- test nobody
+ Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, () -> nobodyAdmin.brokers().getInternalConfigurationData());
+ }
+
+
+ @Test
+ public void testBacklogQuotaCheck() throws PulsarAdminException {
+ final String clusterName = getPulsarService().getConfiguration().getClusterName();
+ final String brokerId = getPulsarService().getBrokerId();
+ superUserAdmin.brokers().backlogQuotaCheck();
+ // test allow broker operation
+ verify(spyAuthorizationService)
+ .allowBrokerOperationAsync(eq(clusterName), eq(brokerId), eq(BrokerOperation.CHECK_BACKLOG_QUOTA), any(), any(), any());
+ // fallback to superuser
+ verify(spyAuthorizationService).isSuperUser(any(), any());
+
+ // ---- test nobody
+ Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, () -> nobodyAdmin.brokers().backlogQuotaCheck());
+ }
+
+ @Test
+ public void testHealthCheck() throws PulsarAdminException {
+ final String clusterName = getPulsarService().getConfiguration().getClusterName();
+ final String brokerId = getPulsarService().getBrokerId();
+ superUserAdmin.brokers().healthcheck(TopicVersion.V2);
+ // test allow broker operation
+ verify(spyAuthorizationService)
+ .allowBrokerOperationAsync(eq(clusterName), eq(brokerId), eq(BrokerOperation.HEALTH_CHECK), any(), any(), any());
+ // fallback to superuser
+ verify(spyAuthorizationService).isSuperUser(any(), any());
+
+ // ---- test nobody
+ Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class, () -> nobodyAdmin.brokers().healthcheck(TopicVersion.V2));
+ }
+}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokersImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokersImpl.java
index 35b261b196e..b0cd3edeb21 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokersImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokersImpl.java
@@ -162,7 +162,7 @@ public class BrokersImpl extends BaseResource implements Brokers {
@Override
public CompletableFuture<Void> backlogQuotaCheckAsync() {
- WebTarget path = adminBrokers.path("backlogQuotaCheck");
+ WebTarget path = adminBrokers.path("backlog-quota-check");
return asyncGetRequest(path, new FutureCallback<Void>() {});
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/BrokerOperation.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/BrokerOperation.java
new file mode 100644
index 00000000000..de053fea6ad
--- /dev/null
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/BrokerOperation.java
@@ -0,0 +1,39 @@
+/*
+ * 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.common.policies.data;
+
+public enum BrokerOperation {
+ LIST_BROKERS,
+ GET_BROKER,
+
+ GET_LEADER_BROKER,
+ LIST_OWNED_NAMESPACES,
+
+ LIST_DYNAMIC_CONFIGURATIONS,
+ UPDATE_DYNAMIC_CONFIGURATION,
+ DELETE_DYNAMIC_CONFIGURATION,
+
+ LIST_RUNTIME_CONFIGURATIONS,
+
+ GET_INTERNAL_CONFIGURATION_DATA,
+
+ CHECK_BACKLOG_QUOTA,
+ HEALTH_CHECK,
+ SHUTDOWN
+}