You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2024/03/14 04:27:37 UTC
(pulsar) branch branch-2.11 updated: [improve][sec] Align some namespace level policy authorisation check (#21640) (#22248)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.11 by this push:
new 44ae31a85ca [improve][sec] Align some namespace level policy authorisation check (#21640) (#22248)
44ae31a85ca is described below
commit 44ae31a85ca05086d7c4cf1e60708c365a43d4b3
Author: Jiwei Guo <te...@apache.org>
AuthorDate: Thu Mar 14 12:27:30 2024 +0800
[improve][sec] Align some namespace level policy authorisation check (#21640) (#22248)
---
.../pulsar/broker/admin/impl/NamespacesBase.java | 30 +++++++++++++---------
.../apache/pulsar/broker/admin/v2/Namespaces.java | 3 ++-
2 files changed, 20 insertions(+), 13 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
index 610b799fba6..0a602a631c2 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
@@ -1349,7 +1349,8 @@ public abstract class NamespacesBase extends AdminResource {
protected CompletableFuture<Void> internalSetPublishRateAsync(PublishRate maxPublishMessageRate) {
log.info("[{}] Set namespace publish-rate {}/{}", clientAppId(), namespaceName, maxPublishMessageRate);
- return validateSuperUserAccessAsync().thenCompose(__ -> updatePoliciesAsync(namespaceName, policies -> {
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.RATE, PolicyOperation.WRITE)
+ .thenCompose(__ -> updatePoliciesAsync(namespaceName, policies -> {
policies.publishMaxMessageRate.put(pulsar().getConfiguration().getClusterName(), maxPublishMessageRate);
log.info("[{}] Successfully updated the publish_max_message_rate for cluster on namespace {}",
clientAppId(), namespaceName);
@@ -1378,7 +1379,8 @@ public abstract class NamespacesBase extends AdminResource {
protected CompletableFuture<Void> internalRemovePublishRateAsync() {
log.info("[{}] Remove namespace publish-rate {}/{}", clientAppId(), namespaceName, topicName);
- return validateSuperUserAccessAsync().thenCompose(__ -> updatePoliciesAsync(namespaceName, policies -> {
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.RATE, PolicyOperation.WRITE)
+ .thenCompose(__ -> updatePoliciesAsync(namespaceName, policies -> {
if (policies.publishMaxMessageRate != null) {
policies.publishMaxMessageRate.remove(pulsar().getConfiguration().getClusterName());
}
@@ -1417,7 +1419,8 @@ public abstract class NamespacesBase extends AdminResource {
@SuppressWarnings("deprecation")
protected CompletableFuture<Void> internalSetTopicDispatchRateAsync(DispatchRateImpl dispatchRate) {
log.info("[{}] Set namespace dispatch-rate {}/{}", clientAppId(), namespaceName, dispatchRate);
- return validateSuperUserAccessAsync().thenCompose(__ -> updatePoliciesAsync(namespaceName, policies -> {
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.RATE, PolicyOperation.WRITE)
+ .thenCompose(__ -> updatePoliciesAsync(namespaceName, policies -> {
policies.topicDispatchRate.put(pulsar().getConfiguration().getClusterName(), dispatchRate);
policies.clusterDispatchRate.put(pulsar().getConfiguration().getClusterName(), dispatchRate);
log.info("[{}] Successfully updated the dispatchRate for cluster on namespace {}", clientAppId(),
@@ -1444,7 +1447,8 @@ public abstract class NamespacesBase extends AdminResource {
}
protected CompletableFuture<Void> internalDeleteTopicDispatchRateAsync() {
- return validateSuperUserAccessAsync().thenCompose(__ -> updatePoliciesAsync(namespaceName, policies -> {
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.RATE, PolicyOperation.WRITE)
+ .thenCompose(__ -> updatePoliciesAsync(namespaceName, policies -> {
policies.topicDispatchRate.remove(pulsar().getConfiguration().getClusterName());
policies.clusterDispatchRate.remove(pulsar().getConfiguration().getClusterName());
log.info("[{}] Successfully delete the dispatchRate for cluster on namespace {}", clientAppId(),
@@ -1461,7 +1465,7 @@ public abstract class NamespacesBase extends AdminResource {
}
protected CompletableFuture<Void> internalSetSubscriptionDispatchRateAsync(DispatchRateImpl dispatchRate) {
- return validateSuperUserAccessAsync()
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.RATE, PolicyOperation.WRITE)
.thenCompose(__ -> updatePoliciesAsync(namespaceName, policies -> {
policies.subscriptionDispatchRate.put(pulsar().getConfiguration().getClusterName(), dispatchRate);
log.info("[{}] Successfully updated the subscriptionDispatchRate for cluster on namespace {}",
@@ -1471,7 +1475,7 @@ public abstract class NamespacesBase extends AdminResource {
}
protected CompletableFuture<Void> internalDeleteSubscriptionDispatchRateAsync() {
- return validateSuperUserAccessAsync()
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.RATE, PolicyOperation.WRITE)
.thenCompose(__ -> updatePoliciesAsync(namespaceName, policies -> {
policies.subscriptionDispatchRate.remove(pulsar().getConfiguration().getClusterName());
log.info("[{}] Successfully delete the subscriptionDispatchRate for cluster on namespace {}",
@@ -1489,7 +1493,8 @@ public abstract class NamespacesBase extends AdminResource {
protected CompletableFuture<Void> internalSetSubscribeRateAsync(SubscribeRate subscribeRate) {
log.info("[{}] Set namespace subscribe-rate {}/{}", clientAppId(), namespaceName, subscribeRate);
- return validateSuperUserAccessAsync().thenCompose(__ -> updatePoliciesAsync(namespaceName, policies -> {
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.RATE, PolicyOperation.WRITE)
+ .thenCompose(__ -> updatePoliciesAsync(namespaceName, policies -> {
policies.clusterSubscribeRate.put(pulsar().getConfiguration().getClusterName(), subscribeRate);
log.info("[{}] Successfully updated the subscribeRate for cluster on namespace {}", clientAppId(),
namespaceName);
@@ -1498,7 +1503,8 @@ public abstract class NamespacesBase extends AdminResource {
}
protected CompletableFuture<Void> internalDeleteSubscribeRateAsync() {
- return validateSuperUserAccessAsync().thenCompose(__ -> updatePoliciesAsync(namespaceName, policies -> {
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.RATE, PolicyOperation.WRITE)
+ .thenCompose(__ -> updatePoliciesAsync(namespaceName, policies -> {
policies.clusterSubscribeRate.remove(pulsar().getConfiguration().getClusterName());
log.info("[{}] Successfully delete the subscribeRate for cluster on namespace {}", clientAppId(),
namespaceName);
@@ -1514,7 +1520,7 @@ public abstract class NamespacesBase extends AdminResource {
}
protected void internalRemoveReplicatorDispatchRate() {
- validateSuperUserAccess();
+ validateNamespacePolicyOperation(namespaceName, PolicyName.REPLICATION_RATE, PolicyOperation.WRITE);
try {
updatePolicies(namespaceName, policies -> {
policies.replicatorDispatchRate.remove(pulsar().getConfiguration().getClusterName());
@@ -1530,7 +1536,7 @@ public abstract class NamespacesBase extends AdminResource {
}
protected void internalSetReplicatorDispatchRate(DispatchRateImpl dispatchRate) {
- validateSuperUserAccess();
+ validateNamespacePolicyOperation(namespaceName, PolicyName.REPLICATION_RATE, PolicyOperation.WRITE);
log.info("[{}] Set namespace replicator dispatch-rate {}/{}", clientAppId(), namespaceName, dispatchRate);
try {
updatePolicies(namespaceName, policies -> {
@@ -1906,7 +1912,7 @@ public abstract class NamespacesBase extends AdminResource {
}
protected void internalSetInactiveTopic(InactiveTopicPolicies inactiveTopicPolicies) {
- validateSuperUserAccess();
+ validateNamespacePolicyOperation(namespaceName, PolicyName.INACTIVE_TOPIC, PolicyOperation.WRITE);
validatePoliciesReadOnlyAccess();
internalSetPolicies("inactive_topic_policies", inactiveTopicPolicies);
}
@@ -2322,7 +2328,7 @@ public abstract class NamespacesBase extends AdminResource {
}
protected void internalSetMaxSubscriptionsPerTopic(Integer maxSubscriptionsPerTopic){
- validateSuperUserAccess();
+ validateNamespacePolicyOperation(namespaceName, PolicyName.MAX_SUBSCRIPTIONS, PolicyOperation.WRITE);
validatePoliciesReadOnlyAccess();
if (maxSubscriptionsPerTopic != null && maxSubscriptionsPerTopic < 0) {
throw new RestException(Status.PRECONDITION_FAILED,
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
index 78d1288ede5..e72d01b79ce 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
@@ -463,7 +463,8 @@ public class Namespaces extends NamespacesBase {
@PathParam("tenant") String tenant,
@PathParam("namespace") String namespace) {
validateNamespaceName(tenant, namespace);
- validateAdminAccessForTenantAsync(tenant)
+ validateNamespacePolicyOperationAsync(namespaceName, PolicyName.SUBSCRIPTION_EXPIRATION_TIME,
+ PolicyOperation.READ)
.thenCompose(__ -> getNamespacePoliciesAsync(namespaceName))
.thenAccept(policies -> asyncResponse.resume(policies.subscription_expiration_time_minutes))
.exceptionally(ex -> {