You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/09/01 17:29:21 UTC
[GitHub] [pulsar] KannarFr opened a new pull request, #17411: Check create topic on subscription creation
KannarFr opened a new pull request, #17411:
URL: https://github.com/apache/pulsar/pull/17411
Fixes #17406
### Motivation
Enable CREATE_TOPIC permission check on topic auto creation.
### Modifications
Introduce AuthAction.create_topic for the PulsarAuthorizationProvider
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
This change updated tests and can be verified as follows:
- updated test and give create_topic permission on the namespace
### Does this pull request potentially affect one of the following parts:
- it affects permissions as currently, CREATE_TOPIC wasn't checked on the default authZ provider.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] MarvinCai commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
MarvinCai commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r965955934
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java:
##########
@@ -482,12 +482,12 @@ public CompletableFuture<Boolean> allowNamespaceOperationAsync(NamespaceName nam
case GET_TOPIC:
case GET_TOPICS:
case GET_BUNDLE:
+ case CREATE_TOPIC:
return allowConsumeOrProduceOpsAsync(namespaceName, role, authData);
Review Comment:
seems we're only checking permission for `consume` or `produce` action here, I'm wondering what's the benefit of this pr while we don't have a action for `create_topic` itself
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r997707881
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:
##########
@@ -2361,8 +2361,11 @@ private void maybeOffloadInBackground(CompletableFuture<PositionImpl> promise) {
if (config.getLedgerOffloader() != null
&& config.getLedgerOffloader() != NullLedgerOffloader.INSTANCE
&& config.getLedgerOffloader().getOffloadPolicies() != null
- && config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInBytes() != null
- && config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInBytes() >= 0) {
+ && ((config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInBytes() != null
+ && config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInBytes() >= 0)
+ || (config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInMillis() != null
+ && config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInMillis() >= 0))
Review Comment:
I think the changes made in this file might belong in a separate PR. Would you mind making a separate PR for them?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1089,12 +1091,25 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
}
Optional<Map<String, String>> subscriptionProperties = SubscriptionOption.getPropertiesMap(
subscribe.getSubscriptionPropertiesList());
+
+ boolean createTopicIfDoesNotExist = forceTopicCreation || (isAuthorizedToCreateTopic
+ && service.isAllowAutoTopicCreation(topicName.toString()));
Review Comment:
I think we want the following in order to retain the existing logic:
```suggestion
boolean createTopicIfDoesNotExist = isAuthorizedToCreateTopic || (forceTopicCreation
&& service.isAllowAutoTopicCreation(topicName.toString()));
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1307,7 +1326,25 @@ protected void handleProducer(final CommandProducer cmdProducer) {
producerId, schema == null ? "absent" : "present");
}
- service.getOrCreateTopic(topicName.toString()).thenCompose((Topic topic) -> {
+ boolean createTopicIfDoesNotExist = isAuthorizedToCreateTopic
+ && service.isAllowAutoTopicCreation(topicName.toString());
+ service.getTopic(topicName.toString(), createTopicIfDoesNotExist).thenCompose(optTopic -> {
+ if (optTopic.isEmpty()) {
+ if (isAuthorizedToCreateTopic) {
+ return FutureUtil
+ .failedFuture(new TopicNotFoundException(
+ "Topic " + topicName + " does not exist"));
+ } else {
+ String msg = "Topic to produce does not exists and the Client is not"
+ + " authorized to create topic";
+ log.warn("[{}] {} with role {}", remoteAddress, msg, getPrincipal());
+ ctx.writeAndFlush(Commands.newError(requestId, ServerError.AuthorizationError,
+ msg));
+ return null;
Review Comment:
Similarly, we need to throw an exception here. The error handling includes handling the dangling `producer` future, and we need to do that to prevent a build up of unnecessary futures.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1089,12 +1091,25 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
}
Optional<Map<String, String>> subscriptionProperties = SubscriptionOption.getPropertiesMap(
subscribe.getSubscriptionPropertiesList());
+
+ boolean createTopicIfDoesNotExist = forceTopicCreation || (isAuthorizedToCreateTopic
+ && service.isAllowAutoTopicCreation(topicName.toString()));
service.getTopic(topicName.toString(), createTopicIfDoesNotExist)
.thenCompose(optTopic -> {
if (!optTopic.isPresent()) {
- return FutureUtil
- .failedFuture(new TopicNotFoundException(
- "Topic " + topicName + " does not exist"));
+ if (isAuthorizedToCreateTopic) {
+ return FutureUtil
+ .failedFuture(new TopicNotFoundException(
+ "Topic " + topicName + " does not exist"));
+ } else {
+ String msg = "Topic to subscribe does not exists and the Client is not"
+ + " authorized to create topic";
+ log.warn("[{}] {} with role {}", remoteAddress, msg, getPrincipal());
+ consumers.remove(consumerId, consumerFuture);
+ ctx.writeAndFlush(Commands.newError(requestId, ServerError.AuthorizationError,
+ msg));
+ return null;
Review Comment:
I don't think returning `null` here will give the desired behavior unless we modify the `thenAccept` block. It seems like the current paradigm in this section of the code is to throw an exception, so I think it would be appropriate to do that here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by "KannarFr (via GitHub)" <gi...@apache.org>.
KannarFr commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r1161935338
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1089,12 +1091,25 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
}
Optional<Map<String, String>> subscriptionProperties = SubscriptionOption.getPropertiesMap(
subscribe.getSubscriptionPropertiesList());
+
+ boolean createTopicIfDoesNotExist = forceTopicCreation || (isAuthorizedToCreateTopic
+ && service.isAllowAutoTopicCreation(topicName.toString()));
service.getTopic(topicName.toString(), createTopicIfDoesNotExist)
.thenCompose(optTopic -> {
if (!optTopic.isPresent()) {
- return FutureUtil
- .failedFuture(new TopicNotFoundException(
- "Topic " + topicName + " does not exist"));
+ if (isAuthorizedToCreateTopic) {
+ return FutureUtil
+ .failedFuture(new TopicNotFoundException(
+ "Topic " + topicName + " does not exist"));
+ } else {
+ String msg = "Topic to subscribe does not exists and the Client is not"
+ + " authorized to create topic";
+ log.warn("[{}] {} with role {}", remoteAddress, msg, getPrincipal());
+ consumers.remove(consumerId, consumerFuture);
+ ctx.writeAndFlush(Commands.newError(requestId, ServerError.AuthorizationError,
+ msg));
+ return null;
Review Comment:
I just rebased it from master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Check create topic permission on topic creation using pulsar proto clients [pulsar]
Posted by "Nowadays (via GitHub)" <gi...@apache.org>.
Nowadays commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-1748242283
Is there any way to move forward with this PR ?
publishing/consuming and creating/deleting a topic are clearly different actions, and thus should require different authorization.
Why would there be an api to create/delete topics if that was not the case ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
KannarFr commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r975725523
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1080,12 +1082,25 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
}
Optional<Map<String, String>> subscriptionProperties = SubscriptionOption.getPropertiesMap(
subscribe.getSubscriptionPropertiesList());
+
+ boolean createTopicIfDoesNotExist = isAuthorizedToCreateTopic && forceTopicCreation
+ && service.isAllowAutoTopicCreation(topicName.toString());
Review Comment:
LGTM.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
KannarFr commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r998139387
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:
##########
@@ -2361,8 +2361,11 @@ private void maybeOffloadInBackground(CompletableFuture<PositionImpl> promise) {
if (config.getLedgerOffloader() != null
&& config.getLedgerOffloader() != NullLedgerOffloader.INSTANCE
&& config.getLedgerOffloader().getOffloadPolicies() != null
- && config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInBytes() != null
- && config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInBytes() >= 0) {
+ && ((config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInBytes() != null
+ && config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInBytes() >= 0)
+ || (config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInMillis() != null
+ && config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInMillis() >= 0))
Review Comment:
Wow clearly. My bad.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
KannarFr commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r967045383
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java:
##########
@@ -482,12 +482,12 @@ public CompletableFuture<Boolean> allowNamespaceOperationAsync(NamespaceName nam
case GET_TOPIC:
case GET_TOPICS:
case GET_BUNDLE:
+ case CREATE_TOPIC:
return allowConsumeOrProduceOpsAsync(namespaceName, role, authData);
Review Comment:
It is not solving the issue. It is required to make the test pass to solve the issue. I could create a create_topic AuthAction, but it won't be backward compatible. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by "michaeljmarshall (via GitHub)" <gi...@apache.org>.
michaeljmarshall commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r1170845565
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1109,13 +1109,18 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
TopicOperation.CONSUME
);
+ CompletableFuture<Boolean> isAuthorizedToCreateTopicFuture = isNamespaceOperationAllowed(
+ topicName.getNamespaceObject(),
+ NamespaceOperation.CREATE_TOPIC
+ );
Review Comment:
As I mentioned before, this does not look backwards/forwards compatible. I don't think we can add required `AuthActions` without finding a way to make them compatible. The idea is that users should be able to upgrade then downgrade and things should still work. Can you confirm @KannarFr?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by "michaeljmarshall (via GitHub)" <gi...@apache.org>.
michaeljmarshall commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-1515484648
> > Perhaps we can introduce an additional check that calls the authorization service and then we'll implement the default PulsarAuthorizationProvider in such a way that there are no breaking change
>
> This LGTM. Can you drive this discussion on the ML if you think it's required? Shall I do it?
I won't be able to drive this discussion. Let me know if you have any questions though.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Let authorization providers restrict topic auto-creation [pulsar]
Posted by "KannarFr (via GitHub)" <gi...@apache.org>.
KannarFr commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-5742848398
@michaeljmarshall @nodece I reworked this PR to address your change requests, could you take another look?
- **Breaking change** (@michaeljmarshall): as you suggested, it is now an additional authorization check whose default keeps the current behavior. `AuthorizationProvider#allowTopicAutoCreationAsync` is a default method returning `true`, so the default `PulsarAuthorizationProvider` and existing custom providers are unaffected; a provider opts in by overriding it. It is only called when the auto topic creation settings would create the topic, and a refusal behaves as if auto-creation were disabled for that client.
- **`AuthAction`** (@nodece): no new `AuthAction`. A provider can map the check to `NamespaceOperation.CREATE_TOPIC` or anything else.
- **`handleProducer`** (@michaeljmarshall): producer, subscribe, partitioned metadata, lookup (binary and HTTP) and scalable topic lookups all go through it, with the proxy role and original principal handled as for topic operations.
The description is updated with the details and tests. Since it adds a method to a plugin interface, I'm happy to write a PIP if you prefer to go through one.
@lhotari would you mind approving the CI workflows when you have a moment?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Let authorization providers restrict topic auto-creation [pulsar]
Posted by "KannarFr (via GitHub)" <gi...@apache.org>.
KannarFr commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r4318198131
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -718,6 +718,29 @@ private CompletableFuture<Boolean> isTopicOperationAllowed(TopicName topicName,
}
}
+ /**
+ * Whether a topic that does not exist may be created for this client: the auto topic creation settings allow it
+ * and the authorization provider lets this client trigger it.
+ */
+ public CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(TopicName topicName) {
+ return service.isAllowAutoTopicCreationAsync(topicName).thenCompose(isAllowed -> {
+ if (!isAllowed || !service.isAuthorizationEnabled()) {
+ return CompletableFuture.completedFuture(isAllowed);
+ }
+ return service.getAuthorizationService().allowTopicAutoCreationAsync(topicName, originalPrincipal,
Review Comment:
Fixed. `BrokerService.isAllowAutoTopicCreationAsync` now takes the provider check as a `Supplier` and only calls it on the branches where the auto-creation settings decide, i.e. when the topic would be created from nothing. A partition covered by existing partitioned metadata, an active segment or a system topic never reaches the provider. The SPI javadoc is updated to match.
`testMissingPartitionsOfExistingPartitionedTopicAreCreated` writes only the partitioned metadata, with no partitions. It then checks that a role refused auto-creation still gets the partitions created when it produces and consumes, over both binary and HTTP lookup.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1461,13 +1484,13 @@ protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata pa
private CompletableFuture<Void> lookupPartitionMetadata(TopicName topicName, long requestId,
boolean metadataAutoCreationEnabled) {
- return FutureUtil.supplySafely(() -> getBrokerService().isAllowAutoTopicCreationAsync(topicName))
- .<CompletableFuture<Void>>handle((brokerAllowAutoCreate, ex) -> {
+ return FutureUtil.supplySafely(() -> isAllowAutoTopicCreationAsync(topicName))
Review Comment:
Done. Without `metadataAutoCreationEnabled`, the request goes straight to `lookupExistingTopicMetadata`. `testPartitionMetadataWithoutAutoCreationSkipsCreationPolicy` covers it: the creation-policy stub fails, and the request still returns the existing metadata.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -718,6 +718,29 @@ private CompletableFuture<Boolean> isTopicOperationAllowed(TopicName topicName,
}
}
+ /**
+ * Whether a topic that does not exist may be created for this client: the auto topic creation settings allow it
+ * and the authorization provider lets this client trigger it.
+ */
+ public CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(TopicName topicName) {
+ return service.isAllowAutoTopicCreationAsync(topicName).thenCompose(isAllowed -> {
+ if (!isAllowed || !service.isAuthorizationEnabled()) {
+ return CompletableFuture.completedFuture(isAllowed);
+ }
+ return service.getAuthorizationService().allowTopicAutoCreationAsync(topicName, originalPrincipal,
+ authRole, originalAuthData != null ? originalAuthData : authenticationData, authenticationData);
+ });
+ }
+
+ /**
+ * {@link BrokerService#getOrCreateTopic(String)} for this client, see {@link #isAllowAutoTopicCreationAsync}.
+ */
+ private CompletableFuture<Topic> getOrCreateTopic(TopicName topicName) {
Review Comment:
Done, and on more paths than the producer:
- Producer, subscribe and admin create-subscription load the topic first and only ask the provider when it's missing.
- Clients also send a partitioned-metadata request with auto-creation on every connect, so the check is passed down into `fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync`. It only runs once the topic is known to be missing.
- The HTTP partitioned-metadata path checks existence before asking too.
The one exception is HTTP for non-persistent topics, where the broker doesn't check existence. The javadoc says so, and also that a failed future fails the request. `testExistingTopicDoesNotDependOnAutoCreationCheck` uses a provider whose check always fails, and shows that producers and consumers on an existing topic still work over binary and HTTP.
Two related changes:
- A refusal during a binary partitioned-metadata request now answers `TopicNotFound` right away, the same as when the settings forbid creation.
- A failed check answers `AuthorizationError`, the same as a failed topic-operation check, rather than the retriable `ServiceNotReady`.
##########
pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/AuthorizationServiceTest.java:
##########
@@ -209,6 +209,19 @@ public void testTopicOperationAsync(String role, String originalRole, boolean sh
checkResult(shouldPass, isAuthorized);
}
+ @Test(dataProvider = "roles")
+ public void testTopicAutoCreationAsync(String role, String originalRole, boolean shouldPass) throws Exception {
+ boolean isAllowed = authorizationService.allowTopicAutoCreationAsync(TopicName.get("topic"),
+ originalRole, role, null, null).get();
+ checkResult(shouldPass, isAllowed);
+ }
+
+ @Test
+ public void testTopicAutoCreationAllowedByDefault() throws Exception {
+ AuthorizationProvider provider = mock(AuthorizationProvider.class, CALLS_REAL_METHODS);
+ assertTrue(provider.allowTopicAutoCreationAsync(TopicName.get("topic"), "any-role", null).get());
Review Comment:
Done. It uses AssertJ, and checks the default method through `PulsarAuthorizationProvider`, which doesn't override it.
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicAutoCreationAuthorizationTest.java:
##########
@@ -0,0 +1,206 @@
+/*
+ * 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.client.api;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import com.google.common.collect.Sets;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
+import org.apache.pulsar.broker.namespace.TopicExistsInfo;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.AuthorizationProducerConsumerTest.ClientAuthentication;
+import org.apache.pulsar.client.api.AuthorizationProducerConsumerTest.TestAuthenticationProvider;
+import org.apache.pulsar.client.api.AuthorizationProducerConsumerTest.TestAuthorizationProvider;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.common.policies.data.TopicOperation;
+import org.apache.pulsar.common.policies.data.TopicType;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+/**
+ * Topic auto-creation goes through {@code AuthorizationProvider#allowTopicAutoCreationAsync}.
+ *
+ * <p>Runs its own broker rather than {@code SharedPulsarBaseTest}, because it needs authentication and a custom
+ * authorization provider.
+ */
+@Test(groups = "broker-api")
+public class TopicAutoCreationAuthorizationTest extends ProducerConsumerBase {
+
+ private static final String CREATOR_ROLE = "creator";
+ private static final String USER_ROLE = "user";
+ private static final String NAMESPACE = "my-property/my-ns";
+
+ private TopicType autoCreationType = TopicType.NON_PARTITIONED;
+ private PulsarAdmin superAdmin;
+
+ /**
+ * Lets both roles produce, consume and look up, but only {@link #CREATOR_ROLE} trigger topic auto-creation.
+ */
+ public static class AutoCreationAuthorizationProvider extends TestAuthorizationProvider {
+
+ private static final Set<String> CLIENT_ROLES = Set.of(CREATOR_ROLE, USER_ROLE);
+
+ @Override
+ public CompletableFuture<Boolean> allowTopicOperationAsync(TopicName topic, String role,
+ TopicOperation operation,
+ AuthenticationDataSource authData) {
+ return CompletableFuture.completedFuture(CLIENT_ROLES.contains(role));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> allowTopicAutoCreationAsync(TopicName topic, String role,
+ AuthenticationDataSource authData) {
+ return CompletableFuture.completedFuture(CREATOR_ROLE.equals(role));
+ }
+ }
+
+ @BeforeMethod(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ conf.setAuthenticationEnabled(true);
+ conf.setAuthorizationEnabled(true);
+ conf.setSuperUserRoles(Set.of("superUser"));
+ conf.setAuthenticationProviders(Set.of(TestAuthenticationProvider.class.getName()));
+ conf.setAuthorizationProvider(AutoCreationAuthorizationProvider.class.getName());
+ conf.setAllowAutoTopicCreation(true);
+ conf.setAllowAutoTopicCreationType(autoCreationType);
+ conf.setTopicLevelPoliciesEnabled(false);
+ conf.setClusterName("test");
+ super.init();
+
+ superAdmin = PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString())
+ .authentication(new ClientAuthentication("superUser")).build();
+ superAdmin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
+ superAdmin.tenants().createTenant("my-property",
+ new TenantInfoImpl(Sets.newHashSet(), Sets.newHashSet("test")));
+ superAdmin.namespaces().createNamespace(NAMESPACE, Sets.newHashSet("test"));
+ }
+
+ @AfterMethod(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ if (superAdmin != null) {
+ superAdmin.close();
+ superAdmin = null;
+ }
+ super.internalCleanup();
+ }
+
+ private PulsarClient newClient(String role, String serviceUrl) throws PulsarClientException {
+ return PulsarClient.builder().serviceUrl(serviceUrl)
+ .authentication(new ClientAuthentication(role))
+ .operationTimeout(3, TimeUnit.SECONDS)
+ .build();
+ }
+
+ private boolean topicExists(String topic) throws Exception {
+ TopicExistsInfo info = pulsar.getNamespaceService().checkTopicExistsAsync(TopicName.get(topic)).get();
+ try {
+ return info.isExists();
+ } finally {
+ info.recycle();
+ }
+ }
+
+ @Test(timeOut = 60000)
+ public void testProducerAutoCreationFollowsAuthorization() throws Exception {
Review Comment:
Added, all with a real provider:
- **Proxy, binary:** `ProxyTopicAutoCreationAuthorizationTest` in pulsar-proxy. The proxy role may auto-create, the original principal may not, and the producer gets `TopicDoesNotExistException`.
- **Proxy, HTTP:** a lookup with `X-Original-Principal`, which returns 404 for a refused principal and 200 for an allowed one. While adding it, I found and fixed the HTTP side checking the original principal with the proxy's auth data. It now uses `originalPrincipalAuthData()`, as topic operations do.
- **Admin create-subscription.**
- **HTTP partitioned metadata** with `checkAllowAutoCreation`, through the HTTP transport cases.
- **`DagWatchSession`:** a V5 client looking up a missing scalable topic.
- **Existing partitions**, as in the first reply.
The manual `cleanup()`/`setup()` is gone: the partitioned case uses a second namespace with a partitioned auto-creation override, and the cases run over both lookup transports.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Let authorization providers restrict topic auto-creation [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r4327730621
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
@@ -1259,6 +1262,30 @@ public CompletableFuture<Optional<Topic>> getTopic(final String topic, boolean c
return getTopic(topic, createIfMissing, null);
}
+ /**
+ * Loads the topic and, when it does not exist, creates it if
+ * {@link #isAllowAutoTopicCreationAsync(TopicName, Supplier)} allows it with {@code newTopicCheck}. The check is
+ * only made for a topic that does not exist, so loading an existing topic does not depend on it.
+ */
+ public CompletableFuture<Optional<Topic>> getTopicCreatingIfAllowed(
+ final TopicName topicName, final Supplier<CompletableFuture<Boolean>> newTopicCheck) {
+ CompletableFuture<Optional<Topic>> existingTopic = topicName.isPersistent()
+ ? getTopic(topicName.toString(), false)
+ // A non-persistent topic only exists while it is loaded; trying to load it would report a failure
+ // to topic event listeners.
+ : CompletableFuture.completedFuture(getTopicReference(topicName.toString()));
+ return existingTopic.thenCompose(optTopic -> {
+ if (optTopic.isPresent()) {
+ return CompletableFuture.completedFuture(optTopic);
+ }
+ // A load that found the topic missing removes its empty future from the cache asynchronously; remove it
+ // now so that creating the topic does not get that future back.
+ topics.remove(topicName.toString(), existingTopic);
Review Comment:
Thanks, this does what I had in mind: the loser of two racing loads now evicts the completed-empty future the winner left in the cache ([`BrokerService.java:1289-1297`](https://github.com/apache/pulsar/blob/ad52ccb3578d115011bf5cd57677d5490e498213/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java#L1289-L1297)), and leaving in-flight or successful loads alone is right. One optional, even narrower thing: the eviction runs before the asynchronous creation check, so a load-only request arriving in between could cache a fresh empty future that the follow-up create then picks up. Evicting inside the `thenCompose` just before `getTopic(name, isAllowed)` would close that too, but I would not hold the PR for it. Resolving.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
KannarFr commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-1280983197
@michaeljmarshall are you ok with the way I've done it in handleProducer?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] github-actions[bot] commented on pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-1333056881
The pr had no activity for 30 days, mark with Stale label.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r975579399
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1080,12 +1082,25 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
}
Optional<Map<String, String>> subscriptionProperties = SubscriptionOption.getPropertiesMap(
subscribe.getSubscriptionPropertiesList());
+
+ boolean createTopicIfDoesNotExist = isAuthorizedToCreateTopic && forceTopicCreation
+ && service.isAllowAutoTopicCreation(topicName.toString());
Review Comment:
By making each of these required, it breaks the current contract, which is that permission to produce to or consume from a topic is sufficient to auto create a topic when auto topic creation is enabled.
I think it would make more sense to add this feature as an alternative to auto topic creation since we're explicitly giving a producer or consumer permission to create a topic. I think changing the PR to align with my suggestion is as simple as using `or` instead of `and` in this part of the code and any other relevant parts:
```suggestion
boolean createTopicIfDoesNotExist = forceTopicCreation || (isAuthorizedToCreateTopic
&& service.isAllowAutoTopicCreation(topicName.toString()));
```
Note that the `forceTopicCreation` is from the consumer, and I think we should honor that input even when the consumer has permission to create the topic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Let authorization providers restrict topic auto-creation [pulsar]
Posted by "KannarFr (via GitHub)" <gi...@apache.org>.
KannarFr commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r4323393678
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
@@ -1259,6 +1262,30 @@ public CompletableFuture<Optional<Topic>> getTopic(final String topic, boolean c
return getTopic(topic, createIfMissing, null);
}
+ /**
+ * Loads the topic and, when it does not exist, creates it if
+ * {@link #isAllowAutoTopicCreationAsync(TopicName, Supplier)} allows it with {@code newTopicCheck}. The check is
+ * only made for a topic that does not exist, so loading an existing topic does not depend on it.
+ */
+ public CompletableFuture<Optional<Topic>> getTopicCreatingIfAllowed(
+ final TopicName topicName, final Supplier<CompletableFuture<Boolean>> newTopicCheck) {
+ CompletableFuture<Optional<Topic>> existingTopic = topicName.isPersistent()
+ ? getTopic(topicName.toString(), false)
+ // A non-persistent topic only exists while it is loaded; trying to load it would report a failure
+ // to topic event listeners.
+ : CompletableFuture.completedFuture(getTopicReference(topicName.toString()));
+ return existingTopic.thenCompose(optTopic -> {
+ if (optTopic.isPresent()) {
+ return CompletableFuture.completedFuture(optTopic);
+ }
+ // A load that found the topic missing removes its empty future from the cache asynchronously; remove it
+ // now so that creating the topic does not get that future back.
+ topics.remove(topicName.toString(), existingTopic);
Review Comment:
Good catch, thanks. Done in ad52ccb3578: `getTopicCreatingIfAllowed` now evicts whatever completed-empty future is cached under the topic name (`topics.get(name)`, then `remove(name, cached)` when it is done and empty), not only the caller's own future, so the loser of two racing loads no longer picks up the winner's empty one.
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicAutoCreationAuthorizationTest.java:
##########
@@ -0,0 +1,206 @@
+/*
+ * 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.client.api;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import com.google.common.collect.Sets;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
+import org.apache.pulsar.broker.namespace.TopicExistsInfo;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.AuthorizationProducerConsumerTest.ClientAuthentication;
+import org.apache.pulsar.client.api.AuthorizationProducerConsumerTest.TestAuthenticationProvider;
+import org.apache.pulsar.client.api.AuthorizationProducerConsumerTest.TestAuthorizationProvider;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.common.policies.data.TopicOperation;
+import org.apache.pulsar.common.policies.data.TopicType;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+/**
+ * Topic auto-creation goes through {@code AuthorizationProvider#allowTopicAutoCreationAsync}.
+ *
+ * <p>Runs its own broker rather than {@code SharedPulsarBaseTest}, because it needs authentication and a custom
+ * authorization provider.
+ */
+@Test(groups = "broker-api")
+public class TopicAutoCreationAuthorizationTest extends ProducerConsumerBase {
+
+ private static final String CREATOR_ROLE = "creator";
+ private static final String USER_ROLE = "user";
+ private static final String NAMESPACE = "my-property/my-ns";
+
+ private TopicType autoCreationType = TopicType.NON_PARTITIONED;
+ private PulsarAdmin superAdmin;
+
+ /**
+ * Lets both roles produce, consume and look up, but only {@link #CREATOR_ROLE} trigger topic auto-creation.
+ */
+ public static class AutoCreationAuthorizationProvider extends TestAuthorizationProvider {
+
+ private static final Set<String> CLIENT_ROLES = Set.of(CREATOR_ROLE, USER_ROLE);
+
+ @Override
+ public CompletableFuture<Boolean> allowTopicOperationAsync(TopicName topic, String role,
+ TopicOperation operation,
+ AuthenticationDataSource authData) {
+ return CompletableFuture.completedFuture(CLIENT_ROLES.contains(role));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> allowTopicAutoCreationAsync(TopicName topic, String role,
+ AuthenticationDataSource authData) {
+ return CompletableFuture.completedFuture(CREATOR_ROLE.equals(role));
+ }
+ }
+
+ @BeforeMethod(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ conf.setAuthenticationEnabled(true);
+ conf.setAuthorizationEnabled(true);
+ conf.setSuperUserRoles(Set.of("superUser"));
+ conf.setAuthenticationProviders(Set.of(TestAuthenticationProvider.class.getName()));
+ conf.setAuthorizationProvider(AutoCreationAuthorizationProvider.class.getName());
+ conf.setAllowAutoTopicCreation(true);
+ conf.setAllowAutoTopicCreationType(autoCreationType);
+ conf.setTopicLevelPoliciesEnabled(false);
+ conf.setClusterName("test");
+ super.init();
+
+ superAdmin = PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString())
+ .authentication(new ClientAuthentication("superUser")).build();
+ superAdmin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
+ superAdmin.tenants().createTenant("my-property",
+ new TenantInfoImpl(Sets.newHashSet(), Sets.newHashSet("test")));
+ superAdmin.namespaces().createNamespace(NAMESPACE, Sets.newHashSet("test"));
+ }
+
+ @AfterMethod(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ if (superAdmin != null) {
+ superAdmin.close();
+ superAdmin = null;
+ }
+ super.internalCleanup();
+ }
+
+ private PulsarClient newClient(String role, String serviceUrl) throws PulsarClientException {
+ return PulsarClient.builder().serviceUrl(serviceUrl)
+ .authentication(new ClientAuthentication(role))
+ .operationTimeout(3, TimeUnit.SECONDS)
+ .build();
+ }
+
+ private boolean topicExists(String topic) throws Exception {
+ TopicExistsInfo info = pulsar.getNamespaceService().checkTopicExistsAsync(TopicName.get(topic)).get();
+ try {
+ return info.isExists();
+ } finally {
+ info.recycle();
+ }
+ }
+
+ @Test(timeOut = 60000)
+ public void testProducerAutoCreationFollowsAuthorization() throws Exception {
Review Comment:
Added in ad52ccb3578: the test provider now records the `AuthenticationDataSource` each role is checked with, and the proxied HTTP lookup test asserts that the original principal gets `AuthenticationDataForwarded` while the proxy gets its own auth data.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Let authorization providers restrict topic auto-creation [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari merged PR #17411:
URL: https://github.com/apache/pulsar/pull/17411
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by "KannarFr (via GitHub)" <gi...@apache.org>.
KannarFr commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-1515125158
> Perhaps we can introduce an additional check that calls the authorization service and then we'll implement the default PulsarAuthorizationProvider in such a way that there are no breaking change
This LGTM. Can you drive this discussion on the ML? Shall I do it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Check create topic permission on topic creation using pulsar proto clients [pulsar]
Posted by "KannarFr (via GitHub)" <gi...@apache.org>.
KannarFr commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-2015009455
Any bumps?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] gaoran10 commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
gaoran10 commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r961752310
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1080,102 +1077,126 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
}
Optional<Map<String, String>> subscriptionProperties = SubscriptionOption.getPropertiesMap(
subscribe.getSubscriptionPropertiesList());
- service.getTopic(topicName.toString(), createTopicIfDoesNotExist)
- .thenCompose(optTopic -> {
- if (!optTopic.isPresent()) {
- return FutureUtil
- .failedFuture(new TopicNotFoundException(
- "Topic " + topicName + " does not exist"));
- }
- Topic topic = optTopic.get();
+ CompletableFuture<Boolean> isAuthorizedToCreateTopicFuture = isNamespaceOperationAllowed(
+ topicName.getNamespaceObject(),
+ NamespaceOperation.CREATE_TOPIC
Review Comment:
Maybe we can add this before [line 1026](https://github.com/apache/pulsar/pull/17411/files#diff-1e0e8195fb5ec5a6d79acbc7d859c025a9b711f94e6ab37c94439e99b3202e84L1026), and use this method `isAuthorizedFuture.thenCombine(isAuthorizedToCreateTopicFuture, (isAuthorized, isAuthorizedToCreateTopic)`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
KannarFr commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r962846141
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1080,102 +1077,126 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
}
Optional<Map<String, String>> subscriptionProperties = SubscriptionOption.getPropertiesMap(
subscribe.getSubscriptionPropertiesList());
- service.getTopic(topicName.toString(), createTopicIfDoesNotExist)
- .thenCompose(optTopic -> {
- if (!optTopic.isPresent()) {
- return FutureUtil
- .failedFuture(new TopicNotFoundException(
- "Topic " + topicName + " does not exist"));
- }
- Topic topic = optTopic.get();
+ CompletableFuture<Boolean> isAuthorizedToCreateTopicFuture = isNamespaceOperationAllowed(
+ topicName.getNamespaceObject(),
+ NamespaceOperation.CREATE_TOPIC
Review Comment:
This will prevent produce/consume of topic if it already exists but create_topic operation not authorized, nope?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r998856483
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:
##########
@@ -2361,8 +2361,11 @@ private void maybeOffloadInBackground(CompletableFuture<PositionImpl> promise) {
if (config.getLedgerOffloader() != null
&& config.getLedgerOffloader() != NullLedgerOffloader.INSTANCE
&& config.getLedgerOffloader().getOffloadPolicies() != null
- && config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInBytes() != null
- && config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInBytes() >= 0) {
+ && ((config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInBytes() != null
+ && config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInBytes() >= 0)
+ || (config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInMillis() != null
+ && config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInMillis() >= 0))
Review Comment:
No problem.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Nowadays commented on pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by "Nowadays (via GitHub)" <gi...@apache.org>.
Nowadays commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-1658612509
Is there any progression on this issue ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r1161895207
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1089,12 +1091,25 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
}
Optional<Map<String, String>> subscriptionProperties = SubscriptionOption.getPropertiesMap(
subscribe.getSubscriptionPropertiesList());
+
+ boolean createTopicIfDoesNotExist = forceTopicCreation || (isAuthorizedToCreateTopic
+ && service.isAllowAutoTopicCreation(topicName.toString()));
service.getTopic(topicName.toString(), createTopicIfDoesNotExist)
.thenCompose(optTopic -> {
if (!optTopic.isPresent()) {
- return FutureUtil
- .failedFuture(new TopicNotFoundException(
- "Topic " + topicName + " does not exist"));
+ if (isAuthorizedToCreateTopic) {
+ return FutureUtil
+ .failedFuture(new TopicNotFoundException(
+ "Topic " + topicName + " does not exist"));
+ } else {
+ String msg = "Topic to subscribe does not exists and the Client is not"
+ + " authorized to create topic";
+ log.warn("[{}] {} with role {}", remoteAddress, msg, getPrincipal());
+ consumers.remove(consumerId, consumerFuture);
+ ctx.writeAndFlush(Commands.newError(requestId, ServerError.AuthorizationError,
+ msg));
+ return null;
Review Comment:
Hi @michaeljmarshall
Could you retake a look? Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by "KannarFr (via GitHub)" <gi...@apache.org>.
KannarFr commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-1514468941
> After reviewing this, here are a collection of thoughts on Pulsar's Authorization.
>
> At it's core, I think the problem described by #17406 is that `allowAutoTopicCreation` is a configuration about permission/authorization.
Yes?
> In my view, the `allowAutoTopicCreation=true` says "a role with permission to produce/consume to a topic also has permission to create that topic".
The HTTP admin API does not comply with this sentence. The CREATE_TOPIC operation is defined and used by HTTP admin API authz checks. I agree that this is introducing breaking changes in the permissions system and this is a problem, but there is authZ plugin provider providing this operation check and does not verify it during producer/consumer. So, I have no idea what the best answer is, but we can't stay and need to find a solution or make a decision here.
> This change proposes that `allowAutoTopicCreation=true` and produce/consumer permissions are insufficient, and that a role must also have explicit permission to create a topic.
Well, this is the default behavior of every configurations keys in pulsar. Like namespace policies or tiered storage configurations, there is a default value that can be overridden by custom namespace policies.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by "KannarFr (via GitHub)" <gi...@apache.org>.
KannarFr commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r1171121469
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1109,13 +1109,18 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
TopicOperation.CONSUME
);
+ CompletableFuture<Boolean> isAuthorizedToCreateTopicFuture = isNamespaceOperationAllowed(
+ topicName.getNamespaceObject(),
+ NamespaceOperation.CREATE_TOPIC
+ );
Review Comment:
That would be nice, but as I answered you am I not sure we can really do it has CREATE_TOPIC check is already performed in HTTP admin checks.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1109,13 +1109,18 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
TopicOperation.CONSUME
);
+ CompletableFuture<Boolean> isAuthorizedToCreateTopicFuture = isNamespaceOperationAllowed(
+ topicName.getNamespaceObject(),
+ NamespaceOperation.CREATE_TOPIC
+ );
Review Comment:
That would be nice, but as I answered you am I not sure we can really do it as CREATE_TOPIC check is already performed in HTTP admin checks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by "michaeljmarshall (via GitHub)" <gi...@apache.org>.
michaeljmarshall commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r1171527978
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1109,13 +1109,18 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
TopicOperation.CONSUME
);
+ CompletableFuture<Boolean> isAuthorizedToCreateTopicFuture = isNamespaceOperationAllowed(
+ topicName.getNamespaceObject(),
+ NamespaceOperation.CREATE_TOPIC
+ );
Review Comment:
While I agree that the current design is inconsistent, I don't think we should add logic that breaks existing use cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Let authorization providers restrict topic auto-creation [pulsar]
Posted by "KannarFr (via GitHub)" <gi...@apache.org>.
KannarFr commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-6276016753
@lhotari I've addressed your review, PTAL when you have a moment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
KannarFr commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r966206640
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java:
##########
@@ -482,12 +482,12 @@ public CompletableFuture<Boolean> allowNamespaceOperationAsync(NamespaceName nam
case GET_TOPIC:
case GET_TOPICS:
case GET_BUNDLE:
+ case CREATE_TOPIC:
return allowConsumeOrProduceOpsAsync(namespaceName, role, authData);
Review Comment:
The default authZ provider hasn't but it AuthorizationProvider is a plugin interface that we are using with biscuit token: https://github.com/CleverCloud/biscuit-pulsar.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
KannarFr commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r962846141
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1080,102 +1077,126 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
}
Optional<Map<String, String>> subscriptionProperties = SubscriptionOption.getPropertiesMap(
subscribe.getSubscriptionPropertiesList());
- service.getTopic(topicName.toString(), createTopicIfDoesNotExist)
- .thenCompose(optTopic -> {
- if (!optTopic.isPresent()) {
- return FutureUtil
- .failedFuture(new TopicNotFoundException(
- "Topic " + topicName + " does not exist"));
- }
- Topic topic = optTopic.get();
+ CompletableFuture<Boolean> isAuthorizedToCreateTopicFuture = isNamespaceOperationAllowed(
+ topicName.getNamespaceObject(),
+ NamespaceOperation.CREATE_TOPIC
Review Comment:
This will prevent produce/consume of topic if it already exist but create_topic operation not authorized, nope?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] github-actions[bot] commented on pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-1281833204
@KannarFr Please add the following content to your PR description and select a checkbox:
```
- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Check create topic permission on topic creation using pulsar proto clients [pulsar]
Posted by "KannarFr (via GitHub)" <gi...@apache.org>.
KannarFr commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-1747772227
@michaeljmarshall @gaoran10 @MarvinCai, it makes sense to rebase it..?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Let authorization providers restrict topic auto-creation [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r4315943262
##########
pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/AuthorizationServiceTest.java:
##########
@@ -209,6 +209,19 @@ public void testTopicOperationAsync(String role, String originalRole, boolean sh
checkResult(shouldPass, isAuthorized);
}
+ @Test(dataProvider = "roles")
+ public void testTopicAutoCreationAsync(String role, String originalRole, boolean shouldPass) throws Exception {
+ boolean isAllowed = authorizationService.allowTopicAutoCreationAsync(TopicName.get("topic"),
+ originalRole, role, null, null).get();
+ checkResult(shouldPass, isAllowed);
+ }
+
+ @Test
+ public void testTopicAutoCreationAllowedByDefault() throws Exception {
+ AuthorizationProvider provider = mock(AuthorizationProvider.class, CALLS_REAL_METHODS);
+ assertTrue(provider.allowTopicAutoCreationAsync(TopicName.get("topic"), "any-role", null).get());
Review Comment:
**[QUALITY] prefer AssertJ and a concrete default-implementation check**
Small test nit: `assertTrue(...)` could be `assertThat(...).isTrue()` to match the AssertJ used in the new integration test. A tiny concrete `AuthorizationProvider` subclass would also exercise the default method without `mock(..., CALLS_REAL_METHODS)`.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -718,6 +718,29 @@ private CompletableFuture<Boolean> isTopicOperationAllowed(TopicName topicName,
}
}
+ /**
+ * Whether a topic that does not exist may be created for this client: the auto topic creation settings allow it
+ * and the authorization provider lets this client trigger it.
+ */
+ public CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(TopicName topicName) {
+ return service.isAllowAutoTopicCreationAsync(topicName).thenCompose(isAllowed -> {
+ if (!isAllowed || !service.isAuthorizationEnabled()) {
+ return CompletableFuture.completedFuture(isAllowed);
+ }
+ return service.getAuthorizationService().allowTopicAutoCreationAsync(topicName, originalPrincipal,
Review Comment:
**[BUG] the provider check also refuses existing partitions and active segments that merely have to be (re)materialized**
`BrokerService.isAllowAutoTopicCreationAsync` deliberately returns `true` for `x-partition-N` when the partitioned-topic metadata exists with more than N partitions, whatever the auto-creation setting says ([`BrokerService.java:4586-4599`](https://github.com/apache/pulsar/blob/536ba91618912d5d2a893a08aab8760aecb4426b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java#L4586-L4599)). The same method allows active `segment://` topics and system topics. Those are not new topics, they are the broker loading something the admin already created, and the `getTopic(..., true)` is what lets it do so when there is no loaded topic or ledger.
This new check is layered on top of that result, so a role that may produce/consume but may not trigger auto-creation gets `TopicNotFound` on a partition of a partitioned topic an admin created explicitly, as soon as that partition has to be created again (for example after inactive-topic deletion removed the partition while the parent metadata stays). That contradicts the javadoc on the SPI ("a refusal only prevents creating a missing topic") and the PR description ("existing topics are used as before").
Could the hook only be consulted when the topic would really be created from nothing, i.e. skip it when the topic name is a partition (or segment) covered by existing partitioned/scalable metadata? A test with an admin-created partitioned topic and a role that is denied auto-creation would pin it.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -718,6 +718,29 @@ private CompletableFuture<Boolean> isTopicOperationAllowed(TopicName topicName,
}
}
+ /**
+ * Whether a topic that does not exist may be created for this client: the auto topic creation settings allow it
+ * and the authorization provider lets this client trigger it.
+ */
+ public CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(TopicName topicName) {
+ return service.isAllowAutoTopicCreationAsync(topicName).thenCompose(isAllowed -> {
+ if (!isAllowed || !service.isAuthorizationEnabled()) {
+ return CompletableFuture.completedFuture(isAllowed);
+ }
+ return service.getAuthorizationService().allowTopicAutoCreationAsync(topicName, originalPrincipal,
+ authRole, originalAuthData != null ? originalAuthData : authenticationData, authenticationData);
+ });
+ }
+
+ /**
+ * {@link BrokerService#getOrCreateTopic(String)} for this client, see {@link #isAllowAutoTopicCreationAsync}.
+ */
+ private CompletableFuture<Topic> getOrCreateTopic(TopicName topicName) {
Review Comment:
**[QUALITY] the provider is consulted on every producer connect, including for topics that already exist**
`getOrCreateTopic` (and the `forceTopicCreation` branch in `handleSubscribe`) awaits `allowTopicAutoCreationAsync` before the topic cache is consulted. The javadoc on the SPI documents this, but it means a provider that does a remote or ACL lookup adds that cost to every (re)connect of every producer and consumer, and if the provider future fails, connects to existing topics fail too, although the check can never matter for them (a `false` result does not reject an existing topic).
Would it be feasible to try `service.getTopic(name, false)` first and only ask the provider when that comes back empty? If you prefer to keep the current shape, I think the javadoc should at least say that a failure of the returned future fails the client request.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1461,13 +1484,13 @@ protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata pa
private CompletableFuture<Void> lookupPartitionMetadata(TopicName topicName, long requestId,
boolean metadataAutoCreationEnabled) {
- return FutureUtil.supplySafely(() -> getBrokerService().isAllowAutoTopicCreationAsync(topicName))
- .<CompletableFuture<Void>>handle((brokerAllowAutoCreate, ex) -> {
+ return FutureUtil.supplySafely(() -> isAllowAutoTopicCreationAsync(topicName))
Review Comment:
**[BUG] read-only partitioned-metadata requests now depend on the creation check**
The provider hook runs before `metadataAutoCreationEnabled` is looked at, so a client that explicitly asked not to auto-create (the common `lookupExistingTopicMetadata` case) still waits for the provider, and if the provider completes exceptionally the request ends in `MetadataError` even though the topic metadata exists and nothing could be created.
Suggest evaluating the combined check only when `metadataAutoCreationEnabled` is true, and going straight to `lookupExistingTopicMetadata` otherwise. That keeps the pre-existing read path independent of the new SPI.
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicAutoCreationAuthorizationTest.java:
##########
@@ -0,0 +1,206 @@
+/*
+ * 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.client.api;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import com.google.common.collect.Sets;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
+import org.apache.pulsar.broker.namespace.TopicExistsInfo;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.AuthorizationProducerConsumerTest.ClientAuthentication;
+import org.apache.pulsar.client.api.AuthorizationProducerConsumerTest.TestAuthenticationProvider;
+import org.apache.pulsar.client.api.AuthorizationProducerConsumerTest.TestAuthorizationProvider;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.common.policies.data.TopicOperation;
+import org.apache.pulsar.common.policies.data.TopicType;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+/**
+ * Topic auto-creation goes through {@code AuthorizationProvider#allowTopicAutoCreationAsync}.
+ *
+ * <p>Runs its own broker rather than {@code SharedPulsarBaseTest}, because it needs authentication and a custom
+ * authorization provider.
+ */
+@Test(groups = "broker-api")
+public class TopicAutoCreationAuthorizationTest extends ProducerConsumerBase {
+
+ private static final String CREATOR_ROLE = "creator";
+ private static final String USER_ROLE = "user";
+ private static final String NAMESPACE = "my-property/my-ns";
+
+ private TopicType autoCreationType = TopicType.NON_PARTITIONED;
+ private PulsarAdmin superAdmin;
+
+ /**
+ * Lets both roles produce, consume and look up, but only {@link #CREATOR_ROLE} trigger topic auto-creation.
+ */
+ public static class AutoCreationAuthorizationProvider extends TestAuthorizationProvider {
+
+ private static final Set<String> CLIENT_ROLES = Set.of(CREATOR_ROLE, USER_ROLE);
+
+ @Override
+ public CompletableFuture<Boolean> allowTopicOperationAsync(TopicName topic, String role,
+ TopicOperation operation,
+ AuthenticationDataSource authData) {
+ return CompletableFuture.completedFuture(CLIENT_ROLES.contains(role));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> allowTopicAutoCreationAsync(TopicName topic, String role,
+ AuthenticationDataSource authData) {
+ return CompletableFuture.completedFuture(CREATOR_ROLE.equals(role));
+ }
+ }
+
+ @BeforeMethod(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ conf.setAuthenticationEnabled(true);
+ conf.setAuthorizationEnabled(true);
+ conf.setSuperUserRoles(Set.of("superUser"));
+ conf.setAuthenticationProviders(Set.of(TestAuthenticationProvider.class.getName()));
+ conf.setAuthorizationProvider(AutoCreationAuthorizationProvider.class.getName());
+ conf.setAllowAutoTopicCreation(true);
+ conf.setAllowAutoTopicCreationType(autoCreationType);
+ conf.setTopicLevelPoliciesEnabled(false);
+ conf.setClusterName("test");
+ super.init();
+
+ superAdmin = PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString())
+ .authentication(new ClientAuthentication("superUser")).build();
+ superAdmin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
+ superAdmin.tenants().createTenant("my-property",
+ new TenantInfoImpl(Sets.newHashSet(), Sets.newHashSet("test")));
+ superAdmin.namespaces().createNamespace(NAMESPACE, Sets.newHashSet("test"));
+ }
+
+ @AfterMethod(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ if (superAdmin != null) {
+ superAdmin.close();
+ superAdmin = null;
+ }
+ super.internalCleanup();
+ }
+
+ private PulsarClient newClient(String role, String serviceUrl) throws PulsarClientException {
+ return PulsarClient.builder().serviceUrl(serviceUrl)
+ .authentication(new ClientAuthentication(role))
+ .operationTimeout(3, TimeUnit.SECONDS)
+ .build();
+ }
+
+ private boolean topicExists(String topic) throws Exception {
+ TopicExistsInfo info = pulsar.getNamespaceService().checkTopicExistsAsync(TopicName.get(topic)).get();
+ try {
+ return info.isExists();
+ } finally {
+ info.recycle();
+ }
+ }
+
+ @Test(timeOut = 60000)
+ public void testProducerAutoCreationFollowsAuthorization() throws Exception {
Review Comment:
**[QUALITY] integration test does not cover several of the entry points the PR changes**
The test covers the binary producer and consumer, an HTTP lookup and partitioned auto-creation, which is a good start. These entry points touched by the PR have no case with a real provider: a proxied request with an original principal (the `AuthorizationServiceTest` matrix only covers the service, not `ServerCnx`/`PulsarWebResource`), the admin create-subscription path, HTTP partitioned-metadata with `checkAllowAutoCreation`, `DagWatchSession` (only mocked in `DagWatchSessionTest`, and with a stub that returns `true`), and the existing-partitions case from the comment on [`ServerCnx.java:730`](https://github.com/apache/pulsar/blob/536ba91618912d5d2a893a08aab8760aecb4426b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java#L730).
Also, `testPartitionedAutoCreationFollowsAuthorization` calls `cleanup()`/`setup()` by hand inside a class with `@BeforeMethod`/`@AfterMethod` and flips a field; a separate test class (or a data provider for the topic type) would be less brittle.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Let authorization providers restrict topic auto-creation [pulsar]
Posted by "KannarFr (via GitHub)" <gi...@apache.org>.
KannarFr commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-6275988811
Thanks for the thorough review, Lari. It's addressed in two commits: ad065668da0 has the fixes for your comments, and 66c62c45765 has the remaining tests plus two fixes found while testing. Three things go a bit beyond what you asked, so I'll flag them:
- `BrokerService.getTopicCreatingIfAllowed` holds the "load first, only check if missing" logic in one place, used by `ServerCnx` and by admin create-subscription.
- In `fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync`, the provider is only asked after the remote-cluster lookup. A topic that already exists in a peer cluster is re-materialized, not created, so it shouldn't be blocked.
- The authenticated `PersistentTopicsBase.getPartitionedTopicMetadata(pulsar, clientAppId, …)` now asks the provider too. It knows who the caller is, but it was auto-creating without asking.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Let authorization providers restrict topic auto-creation [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r4318720538
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
@@ -1259,6 +1262,30 @@ public CompletableFuture<Optional<Topic>> getTopic(final String topic, boolean c
return getTopic(topic, createIfMissing, null);
}
+ /**
+ * Loads the topic and, when it does not exist, creates it if
+ * {@link #isAllowAutoTopicCreationAsync(TopicName, Supplier)} allows it with {@code newTopicCheck}. The check is
+ * only made for a topic that does not exist, so loading an existing topic does not depend on it.
+ */
+ public CompletableFuture<Optional<Topic>> getTopicCreatingIfAllowed(
+ final TopicName topicName, final Supplier<CompletableFuture<Boolean>> newTopicCheck) {
+ CompletableFuture<Optional<Topic>> existingTopic = topicName.isPersistent()
+ ? getTopic(topicName.toString(), false)
+ // A non-persistent topic only exists while it is loaded; trying to load it would report a failure
+ // to topic event listeners.
+ : CompletableFuture.completedFuture(getTopicReference(topicName.toString()));
+ return existingTopic.thenCompose(optTopic -> {
+ if (optTopic.isPresent()) {
+ return CompletableFuture.completedFuture(optTopic);
+ }
+ // A load that found the topic missing removes its empty future from the cache asynchronously; remove it
+ // now so that creating the topic does not get that future back.
+ topics.remove(topicName.toString(), existingTopic);
Review Comment:
**[QUALITY] non-blocking: the eviction in getTopicCreatingIfAllowed misses the cached future when two loads race**
Nice that the load-first step is in one place now. A small, narrow race to be aware of: `topics.remove(name, existingTopic)` only evicts the future this caller received. When two load-only `getTopic(name, false)` calls race, one installs its context future `C` in `topics` and the other gets a different forwarding future `F` ([`BrokerService.java:1462-1477`](https://github.com/apache/pulsar/blob/66c62c45765babbd64dc55db2c2cadf419e5dbf8/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java#L1462-L1477)). If the ledger turns out to be missing, `C` is removed asynchronously on the executor ([`BrokerService.java:2443`](https://github.com/apache/pulsar/blob/66c62c45765babbd64dc55db2c2cadf419e5dbf8/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java#L2443)), so for the loser `remove(name, F)` is a no-op and the follow-up `getTopic(name, true)` can still pick up the empty `C` and report not found to a role that is allowed to create.
It needs a topic that exists in metadata but has no managed ledger, plus the two loads racing, so I would not hold the PR for it. If it is cheap, also evicting whatever completed-empty future is cached under that name (`topics.get(name)`, then `remove(name, cached)` when it is done and empty) would close it.
##########
pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/AuthorizationServiceTest.java:
##########
@@ -209,6 +209,19 @@ public void testTopicOperationAsync(String role, String originalRole, boolean sh
checkResult(shouldPass, isAuthorized);
}
+ @Test(dataProvider = "roles")
+ public void testTopicAutoCreationAsync(String role, String originalRole, boolean shouldPass) throws Exception {
+ boolean isAllowed = authorizationService.allowTopicAutoCreationAsync(TopicName.get("topic"),
+ originalRole, role, null, null).get();
+ checkResult(shouldPass, isAllowed);
+ }
+
+ @Test
+ public void testTopicAutoCreationAllowedByDefault() throws Exception {
+ AuthorizationProvider provider = mock(AuthorizationProvider.class, CALLS_REAL_METHODS);
+ assertTrue(provider.allowTopicAutoCreationAsync(TopicName.get("topic"), "any-role", null).get());
Review Comment:
Thanks, that is cleaner. Resolving.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -718,6 +718,29 @@ private CompletableFuture<Boolean> isTopicOperationAllowed(TopicName topicName,
}
}
+ /**
+ * Whether a topic that does not exist may be created for this client: the auto topic creation settings allow it
+ * and the authorization provider lets this client trigger it.
+ */
+ public CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(TopicName topicName) {
+ return service.isAllowAutoTopicCreationAsync(topicName).thenCompose(isAllowed -> {
+ if (!isAllowed || !service.isAuthorizationEnabled()) {
+ return CompletableFuture.completedFuture(isAllowed);
+ }
+ return service.getAuthorizationService().allowTopicAutoCreationAsync(topicName, originalPrincipal,
Review Comment:
Thanks, this reads right now. The provider is only asked on the branches where the settings would create the topic from nothing, so partitions of existing partitioned metadata, active segments and system topics skip it ([`BrokerService.java:4654-4663`](https://github.com/apache/pulsar/blob/66c62c45765babbd64dc55db2c2cadf419e5dbf8/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java#L4654-L4663)), and `testMissingPartitionsOfExistingPartitionedTopicAreCreated` pins the partitions-only case over both lookup transports. Resolving.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1461,13 +1484,13 @@ protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata pa
private CompletableFuture<Void> lookupPartitionMetadata(TopicName topicName, long requestId,
boolean metadataAutoCreationEnabled) {
- return FutureUtil.supplySafely(() -> getBrokerService().isAllowAutoTopicCreationAsync(topicName))
- .<CompletableFuture<Void>>handle((brokerAllowAutoCreate, ex) -> {
+ return FutureUtil.supplySafely(() -> isAllowAutoTopicCreationAsync(topicName))
Review Comment:
Looks good. [`ServerCnx.java:1511`](https://github.com/apache/pulsar/blob/66c62c45765babbd64dc55db2c2cadf419e5dbf8/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java#L1511) now skips every creation check when the client did not ask for auto-creation, and `testPartitionMetadataWithoutAutoCreationSkipsCreationPolicy` would fail on the old ordering. Resolving.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] github-actions[bot] commented on pull request #17411: Check create topic on subscription creation
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-1234589394
@KannarFr Please provide a correct documentation label for your PR.
Instructions see [Pulsar Documentation Label Guide](https://docs.google.com/document/d/1Qw7LHQdXWBW9t2-r-A7QdFDBwmZh6ytB4guwMoXHqc0).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
KannarFr commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-1297212734
@michaeljmarshall there is something I missed?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
KannarFr commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r998152854
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1089,12 +1091,25 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
}
Optional<Map<String, String>> subscriptionProperties = SubscriptionOption.getPropertiesMap(
subscribe.getSubscriptionPropertiesList());
+
+ boolean createTopicIfDoesNotExist = forceTopicCreation || (isAuthorizedToCreateTopic
+ && service.isAllowAutoTopicCreation(topicName.toString()));
service.getTopic(topicName.toString(), createTopicIfDoesNotExist)
.thenCompose(optTopic -> {
if (!optTopic.isPresent()) {
- return FutureUtil
- .failedFuture(new TopicNotFoundException(
- "Topic " + topicName + " does not exist"));
+ if (isAuthorizedToCreateTopic) {
+ return FutureUtil
+ .failedFuture(new TopicNotFoundException(
+ "Topic " + topicName + " does not exist"));
+ } else {
+ String msg = "Topic to subscribe does not exists and the Client is not"
+ + " authorized to create topic";
+ log.warn("[{}] {} with role {}", remoteAddress, msg, getPrincipal());
+ consumers.remove(consumerId, consumerFuture);
+ ctx.writeAndFlush(Commands.newError(requestId, ServerError.AuthorizationError,
+ msg));
+ return null;
Review Comment:
Did you see https://github.com/apache/pulsar/pull/17411/files#diff-1e0e8195fb5ec5a6d79acbc7d859c025a9b711f94e6ab37c94439e99b3202e84L1187-L1198 lines? It seems we are returning null and writing errors to the channel. Nope?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
KannarFr commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r962851190
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1080,102 +1077,126 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
}
Optional<Map<String, String>> subscriptionProperties = SubscriptionOption.getPropertiesMap(
subscribe.getSubscriptionPropertiesList());
- service.getTopic(topicName.toString(), createTopicIfDoesNotExist)
- .thenCompose(optTopic -> {
- if (!optTopic.isPresent()) {
- return FutureUtil
- .failedFuture(new TopicNotFoundException(
- "Topic " + topicName + " does not exist"));
- }
- Topic topic = optTopic.get();
+ CompletableFuture<Boolean> isAuthorizedToCreateTopicFuture = isNamespaceOperationAllowed(
+ topicName.getNamespaceObject(),
+ NamespaceOperation.CREATE_TOPIC
Review Comment:
Oh nvm, got it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
KannarFr commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r962856750
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1080,102 +1077,126 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
}
Optional<Map<String, String>> subscriptionProperties = SubscriptionOption.getPropertiesMap(
subscribe.getSubscriptionPropertiesList());
- service.getTopic(topicName.toString(), createTopicIfDoesNotExist)
- .thenCompose(optTopic -> {
- if (!optTopic.isPresent()) {
- return FutureUtil
- .failedFuture(new TopicNotFoundException(
- "Topic " + topicName + " does not exist"));
- }
- Topic topic = optTopic.get();
+ CompletableFuture<Boolean> isAuthorizedToCreateTopicFuture = isNamespaceOperationAllowed(
+ topicName.getNamespaceObject(),
+ NamespaceOperation.CREATE_TOPIC
Review Comment:
@gaoran10 done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
KannarFr commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r968349481
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java:
##########
@@ -482,12 +482,12 @@ public CompletableFuture<Boolean> allowNamespaceOperationAsync(NamespaceName nam
case GET_TOPIC:
case GET_TOPICS:
case GET_BUNDLE:
+ case CREATE_TOPIC:
return allowConsumeOrProduceOpsAsync(namespaceName, role, authData);
Review Comment:
It's a breaking change but I added your suggestion. WDYT?
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java:
##########
@@ -482,12 +482,12 @@ public CompletableFuture<Boolean> allowNamespaceOperationAsync(NamespaceName nam
case GET_TOPIC:
case GET_TOPICS:
case GET_BUNDLE:
+ case CREATE_TOPIC:
return allowConsumeOrProduceOpsAsync(namespaceName, role, authData);
Review Comment:
It is not solving the issue. It is required to make the test pass to solve the issue. I could create a create_topic AuthAction, but it won't be backward compatible. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by "KannarFr (via GitHub)" <gi...@apache.org>.
KannarFr commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-1658733566
Unfortunately, no. I don't know why this is stuck @michaeljmarshall?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by "KannarFr (via GitHub)" <gi...@apache.org>.
KannarFr commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-1658734399
@gaoran10 @MarvinCai
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by "KannarFr (via GitHub)" <gi...@apache.org>.
KannarFr commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-1513763791
@poorbarcode, @michaeljmarshall looks not available, can you ping someone else? It would be cool to merge it upstream asap :D.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by "michaeljmarshall (via GitHub)" <gi...@apache.org>.
michaeljmarshall commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-1514978383
> The HTTP admin API does not comply with this sentence.
Fair point. I hadn't considered the pulsar and http endpoints together when writing that generalization.
> I agree that this is introducing breaking changes in the permissions system and this is a problem, but there is authZ plugin provider providing this operation check and does not verify it during producer/consumer.
Is there another way to achieve this? Perhaps we can introduce an additional check that calls the authorization service and then we'll implement the default PulsarAuthorizationProvider in such a way that there are no breaking changes.
I think it might make sense to discuss this inconsistency on the pulsar mailing list.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] github-actions[bot] commented on pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-1573031932
The pr had no activity for 30 days, mark with Stale label.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by "KannarFr (via GitHub)" <gi...@apache.org>.
KannarFr commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-1573386775
I sent
> Hi,
>
> CREATE_TOPIC authorization check is not performed when trying to PRODUCE/CONSUME a topic, it has been referenced: https://github.com/apache/pulsar/issues/17406.
>
> I opened a PR to fix it https://github.com/apache/pulsar/pull/17411, but Michael reported issues about backward compatibility (which is totally correct). Adding support of CREATE_TOPIC authorization as-is will break current authorization system. I noticed that HTTP Admin API verifies the CREATE_TOPIC right when creating topic, so we have inconsistencies between pulsar binary protocol and the HTTP admin API on this.
>
> Also, the AuthorizationProvider is an interface exposing the CREATE_TOPIC feature for authZ plugins. But it is inconsistent too.
>
> Michael suggested to fix this interface to support the CREATE_TOPIC check and adapt the pulsar's DefaultAuthzProvider to continue as-is.
>
> I'd like to know what do you think?
>
> Thanks,
>
> Kannar
On the ML the 20/04/2023 and still have no answers @michaeljmarshall @nodece.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r998859353
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1089,12 +1091,25 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
}
Optional<Map<String, String>> subscriptionProperties = SubscriptionOption.getPropertiesMap(
subscribe.getSubscriptionPropertiesList());
+
+ boolean createTopicIfDoesNotExist = forceTopicCreation || (isAuthorizedToCreateTopic
+ && service.isAllowAutoTopicCreation(topicName.toString()));
service.getTopic(topicName.toString(), createTopicIfDoesNotExist)
.thenCompose(optTopic -> {
if (!optTopic.isPresent()) {
- return FutureUtil
- .failedFuture(new TopicNotFoundException(
- "Topic " + topicName + " does not exist"));
+ if (isAuthorizedToCreateTopic) {
+ return FutureUtil
+ .failedFuture(new TopicNotFoundException(
+ "Topic " + topicName + " does not exist"));
+ } else {
+ String msg = "Topic to subscribe does not exists and the Client is not"
+ + " authorized to create topic";
+ log.warn("[{}] {} with role {}", remoteAddress, msg, getPrincipal());
+ consumers.remove(consumerId, consumerFuture);
+ ctx.writeAndFlush(Commands.newError(requestId, ServerError.AuthorizationError,
+ msg));
+ return null;
Review Comment:
If we return `null` here, we will go to the `thenAccept` block and will call `consumerFuture.complete(consumer)`. We don't want to complete the future there. I think we wnat to run the lines 1187 to 1198 that you reference. Returning a failed future ensures that we jump to the `exceptionally` block on 1169 that will send the error response.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
KannarFr commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r999299016
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1089,12 +1091,25 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
}
Optional<Map<String, String>> subscriptionProperties = SubscriptionOption.getPropertiesMap(
subscribe.getSubscriptionPropertiesList());
+
+ boolean createTopicIfDoesNotExist = forceTopicCreation || (isAuthorizedToCreateTopic
+ && service.isAllowAutoTopicCreation(topicName.toString()));
service.getTopic(topicName.toString(), createTopicIfDoesNotExist)
.thenCompose(optTopic -> {
if (!optTopic.isPresent()) {
- return FutureUtil
- .failedFuture(new TopicNotFoundException(
- "Topic " + topicName + " does not exist"));
+ if (isAuthorizedToCreateTopic) {
+ return FutureUtil
+ .failedFuture(new TopicNotFoundException(
+ "Topic " + topicName + " does not exist"));
+ } else {
+ String msg = "Topic to subscribe does not exists and the Client is not"
+ + " authorized to create topic";
+ log.warn("[{}] {} with role {}", remoteAddress, msg, getPrincipal());
+ consumers.remove(consumerId, consumerFuture);
+ ctx.writeAndFlush(Commands.newError(requestId, ServerError.AuthorizationError,
+ msg));
+ return null;
Review Comment:
Ok, does e34524abd0a9cea41e7968f518e6929647e4e140 LGTY?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1307,7 +1326,25 @@ protected void handleProducer(final CommandProducer cmdProducer) {
producerId, schema == null ? "absent" : "present");
}
- service.getOrCreateTopic(topicName.toString()).thenCompose((Topic topic) -> {
+ boolean createTopicIfDoesNotExist = isAuthorizedToCreateTopic
+ && service.isAllowAutoTopicCreation(topicName.toString());
+ service.getTopic(topicName.toString(), createTopicIfDoesNotExist).thenCompose(optTopic -> {
+ if (optTopic.isEmpty()) {
+ if (isAuthorizedToCreateTopic) {
+ return FutureUtil
+ .failedFuture(new TopicNotFoundException(
+ "Topic " + topicName + " does not exist"));
+ } else {
+ String msg = "Topic to produce does not exists and the Client is not"
+ + " authorized to create topic";
+ log.warn("[{}] {} with role {}", remoteAddress, msg, getPrincipal());
+ ctx.writeAndFlush(Commands.newError(requestId, ServerError.AuthorizationError,
+ msg));
+ return null;
Review Comment:
See previous thread.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] KannarFr commented on pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
KannarFr commented on PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#issuecomment-1251103337
@gaoran10 the failing tests are due to CI flakyness, nope?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] MarvinCai commented on a diff in pull request #17411: Check create topic permission on topic creation using pulsar proto clients
Posted by GitBox <gi...@apache.org>.
MarvinCai commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r966924574
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java:
##########
@@ -482,12 +482,12 @@ public CompletableFuture<Boolean> allowNamespaceOperationAsync(NamespaceName nam
case GET_TOPIC:
case GET_TOPICS:
case GET_BUNDLE:
+ case CREATE_TOPIC:
return allowConsumeOrProduceOpsAsync(namespaceName, role, authData);
Review Comment:
I see AuthorizationProviderBiscuit is check for namespace operator in `allowNamespaceOperationAsync`, but I don't see how moving up the `case CREATE_TOPIC:` in PulsarAuthorizationProvider is able solving the issue you described?
A client that subscribe to a topic that does not exist will still able to create the topic cause it has consume permission?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Let authorization providers restrict topic auto-creation [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #17411:
URL: https://github.com/apache/pulsar/pull/17411#discussion_r4318723664
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -718,6 +718,29 @@ private CompletableFuture<Boolean> isTopicOperationAllowed(TopicName topicName,
}
}
+ /**
+ * Whether a topic that does not exist may be created for this client: the auto topic creation settings allow it
+ * and the authorization provider lets this client trigger it.
+ */
+ public CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(TopicName topicName) {
+ return service.isAllowAutoTopicCreationAsync(topicName).thenCompose(isAllowed -> {
+ if (!isAllowed || !service.isAuthorizationEnabled()) {
+ return CompletableFuture.completedFuture(isAllowed);
+ }
+ return service.getAuthorizationService().allowTopicAutoCreationAsync(topicName, originalPrincipal,
+ authRole, originalAuthData != null ? originalAuthData : authenticationData, authenticationData);
+ });
+ }
+
+ /**
+ * {@link BrokerService#getOrCreateTopic(String)} for this client, see {@link #isAllowAutoTopicCreationAsync}.
+ */
+ private CompletableFuture<Topic> getOrCreateTopic(TopicName topicName) {
Review Comment:
Thanks for going further than the producer path. Loading first and only asking the provider for a missing topic is in one place now ([`BrokerService.java:1272-1291`](https://github.com/apache/pulsar/blob/66c62c45765babbd64dc55db2c2cadf419e5dbf8/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java#L1272-L1291)), and `testExistingTopicDoesNotDependOnAutoCreationCheck` shows existing topics keep working when the provider fails. I left one small, non-blocking note on the eviction line in that helper. Resolving this thread.
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicAutoCreationAuthorizationTest.java:
##########
@@ -0,0 +1,206 @@
+/*
+ * 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.client.api;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import com.google.common.collect.Sets;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
+import org.apache.pulsar.broker.namespace.TopicExistsInfo;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.AuthorizationProducerConsumerTest.ClientAuthentication;
+import org.apache.pulsar.client.api.AuthorizationProducerConsumerTest.TestAuthenticationProvider;
+import org.apache.pulsar.client.api.AuthorizationProducerConsumerTest.TestAuthorizationProvider;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.common.policies.data.TopicOperation;
+import org.apache.pulsar.common.policies.data.TopicType;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+/**
+ * Topic auto-creation goes through {@code AuthorizationProvider#allowTopicAutoCreationAsync}.
+ *
+ * <p>Runs its own broker rather than {@code SharedPulsarBaseTest}, because it needs authentication and a custom
+ * authorization provider.
+ */
+@Test(groups = "broker-api")
+public class TopicAutoCreationAuthorizationTest extends ProducerConsumerBase {
+
+ private static final String CREATOR_ROLE = "creator";
+ private static final String USER_ROLE = "user";
+ private static final String NAMESPACE = "my-property/my-ns";
+
+ private TopicType autoCreationType = TopicType.NON_PARTITIONED;
+ private PulsarAdmin superAdmin;
+
+ /**
+ * Lets both roles produce, consume and look up, but only {@link #CREATOR_ROLE} trigger topic auto-creation.
+ */
+ public static class AutoCreationAuthorizationProvider extends TestAuthorizationProvider {
+
+ private static final Set<String> CLIENT_ROLES = Set.of(CREATOR_ROLE, USER_ROLE);
+
+ @Override
+ public CompletableFuture<Boolean> allowTopicOperationAsync(TopicName topic, String role,
+ TopicOperation operation,
+ AuthenticationDataSource authData) {
+ return CompletableFuture.completedFuture(CLIENT_ROLES.contains(role));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> allowTopicAutoCreationAsync(TopicName topic, String role,
+ AuthenticationDataSource authData) {
+ return CompletableFuture.completedFuture(CREATOR_ROLE.equals(role));
+ }
+ }
+
+ @BeforeMethod(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ conf.setAuthenticationEnabled(true);
+ conf.setAuthorizationEnabled(true);
+ conf.setSuperUserRoles(Set.of("superUser"));
+ conf.setAuthenticationProviders(Set.of(TestAuthenticationProvider.class.getName()));
+ conf.setAuthorizationProvider(AutoCreationAuthorizationProvider.class.getName());
+ conf.setAllowAutoTopicCreation(true);
+ conf.setAllowAutoTopicCreationType(autoCreationType);
+ conf.setTopicLevelPoliciesEnabled(false);
+ conf.setClusterName("test");
+ super.init();
+
+ superAdmin = PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString())
+ .authentication(new ClientAuthentication("superUser")).build();
+ superAdmin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
+ superAdmin.tenants().createTenant("my-property",
+ new TenantInfoImpl(Sets.newHashSet(), Sets.newHashSet("test")));
+ superAdmin.namespaces().createNamespace(NAMESPACE, Sets.newHashSet("test"));
+ }
+
+ @AfterMethod(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ if (superAdmin != null) {
+ superAdmin.close();
+ superAdmin = null;
+ }
+ super.internalCleanup();
+ }
+
+ private PulsarClient newClient(String role, String serviceUrl) throws PulsarClientException {
+ return PulsarClient.builder().serviceUrl(serviceUrl)
+ .authentication(new ClientAuthentication(role))
+ .operationTimeout(3, TimeUnit.SECONDS)
+ .build();
+ }
+
+ private boolean topicExists(String topic) throws Exception {
+ TopicExistsInfo info = pulsar.getNamespaceService().checkTopicExistsAsync(TopicName.get(topic)).get();
+ try {
+ return info.isExists();
+ } finally {
+ info.recycle();
+ }
+ }
+
+ @Test(timeOut = 60000)
+ public void testProducerAutoCreationFollowsAuthorization() throws Exception {
Review Comment:
This covers what I was missing, thank you: the proxied binary and HTTP paths, admin create-subscription, HTTP partitioned metadata, the scalable-topic lookup and the existing-partitions case, with the manual `cleanup()`/`setup()` gone. One optional idea, not required for this PR: the test providers only look at the role, so a provider that records the `AuthenticationDataSource` it receives would also pin which auth data the HTTP path hands to the check. Resolving.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org