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/19 14:07:33 UTC
(pulsar) branch master updated: [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 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 28eea815e5c2 [fix][broker] Check topic permissions for partitions and subscriptions added to a transaction (#26644)
28eea815e5c2 is described below
commit 28eea815e5c223491537f69dbf2b52a9b5563075
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)
---
.../broker/service/BrokerServiceException.java | 8 +++
.../apache/pulsar/broker/service/ServerCnx.java | 59 ++++++++++++++--
.../pulsar/broker/service/ServerCnxTest.java | 81 ++++++++++++++++++++++
3 files changed, 143 insertions(+), 5 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 d523e3f7ec39..b4da96ee1bef 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
@@ -3821,9 +3821,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) {
log.debug()
@@ -3844,10 +3845,57 @@ 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",
- getPrincipal(), txnID
+ authenticationRoleLoggingAnonymizer.anonymize(getPrincipal()), txnID
);
log.warn().attr("msg", msg).log("");
return CompletableFuture.failedFuture(new CoordinatorException.TransactionNotFoundException(msg));
@@ -4254,9 +4302,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) {
log.debug()
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 1a5bb789075a..4d77a3dddd20 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
@@ -4126,6 +4126,87 @@ 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 {
final TransactionMetadataStoreService txnStore = mock(TransactionMetadataStoreService.class);