You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by xy...@apache.org on 2022/08/02 07:06:56 UTC
[pulsar] branch branch-2.8 updated: [fix][broker] Fix passing incorrect authentication data (#16201) (#16840)
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.8 by this push:
new bc67e203316 [fix][broker] Fix passing incorrect authentication data (#16201) (#16840)
bc67e203316 is described below
commit bc67e203316721d91a3271134279b8c6fd62edcb
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Tue Aug 2 15:06:50 2022 +0800
[fix][broker] Fix passing incorrect authentication data (#16201) (#16840)
### Motivation
#16065 fixes the race condition issue, but introduces a new issue. This issue is triggered when the Proxy and Broker work together, when we use the proxy to request the broker to do lookup/subscribe/produce operation, the broker always uses the original authentication data for authorization, not proxy authentication data, which causes this issue.
### Modification
- Fix passing authentication data, differentiate between original auth data and connected auth data by avoid to use the `getAuthenticationData()`, this method name is easy to cause confusion and can not correctly get the authentication data
(cherry picked from commit 936bbbcc6a4e8cf61547aeedf92e84fb3a089502)
Signed-off-by: Zixuan Liu <no...@gmail.com>
---
.../apache/pulsar/broker/service/ServerCnx.java | 98 ++++++++++++----------
1 file changed, 54 insertions(+), 44 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
index 1ac9bf86562..f1896b20d35 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -364,22 +364,20 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
// ////
private CompletableFuture<Boolean> isTopicOperationAllowed(TopicName topicName, TopicOperation operation,
- AuthenticationDataSource authData) {
+ AuthenticationDataSource authDataSource, AuthenticationDataSource originalAuthDataSource) {
+ if (!service.isAuthorizationEnabled()) {
+ return CompletableFuture.completedFuture(true);
+ }
CompletableFuture<Boolean> isProxyAuthorizedFuture;
- CompletableFuture<Boolean> isAuthorizedFuture;
- if (service.isAuthorizationEnabled()) {
- if (originalPrincipal != null) {
- isProxyAuthorizedFuture = service.getAuthorizationService().allowTopicOperationAsync(
- topicName, operation, originalPrincipal, authData);
- } else {
- isProxyAuthorizedFuture = CompletableFuture.completedFuture(true);
- }
- isAuthorizedFuture = service.getAuthorizationService().allowTopicOperationAsync(
- topicName, operation, authRole, authData);
+ if (originalPrincipal != null) {
+ isProxyAuthorizedFuture = service.getAuthorizationService().allowTopicOperationAsync(
+ topicName, operation, originalPrincipal,
+ originalAuthDataSource != null ? originalAuthDataSource : authDataSource);
} else {
isProxyAuthorizedFuture = CompletableFuture.completedFuture(true);
- isAuthorizedFuture = CompletableFuture.completedFuture(true);
}
+ CompletableFuture<Boolean> isAuthorizedFuture = service.getAuthorizationService().allowTopicOperationAsync(
+ topicName, operation, authRole, authDataSource);
return isProxyAuthorizedFuture.thenCombine(isAuthorizedFuture, (isProxyAuthorized, isAuthorized) -> {
if (!isProxyAuthorized) {
log.warn("OriginalRole {} is not authorized to perform operation {} on topic {}",
@@ -395,27 +393,17 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
private CompletableFuture<Boolean> isTopicOperationAllowed(TopicName topicName, String subscriptionName,
TopicOperation operation) {
- CompletableFuture<Boolean> isProxyAuthorizedFuture;
- CompletableFuture<Boolean> isAuthorizedFuture;
if (service.isAuthorizationEnabled()) {
- AuthenticationDataSource authData =
- new AuthenticationDataSubscription(getAuthenticationData(), subscriptionName);
- return isTopicOperationAllowed(topicName, operation, authData);
+ AuthenticationDataSource authDataSource =
+ new AuthenticationDataSubscription(authenticationData, subscriptionName);
+ AuthenticationDataSource originalAuthDataSource = null;
+ if (originalAuthData != null) {
+ originalAuthDataSource = new AuthenticationDataSubscription(originalAuthData, subscriptionName);
+ }
+ return isTopicOperationAllowed(topicName, operation, authDataSource, originalAuthDataSource);
} else {
- isProxyAuthorizedFuture = CompletableFuture.completedFuture(true);
- isAuthorizedFuture = CompletableFuture.completedFuture(true);
+ return CompletableFuture.completedFuture(true);
}
- return isProxyAuthorizedFuture.thenCombine(isAuthorizedFuture, (isProxyAuthorized, isAuthorized) -> {
- if (!isProxyAuthorized) {
- log.warn("OriginalRole {} is not authorized to perform operation {} on topic {}, subscription {}",
- originalPrincipal, operation, topicName, subscriptionName);
- }
- if (!isAuthorized) {
- log.warn("Role {} is not authorized to perform operation {} on topic {}, subscription {}",
- authRole, operation, topicName, subscriptionName);
- }
- return isProxyAuthorized && isAuthorized;
- });
}
@Override
@@ -443,7 +431,7 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
lookupSemaphore.release();
return;
}
- isTopicOperationAllowed(topicName, TopicOperation.LOOKUP, getAuthenticationData()).thenApply(
+ isTopicOperationAllowed(topicName, TopicOperation.LOOKUP, authenticationData, originalAuthData).thenApply(
isAuthorized -> {
if (isAuthorized) {
lookupTopicAsync(getBrokerService().pulsar(), topicName, authoritative,
@@ -507,7 +495,7 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
lookupSemaphore.release();
return;
}
- isTopicOperationAllowed(topicName, TopicOperation.LOOKUP, getAuthenticationData()).thenApply(
+ isTopicOperationAllowed(topicName, TopicOperation.LOOKUP, authenticationData, originalAuthData).thenApply(
isAuthorized -> {
if (isAuthorized) {
unsafeGetPartitionedTopicMetadataAsync(getBrokerService().pulsar(), topicName)
@@ -1154,7 +1142,7 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
}
CompletableFuture<Boolean> isAuthorizedFuture = isTopicOperationAllowed(
- topicName, TopicOperation.PRODUCE, getAuthenticationData()
+ topicName, TopicOperation.PRODUCE, authenticationData, originalAuthData
);
isAuthorizedFuture.thenApply(isAuthorized -> {
if (isAuthorized) {
@@ -1813,21 +1801,18 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
private CompletableFuture<Boolean> isNamespaceOperationAllowed(NamespaceName namespaceName,
NamespaceOperation operation) {
+ if (!service.isAuthorizationEnabled()) {
+ return CompletableFuture.completedFuture(true);
+ }
CompletableFuture<Boolean> isProxyAuthorizedFuture;
- CompletableFuture<Boolean> isAuthorizedFuture;
- if (service.isAuthorizationEnabled()) {
- if (originalPrincipal != null) {
- isProxyAuthorizedFuture = service.getAuthorizationService().allowNamespaceOperationAsync(
- namespaceName, operation, originalPrincipal, getAuthenticationData());
- } else {
- isProxyAuthorizedFuture = CompletableFuture.completedFuture(true);
- }
- isAuthorizedFuture = service.getAuthorizationService().allowNamespaceOperationAsync(
- namespaceName, operation, authRole, authenticationData);
+ if (originalPrincipal != null) {
+ isProxyAuthorizedFuture = service.getAuthorizationService().allowNamespaceOperationAsync(
+ namespaceName, operation, originalPrincipal, originalAuthData);
} else {
isProxyAuthorizedFuture = CompletableFuture.completedFuture(true);
- isAuthorizedFuture = CompletableFuture.completedFuture(true);
}
+ CompletableFuture<Boolean> isAuthorizedFuture = service.getAuthorizationService().allowNamespaceOperationAsync(
+ namespaceName, operation, authRole, authenticationData);
return isProxyAuthorizedFuture.thenCombine(isAuthorizedFuture, (isProxyAuthorized, isAuthorized) -> {
if (!isProxyAuthorized) {
log.warn("OriginalRole {} is not authorized to perform operation {} on namespace {}",
@@ -2708,4 +2693,29 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
public boolean hasProducers() {
return !producers.isEmpty();
}
+
+ @VisibleForTesting
+ protected String getOriginalPrincipal() {
+ return originalPrincipal;
+ }
+
+ @VisibleForTesting
+ protected AuthenticationDataSource getAuthData() {
+ return authenticationData;
+ }
+
+ @VisibleForTesting
+ protected AuthenticationDataSource getOriginalAuthData() {
+ return originalAuthData;
+ }
+
+ @VisibleForTesting
+ protected AuthenticationState getOriginalAuthState() {
+ return originalAuthState;
+ }
+
+ @VisibleForTesting
+ protected void setAuthRole(String authRole) {
+ this.authRole = authRole;
+ }
}