You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/09/23 12:37:05 UTC
(pulsar) 15/21: [fix][broker] Check topic permissions for partitions and subscriptions added to a transaction (#26644)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 2407c71052542f1916f0e489c481b9244ab8e14c
Author: kannar <ka...@gmail.com>
AuthorDate: Sat Sep 19 16:07:27 2026 +0200
[fix][broker] Check topic permissions for partitions and subscriptions added to a transaction (#26644)
(cherry picked from commit 28eea815e5c223491537f69dbf2b52a9b5563075)
---
.../broker/service/BrokerServiceException.java | 8 +++
.../apache/pulsar/broker/service/ServerCnx.java | 57 +++++++++++++--
.../pulsar/broker/service/ServerCnxTest.java | 80 ++++++++++++++++++++++
3 files changed, 141 insertions(+), 4 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerServiceException.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerServiceException.java
index 43a95cc86016..95c5550d3b21 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerServiceException.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerServiceException.java
@@ -188,6 +188,12 @@ public class BrokerServiceException extends Exception {
}
}
+ public static class NotAuthorizedException extends BrokerServiceException {
+ public NotAuthorizedException(String msg) {
+ super(msg);
+ }
+ }
+
public static class SubscriptionInvalidCursorPosition extends BrokerServiceException {
public SubscriptionInvalidCursorPosition(String msg) {
super(msg);
@@ -277,6 +283,8 @@ public class BrokerServiceException extends Exception {
return ServerError.InvalidTxnStatus;
} else if (t instanceof NotAllowedException) {
return ServerError.NotAllowedError;
+ } else if (t instanceof NotAuthorizedException) {
+ return ServerError.AuthorizationError;
} else if (t instanceof ProducerFencedException) {
return ServerError.ProducerFenced;
} else if (t instanceof TransactionConflictException) {
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 bb45f53cc3ba..0b8c55520bd6 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
@@ -3009,9 +3009,10 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
if (!isOwner) {
return failedFutureTxnNotOwned(txnID);
}
- return transactionMetadataStoreService
- .addProducedPartitionToTxn(txnID, partitionsList);
+ return checkTxnPartitionsAuthorized(partitionsList);
})
+ .thenCompose(__ -> transactionMetadataStoreService
+ .addProducedPartitionToTxn(txnID, partitionsList))
.whenComplete((v, ex) -> {
if (ex == null) {
if (log.isDebugEnabled()) {
@@ -3032,6 +3033,53 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
});
}
+ /**
+ * The transaction coordinator ends a transaction on its registered partitions with its own identity, so a
+ * client may only register partitions it could produce to.
+ */
+ private CompletableFuture<Void> checkTxnPartitionsAuthorized(List<String> partitions) {
+ List<CompletableFuture<Void>> checks = new ArrayList<>(partitions.size());
+ for (String partition : partitions) {
+ checks.add(checkTxnParticipantAuthorized(partition, null, TopicOperation.PRODUCE));
+ }
+ return FutureUtil.waitForAll(checks);
+ }
+
+ /**
+ * The transaction coordinator ends a transaction on its registered subscriptions with its own identity, so a
+ * client may only register subscriptions it could consume from.
+ */
+ private CompletableFuture<Void> checkTxnSubscriptionsAuthorized(
+ List<org.apache.pulsar.common.api.proto.Subscription> subscriptions) {
+ List<CompletableFuture<Void>> checks = new ArrayList<>(subscriptions.size());
+ for (org.apache.pulsar.common.api.proto.Subscription subscription : subscriptions) {
+ checks.add(checkTxnParticipantAuthorized(subscription.getTopic(), subscription.getSubscription(),
+ TopicOperation.CONSUME));
+ }
+ return FutureUtil.waitForAll(checks);
+ }
+
+ private CompletableFuture<Void> checkTxnParticipantAuthorized(String topic, String subscription,
+ TopicOperation operation) {
+ if (!service.isAuthorizationEnabled()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ final TopicName topicName;
+ try {
+ topicName = TopicName.get(topic);
+ } catch (IllegalArgumentException e) {
+ return CompletableFuture.failedFuture(new BrokerServiceException.NotAllowedException(
+ "Invalid topic name " + topic + ": " + e.getMessage()));
+ }
+ CompletableFuture<Boolean> isAuthorized = subscription == null
+ ? isTopicOperationAllowed(topicName, operation, authenticationData, originalAuthData)
+ : isTopicOperationAllowed(topicName, subscription, operation);
+ return isAuthorized.thenCompose(authorized -> authorized
+ ? CompletableFuture.<Void>completedFuture(null)
+ : CompletableFuture.failedFuture(new BrokerServiceException.NotAuthorizedException(
+ "Client is not authorized to " + operation + " on " + topic)));
+ }
+
private CompletableFuture<Void> failedFutureTxnNotOwned(TxnID txnID) {
String msg = String.format(
"Client (%s) is neither the owner of the transaction %s nor a super user",
@@ -3354,9 +3402,10 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
if (!isOwner) {
return failedFutureTxnNotOwned(txnID);
}
- return transactionMetadataStoreService.addAckedPartitionToTxn(txnID,
- MLTransactionMetadataStore.subscriptionToTxnSubscription(subscriptionsList));
+ return checkTxnSubscriptionsAuthorized(subscriptionsList);
})
+ .thenCompose(__ -> transactionMetadataStoreService.addAckedPartitionToTxn(txnID,
+ MLTransactionMetadataStore.subscriptionToTxnSubscription(subscriptionsList)))
.whenComplete((v, ex) -> {
if (ex == null) {
if (log.isDebugEnabled()) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
index 1c866456eec7..726f71f832b3 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
@@ -3818,6 +3818,86 @@ public class ServerCnxTest {
channel.finish();
}
+ private AuthorizationService mockTopicAuthorization(boolean allowed) {
+ AuthorizationService authorizationService = mock(AuthorizationService.class);
+ doReturn(CompletableFuture.completedFuture(allowed)).when(authorizationService)
+ .allowTopicOperationAsync(Mockito.any(),
+ Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any());
+ doReturn(authorizationService).when(brokerService).getAuthorizationService();
+ svcConfig.setAuthenticationEnabled(true);
+ svcConfig.setAuthorizationEnabled(true);
+ return authorizationService;
+ }
+
+ @Test(timeOut = 30000)
+ public void testAddPartitionToTxnRefusedWhenOnePartitionIsDenied() throws Exception {
+ AuthorizationService authorizationService = mockTopicAuthorization(true);
+ doReturn(CompletableFuture.completedFuture(false)).when(authorizationService)
+ .allowTopicOperationAsync(eq(TopicName.get("tenant/ns/denied")),
+ Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any());
+ final TransactionMetadataStoreService txnStore = mock(TransactionMetadataStoreService.class);
+ when(txnStore.verifyTxnOwnership(any(), any())).thenReturn(CompletableFuture.completedFuture(true));
+ when(pulsar.getTransactionMetadataStoreService()).thenReturn(txnStore);
+ svcConfig.setTransactionCoordinatorEnabled(true);
+ resetChannel();
+ setChannelConnected();
+
+ channel.writeInbound(Commands.newAddPartitionToTxn(89L, 1L, 12L,
+ List.of("tenant/ns/allowed", "tenant/ns/denied")));
+ CommandAddPartitionToTxnResponse response = (CommandAddPartitionToTxnResponse) getResponse();
+
+ assertEquals(response.getError(), ServerError.AuthorizationError);
+ verify(txnStore, never()).addProducedPartitionToTxn(any(TxnID.class), any());
+
+ channel.finish();
+ }
+
+ @Test(timeOut = 30000)
+ public void testAddPartitionToTxnRequiresProducePermission() throws Exception {
+ AuthorizationService authorizationService = mockTopicAuthorization(false);
+ final TransactionMetadataStoreService txnStore = mock(TransactionMetadataStoreService.class);
+ when(txnStore.verifyTxnOwnership(any(), any())).thenReturn(CompletableFuture.completedFuture(true));
+ when(pulsar.getTransactionMetadataStoreService()).thenReturn(txnStore);
+ svcConfig.setTransactionCoordinatorEnabled(true);
+ resetChannel();
+ setChannelConnected();
+
+ channel.writeInbound(Commands.newAddPartitionToTxn(89L, 1L, 12L, List.of("tenant/ns/topic1")));
+ CommandAddPartitionToTxnResponse response = (CommandAddPartitionToTxnResponse) getResponse();
+
+ assertEquals(response.getError(), ServerError.AuthorizationError);
+ verify(authorizationService, times(1)).allowTopicOperationAsync(eq(TopicName.get("tenant/ns/topic1")),
+ eq(TopicOperation.PRODUCE), any(), any(), any(), any());
+ verify(txnStore, never()).addProducedPartitionToTxn(any(TxnID.class), any());
+
+ channel.finish();
+ }
+
+ @Test(timeOut = 30000)
+ public void testAddSubscriptionToTxnRequiresConsumePermission() throws Exception {
+ AuthorizationService authorizationService = mockTopicAuthorization(false);
+ final TransactionMetadataStoreService txnStore = mock(TransactionMetadataStoreService.class);
+ when(txnStore.verifyTxnOwnership(any(), any())).thenReturn(CompletableFuture.completedFuture(true));
+ when(pulsar.getTransactionMetadataStoreService()).thenReturn(txnStore);
+ svcConfig.setTransactionCoordinatorEnabled(true);
+ resetChannel();
+ setChannelConnected();
+ final Subscription sub = new Subscription();
+ sub.setTopic("topic1");
+ sub.setSubscription("sub1");
+
+ channel.writeInbound(Commands.newAddSubscriptionToTxn(89L, 1L, 12L, List.of(sub)));
+ CommandAddSubscriptionToTxnResponse response = (CommandAddSubscriptionToTxnResponse) getResponse();
+
+ assertEquals(response.getError(), ServerError.AuthorizationError);
+ verify(authorizationService, times(1)).allowTopicOperationAsync(eq(TopicName.get("topic1")),
+ eq(TopicOperation.CONSUME), any(), any(), any(), argThat(arg ->
+ arg instanceof AuthenticationDataSubscription
+ && "sub1".equals(((AuthenticationDataSubscription) arg).getSubscription())));
+ verify(txnStore, never()).addAckedPartitionToTxn(any(TxnID.class), any());
+
+ channel.finish();
+ }
@Test(timeOut = 30000)
public void sendEndTxnResponse() throws Exception {