You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by xy...@apache.org on 2022/08/05 18:53:02 UTC
[pulsar] branch branch-2.8 updated (673ed2ce70a -> 6b3e46f20e7)
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a change to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 673ed2ce70a Revert "[fix][client-c++] Close `messages_` when PartitionedConsumer is closed (#16887)"
new 735e823d22c [fix][authorization] Fix multiple roles authorization (#16645)
new d461948b7e5 [fix][client] Remove redundant check for chunked message TotalChunkMsgSize in ConsumerImpl (#16797)
new acb4eba0ec4 [improve][authentication] Improve get the basic authentication config (#16526)
new 05b16e24ba8 [improve][test] Verify the authentication data in the authorization provider (#16900)
new 71076570b4b Fix the compilation error when cherry-picking cdec98a
new a50159328b1 Forget to update memory usage when invalid message (#16835)
new 59339c42b54 [fix][client]Fix MaxQueueSize semaphore release leak in createOpSendMsg (#16915)
new 6b3e46f20e7 Fix testProducerSemaphoreInvalidMessage by removing usages of mockStatic
The 8 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
conf/broker.conf | 7 +
conf/proxy.conf | 7 +
conf/standalone.conf | 6 +
.../AuthenticationProviderBasic.java | 49 ++--
.../MultiRolesTokenAuthorizationProvider.java | 83 +++++-
.../authorization/PulsarAuthorizationProvider.java | 2 +-
.../AuthenticationProviderBasicTest.java | 91 +++++++
.../test/resources/authentication/basic/.htpasswd | 0
.../broker/auth/AuthorizationWithAuthDataTest.java | 298 +++++++++++++++++++++
.../MultiRolesTokenAuthorizationProviderTest.java | 231 ++++++++++++++++
.../client/impl/ProducerMemoryLimitTest.java | 27 ++
.../pulsar/client/impl/ProducerSemaphoreTest.java | 35 +++
.../client/impl/BatchMessageContainerImpl.java | 3 +
.../apache/pulsar/client/impl/ConsumerImpl.java | 8 +-
14 files changed, 815 insertions(+), 32 deletions(-)
create mode 100644 pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authentication/AuthenticationProviderBasicTest.java
copy {pulsar-broker => pulsar-broker-common}/src/test/resources/authentication/basic/.htpasswd (100%)
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationWithAuthDataTest.java
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiRolesTokenAuthorizationProviderTest.java
[pulsar] 06/08: Forget to update memory usage when invalid message (#16835)
Posted by xy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a50159328b1292abeb8c050f893c826f28ad6fb9
Author: ZhangJian He <sh...@gmail.com>
AuthorDate: Fri Jul 29 10:43:50 2022 +0800
Forget to update memory usage when invalid message (#16835)
### Modifications
release memory usage when invalid message.
Only need to release memory usage here, no need to release semaphore. Both add testcases.
coauthored by @pengxiangrui127.
### Verifying this change
- add unit tests for this change
### Documentation
Check the box below or label this PR directly.
Need to update docs?
- [x] `doc-not-needed`
bug fix, no need doc
(cherry picked from commit 57b008a411463bce4c26350177dae4346f7b84d2)
---
.../client/impl/ProducerMemoryLimitTest.java | 27 ++++++++++++++++++++++
.../pulsar/client/impl/ProducerSemaphoreTest.java | 26 +++++++++++++++++++++
.../client/impl/BatchMessageContainerImpl.java | 2 ++
3 files changed, 55 insertions(+)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerMemoryLimitTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerMemoryLimitTest.java
index 264ec306413..0856dfc88b2 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerMemoryLimitTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerMemoryLimitTest.java
@@ -23,6 +23,8 @@ import org.apache.pulsar.client.api.ProducerConsumerBase;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SizeUnit;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -47,6 +49,31 @@ public class ProducerMemoryLimitTest extends ProducerConsumerBase {
super.internalCleanup();
}
+ @Test(timeOut = 10_000)
+ public void testProducerInvalidMessageMemoryRelease() throws Exception {
+ initClientWithMemoryLimit();
+ @Cleanup
+ ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) pulsarClient.newProducer()
+ .topic("testProducerMemoryLimit")
+ .sendTimeout(5, TimeUnit.SECONDS)
+ .batchingMaxPublishDelay(100, TimeUnit.MILLISECONDS)
+ .batchingMaxBytes(10240)
+ .enableBatching(true)
+ .create();
+ this.stopBroker();
+ try {
+ try (MockedStatic<ClientCnx> mockedStatic = Mockito.mockStatic(ClientCnx.class)) {
+ mockedStatic.when(ClientCnx::getMaxMessageSize).thenReturn(8);
+ producer.send("memory-test".getBytes(StandardCharsets.UTF_8));
+ }
+ throw new IllegalStateException("can not reach here");
+ } catch (PulsarClientException.InvalidMessageException ex) {
+ PulsarClientImpl clientImpl = (PulsarClientImpl) this.pulsarClient;
+ final MemoryLimitController memoryLimitController = clientImpl.getMemoryLimitController();
+ Assert.assertEquals(memoryLimitController.currentUsage(), 0);
+ }
+ }
+
@Test(timeOut = 10_000)
public void testProducerTimeoutMemoryRelease() throws Exception {
initClientWithMemoryLimit();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
index d325ed67852..b095b1b90a9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
@@ -25,12 +25,15 @@ import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.util.FutureUtil;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
@@ -55,6 +58,29 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
super.internalCleanup();
}
+ @Test(timeOut = 10_000)
+ public void testProducerSemaphoreInvalidMessage() throws Exception {
+ final int pendingQueueSize = 100;
+
+ @Cleanup
+ ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) pulsarClient.newProducer()
+ .topic("testProducerSemaphoreAcquire")
+ .maxPendingMessages(pendingQueueSize)
+ .enableBatching(false)
+ .create();
+
+ this.stopBroker();
+ try {
+ try (MockedStatic<ClientCnx> mockedStatic = Mockito.mockStatic(ClientCnx.class)) {
+ mockedStatic.when(ClientCnx::getMaxMessageSize).thenReturn(2);
+ producer.send("semaphore-test".getBytes(StandardCharsets.UTF_8));
+ }
+ throw new IllegalStateException("can not reach here");
+ } catch (PulsarClientException.InvalidMessageException ex) {
+ Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
+ }
+ }
+
@Test(timeOut = 30000)
public void testProducerSemaphoreAcquireAndRelease() throws PulsarClientException, ExecutionException, InterruptedException {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java
index 5811db76a64..2d08e098de4 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java
@@ -200,6 +200,8 @@ class BatchMessageContainerImpl extends AbstractBatchMessageContainer {
public OpSendMsg createOpSendMsg() throws IOException {
ByteBuf encryptedPayload = producer.encryptMessage(messageMetadata, getCompressedBatchMetadataAndPayload());
if (encryptedPayload.readableBytes() > ClientCnx.getMaxMessageSize()) {
+ messages.forEach(msg -> producer.client.getMemoryLimitController()
+ .releaseMemory(msg.getUncompressedSize()));
discard(new PulsarClientException.InvalidMessageException(
"Message size is bigger than " + ClientCnx.getMaxMessageSize() + " bytes"));
return null;
[pulsar] 07/08: [fix][client]Fix MaxQueueSize semaphore release leak in createOpSendMsg (#16915)
Posted by xy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 59339c42b54856b54a877489af1e8b8418095045
Author: lixinyang <84...@users.noreply.github.com>
AuthorDate: Wed Aug 3 12:09:13 2022 +0800
[fix][client]Fix MaxQueueSize semaphore release leak in createOpSendMsg (#16915)
(cherry picked from commit d95f6cf366f66bc1e38711bc59cd456c8f53f888)
---
.../apache/pulsar/client/impl/ProducerSemaphoreTest.java | 13 ++++++++++++-
.../pulsar/client/impl/BatchMessageContainerImpl.java | 1 +
2 files changed, 13 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
index b095b1b90a9..8e2876d0623 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
@@ -66,7 +66,7 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) pulsarClient.newProducer()
.topic("testProducerSemaphoreAcquire")
.maxPendingMessages(pendingQueueSize)
- .enableBatching(false)
+ .enableBatching(true)
.create();
this.stopBroker();
@@ -79,6 +79,17 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
} catch (PulsarClientException.InvalidMessageException ex) {
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
}
+
+ producer.conf.setBatchingEnabled(false);
+ try {
+ try (MockedStatic<ClientCnx> mockedStatic = Mockito.mockStatic(ClientCnx.class)) {
+ mockedStatic.when(ClientCnx::getMaxMessageSize).thenReturn(2);
+ producer.send("semaphore-test".getBytes(StandardCharsets.UTF_8));
+ }
+ throw new IllegalStateException("can not reach here");
+ } catch (PulsarClientException.InvalidMessageException ex) {
+ Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
+ }
}
@Test(timeOut = 30000)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java
index 2d08e098de4..bf03d80ce0e 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java
@@ -200,6 +200,7 @@ class BatchMessageContainerImpl extends AbstractBatchMessageContainer {
public OpSendMsg createOpSendMsg() throws IOException {
ByteBuf encryptedPayload = producer.encryptMessage(messageMetadata, getCompressedBatchMetadataAndPayload());
if (encryptedPayload.readableBytes() > ClientCnx.getMaxMessageSize()) {
+ producer.semaphoreRelease(messages.size());
messages.forEach(msg -> producer.client.getMemoryLimitController()
.releaseMemory(msg.getUncompressedSize()));
discard(new PulsarClientException.InvalidMessageException(
[pulsar] 05/08: Fix the compilation error when cherry-picking cdec98a
Posted by xy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 71076570b4bc3af30dd5da6b918f1ea232d92afb
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Thu Aug 4 15:44:28 2022 +0800
Fix the compilation error when cherry-picking cdec98a
---
.../apache/pulsar/broker/auth/AuthorizationWithAuthDataTest.java | 9 ++++++++-
1 file changed, 8 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationWithAuthDataTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationWithAuthDataTest.java
index ccb401f74b3..7476eb37da7 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationWithAuthDataTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationWithAuthDataTest.java
@@ -25,6 +25,7 @@ import io.jsonwebtoken.SignatureAlgorithm;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
+import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
@@ -38,6 +39,7 @@ import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
import org.apache.pulsar.broker.authentication.AuthenticationProviderToken;
import org.apache.pulsar.broker.authentication.utils.AuthTokenUtils;
import org.apache.pulsar.broker.authorization.AuthorizationProvider;
+import org.apache.pulsar.broker.cache.ConfigurationCacheService;
import org.apache.pulsar.client.admin.PulsarAdminBuilder;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.ClientBuilder;
@@ -96,6 +98,11 @@ public class AuthorizationWithAuthDataTest extends MockedPulsarServiceBaseTest {
return CompletableFuture.completedFuture(true);
}
+ @Override
+ public void initialize(ServiceConfiguration conf, ConfigurationCacheService configCache) throws IOException {
+ // No ops
+ }
+
@Override
public CompletableFuture<Boolean> canProduceAsync(TopicName topicName, String role,
AuthenticationDataSource authenticationData) {
@@ -262,7 +269,7 @@ public class AuthorizationWithAuthDataTest extends MockedPulsarServiceBaseTest {
@Test
public void testAdmin() throws PulsarAdminException {
admin.tenants().createTenant("test-tenant-1",
- TenantInfo.builder().allowedClusters(Set.of(configClusterName)).build());
+ TenantInfo.builder().allowedClusters(Collections.singleton(configClusterName)).build());
admin.namespaces().createNamespace("test-tenant-1/test-namespace-1");
String partitionedTopic = UUID.randomUUID().toString();
admin.topics().createPartitionedTopic(partitionedTopic,3);
[pulsar] 01/08: [fix][authorization] Fix multiple roles authorization (#16645)
Posted by xy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 735e823d22cb24336b9a95ec8227d14b22ea9f05
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Mon Jul 25 17:57:19 2022 +0800
[fix][authorization] Fix multiple roles authorization (#16645)
(cherry picked from commit d8483d48cb21e8e99fd56c786e5198f7fe7135f6)
---
.../MultiRolesTokenAuthorizationProvider.java | 83 +++++++-
.../authorization/PulsarAuthorizationProvider.java | 2 +-
.../MultiRolesTokenAuthorizationProviderTest.java | 231 +++++++++++++++++++++
3 files changed, 304 insertions(+), 12 deletions(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java
index ba7d580e88f..9eb07b36754 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java
@@ -26,9 +26,12 @@ import io.jsonwebtoken.RequiredTypeException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
+import javax.ws.rs.core.Response;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
@@ -38,9 +41,12 @@ import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.NamespaceOperation;
import org.apache.pulsar.common.policies.data.PolicyName;
import org.apache.pulsar.common.policies.data.PolicyOperation;
+import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.common.policies.data.TenantOperation;
import org.apache.pulsar.common.policies.data.TopicOperation;
import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.common.util.RestException;
+import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -79,56 +85,111 @@ public class MultiRolesTokenAuthorizationProvider extends PulsarAuthorizationPro
super.initialize(conf, configCache);
}
- private List<String> getRoles(AuthenticationDataSource authData) {
+ @Override
+ public CompletableFuture<Boolean> isSuperUser(String role, AuthenticationDataSource authenticationData,
+ ServiceConfiguration serviceConfiguration) {
+ Set<String> roles = getRoles(authenticationData);
+ if (roles.isEmpty()) {
+ return CompletableFuture.completedFuture(false);
+ }
+ Set<String> superUserRoles = serviceConfiguration.getSuperUserRoles();
+ if (superUserRoles.isEmpty()) {
+ return CompletableFuture.completedFuture(false);
+ }
+
+ return CompletableFuture.completedFuture(roles.stream().anyMatch(superUserRoles::contains));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> validateTenantAdminAccess(String tenantName, String role,
+ AuthenticationDataSource authData) {
+ return isSuperUser(role, authData, conf)
+ .thenCompose(isSuperUser -> {
+ if (isSuperUser) {
+ return CompletableFuture.completedFuture(true);
+ }
+ Set<String> roles = getRoles(authData);
+ if (roles.isEmpty()) {
+ return CompletableFuture.completedFuture(false);
+ }
+
+ return pulsarResources.getTenantResources()
+ .getTenantAsync(tenantName)
+ .thenCompose(op -> {
+ if (op.isPresent()) {
+ TenantInfo tenantInfo = op.get();
+ if (tenantInfo.getAdminRoles() == null || tenantInfo.getAdminRoles().isEmpty()) {
+ return CompletableFuture.completedFuture(false);
+ }
+
+ return CompletableFuture.completedFuture(roles.stream()
+ .anyMatch(n -> tenantInfo.getAdminRoles().contains(n)));
+ } else {
+ throw new RestException(Response.Status.NOT_FOUND, "Tenant does not exist");
+ }
+ }).exceptionally(ex -> {
+ Throwable cause = ex.getCause();
+ if (cause instanceof MetadataStoreException.NotFoundException) {
+ log.warn("Failed to get tenant info data for non existing tenant {}", tenantName);
+ throw new RestException(Response.Status.NOT_FOUND, "Tenant does not exist");
+ }
+ log.error("Failed to get tenant {}", tenantName, cause);
+ throw new RestException(cause);
+ });
+ });
+ }
+
+ private Set<String> getRoles(AuthenticationDataSource authData) {
String token = null;
if (authData.hasDataFromCommand()) {
// Authenticate Pulsar binary connection
token = authData.getCommandData();
if (StringUtils.isBlank(token)) {
- return Collections.emptyList();
+ return Collections.emptySet();
}
} else if (authData.hasDataFromHttp()) {
// The format here should be compliant to RFC-6750
// (https://tools.ietf.org/html/rfc6750#section-2.1). Eg: Authorization: Bearer xxxxxxxxxxxxx
String httpHeaderValue = authData.getHttpHeader(HTTP_HEADER_NAME);
if (httpHeaderValue == null || !httpHeaderValue.startsWith(HTTP_HEADER_VALUE_PREFIX)) {
- return Collections.emptyList();
+ return Collections.emptySet();
}
// Remove prefix
token = httpHeaderValue.substring(HTTP_HEADER_VALUE_PREFIX.length());
}
- if (token == null)
- return Collections.emptyList();
+ if (token == null) {
+ return Collections.emptySet();
+ }
String[] splitToken = token.split("\\.");
if (splitToken.length < 2) {
log.warn("Unable to extract additional roles from JWT token");
- return Collections.emptyList();
+ return Collections.emptySet();
}
String unsignedToken = splitToken[0] + "." + splitToken[1] + ".";
Jwt<?, Claims> jwt = parser.parseClaimsJwt(unsignedToken);
try {
- return Collections.singletonList(jwt.getBody().get(roleClaim, String.class));
+ return new HashSet<>(Collections.singletonList(jwt.getBody().get(roleClaim, String.class)));
} catch (RequiredTypeException requiredTypeException) {
try {
List list = jwt.getBody().get(roleClaim, List.class);
if (list != null) {
- return list;
+ return new HashSet<String>(list);
}
} catch (RequiredTypeException requiredTypeException1) {
- return Collections.emptyList();
+ return Collections.emptySet();
}
}
- return Collections.emptyList();
+ return Collections.emptySet();
}
public CompletableFuture<Boolean> authorize(AuthenticationDataSource authenticationData, Function<String, CompletableFuture<Boolean>> authorizeFunc) {
- List<String> roles = getRoles(authenticationData);
+ Set<String> roles = getRoles(authenticationData);
if (roles.isEmpty()) {
return CompletableFuture.completedFuture(false);
}
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
index 527201ae0cf..f65445553ca 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
@@ -60,7 +60,7 @@ public class PulsarAuthorizationProvider implements AuthorizationProvider {
private static final Logger log = LoggerFactory.getLogger(PulsarAuthorizationProvider.class);
public ServiceConfiguration conf;
- private PulsarResources pulsarResources;
+ protected PulsarResources pulsarResources;
private static final String POLICY_ROOT = "/admin/policies/";
public static final String POLICIES = "policies";
private static final String POLICIES_READONLY_FLAG_PATH = "/admin/flags/policies-readonly";
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiRolesTokenAuthorizationProviderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiRolesTokenAuthorizationProviderTest.java
new file mode 100644
index 00000000000..12d7c71358b
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiRolesTokenAuthorizationProviderTest.java
@@ -0,0 +1,231 @@
+/**
+ * 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.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertThrows;
+import com.google.common.collect.Sets;
+import io.jsonwebtoken.Jwts;
+import io.jsonwebtoken.SignatureAlgorithm;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import javax.crypto.SecretKey;
+import lombok.Cleanup;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.authentication.AuthenticationProviderToken;
+import org.apache.pulsar.broker.authentication.utils.AuthTokenUtils;
+import org.apache.pulsar.broker.authorization.MultiRolesTokenAuthorizationProvider;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminBuilder;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.impl.auth.AuthenticationToken;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.TenantInfo;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+public class MultiRolesTokenAuthorizationProviderTest extends MockedPulsarServiceBaseTest {
+
+ private final SecretKey secretKey = AuthTokenUtils.createSecretKey(SignatureAlgorithm.HS256);
+ private final String superUserToken;
+ private final String normalUserToken;
+
+ public MultiRolesTokenAuthorizationProviderTest() {
+ Map<String, Object> claims = new HashMap<>();
+ Set<String> roles = new HashSet<>();
+ roles.add("user1");
+ roles.add("superUser");
+ claims.put("roles", roles);
+ superUserToken = Jwts.builder()
+ .setClaims(claims)
+ .signWith(secretKey)
+ .compact();
+
+ roles = new HashSet<>();
+ roles.add("normalUser");
+ roles.add("user2");
+ roles.add("user5");
+ claims.put("roles", roles);
+ normalUserToken = Jwts.builder()
+ .setClaims(claims)
+ .signWith(secretKey)
+ .compact();
+ }
+
+ @Override
+ protected void doInitConf() throws Exception {
+ super.doInitConf();
+
+ conf.setAuthenticationEnabled(true);
+ conf.setAuthorizationEnabled(true);
+
+ Set<String> superUserRoles = new HashSet<>();
+ superUserRoles.add("superUser");
+ conf.setSuperUserRoles(superUserRoles);
+
+ Properties properties = new Properties();
+ properties.setProperty("tokenSecretKey",
+ "data:;base64," + Base64.getEncoder().encodeToString(secretKey.getEncoded()));
+ properties.setProperty("tokenAuthClaim", "roles");
+ conf.setProperties(properties);
+
+ conf.setBrokerClientAuthenticationPlugin(AuthenticationToken.class.getName());
+ conf.setBrokerClientAuthenticationParameters(superUserToken);
+
+ Set<String> providers = new HashSet<>();
+ providers.add(AuthenticationProviderToken.class.getName());
+ conf.setAuthenticationProviders(providers);
+ conf.setAuthorizationProvider(MultiRolesTokenAuthorizationProvider.class.getName());
+
+ conf.setClusterName(configClusterName);
+ conf.setNumExecutorThreadPoolSize(5);
+ }
+
+ @BeforeClass
+ @Override
+ protected void setup() throws Exception {
+ super.internalSetup();
+
+ admin.clusters().createCluster(configClusterName,
+ ClusterData.builder()
+ .brokerServiceUrl(brokerUrl.toString())
+ .serviceUrl(getPulsar().getWebServiceAddress())
+ .build()
+ );
+ }
+
+ @BeforeClass
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @Override
+ protected void customizeNewPulsarClientBuilder(ClientBuilder clientBuilder) {
+ clientBuilder.authentication(new AuthenticationToken(superUserToken));
+ }
+
+ @Override
+ protected void customizeNewPulsarAdminBuilder(PulsarAdminBuilder pulsarAdminBuilder) {
+ pulsarAdminBuilder.authentication(new AuthenticationToken(superUserToken));
+ }
+
+ private PulsarAdmin newPulsarAdmin(String token) throws PulsarClientException {
+ return PulsarAdmin.builder()
+ .serviceHttpUrl(pulsar.getWebServiceAddress())
+ .authentication(new AuthenticationToken(token))
+ .requestTimeout(3, TimeUnit.SECONDS)
+ .build();
+ }
+
+ private PulsarClient newPulsarClient(String token) throws PulsarClientException {
+ return PulsarClient.builder()
+ .serviceUrl(pulsar.getBrokerServiceUrl())
+ .authentication(new AuthenticationToken(token))
+ .operationTimeout(3, TimeUnit.SECONDS)
+ .build();
+ }
+
+ @Test
+ public void testAdminRequestWithSuperUserToken() throws Exception {
+ String tenant = "superuser-admin-tenant";
+ @Cleanup
+ PulsarAdmin admin = newPulsarAdmin(superUserToken);
+ admin.tenants().createTenant(tenant, TenantInfo.builder()
+ .allowedClusters(Sets.newHashSet(configClusterName)).build());
+ String namespace = "superuser-admin-namespace";
+ admin.namespaces().createNamespace(tenant + "/" + namespace);
+ admin.brokers().getAllDynamicConfigurations();
+ admin.tenants().getTenants();
+ admin.topics().getList(tenant + "/" + namespace);
+ }
+
+ @Test
+ public void testProduceAndConsumeWithSuperUserToken() throws Exception {
+ String tenant = "superuser-client-tenant";
+ @Cleanup
+ PulsarAdmin admin = newPulsarAdmin(superUserToken);
+ admin.tenants().createTenant(tenant, TenantInfo.builder()
+ .allowedClusters(Sets.newHashSet(configClusterName)).build());
+ String namespace = "superuser-client-namespace";
+ admin.namespaces().createNamespace(tenant + "/" + namespace);
+ String topic = tenant + "/" + namespace + "/" + "test-topic";
+
+ @Cleanup
+ PulsarClient client = newPulsarClient(superUserToken);
+ @Cleanup
+ Producer<byte[]> producer = client.newProducer().topic(topic).create();
+ byte[] body = "hello".getBytes(StandardCharsets.UTF_8);
+ producer.send(body);
+
+ @Cleanup
+ Consumer<byte[]> consumer = client.newConsumer().topic(topic)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscriptionName("test")
+ .subscribe();
+ Message<byte[]> message = consumer.receive(3, TimeUnit.SECONDS);
+ assertNotNull(message);
+ assertEquals(message.getData(), body);
+ }
+
+ @Test
+ public void testAdminRequestWithNormalUserToken() throws Exception {
+ String tenant = "normaluser-admin-tenant";
+ @Cleanup
+ PulsarAdmin admin = newPulsarAdmin(normalUserToken);
+
+ assertThrows(PulsarAdminException.NotAuthorizedException.class,
+ () -> admin.tenants().createTenant(tenant, TenantInfo.builder()
+ .allowedClusters(Sets.newHashSet(configClusterName)).build()));
+ }
+
+ @Test
+ public void testProduceAndConsumeWithNormalUserToken() throws Exception {
+ String tenant = "normaluser-client-tenant";
+ @Cleanup
+ PulsarAdmin admin = newPulsarAdmin(superUserToken);
+ admin.tenants().createTenant(tenant, TenantInfo.builder()
+ .allowedClusters(Sets.newHashSet(configClusterName)).build());
+ String namespace = "normaluser-client-namespace";
+ admin.namespaces().createNamespace(tenant + "/" + namespace);
+ String topic = tenant + "/" + namespace + "/" + "test-topic";
+
+ @Cleanup
+ PulsarClient client = newPulsarClient(normalUserToken);
+ assertThrows(PulsarClientException.AuthorizationException.class, () -> {
+ @Cleanup
+ Producer<byte[]> ignored = client.newProducer().topic(topic).create();
+ });
+
+ assertThrows(PulsarClientException.AuthorizationException.class, () -> {
+ @Cleanup
+ Consumer<byte[]> ignored = client.newConsumer().topic(topic)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscriptionName("test")
+ .subscribe();
+ });
+ }
+}
[pulsar] 02/08: [fix][client] Remove redundant check for chunked message TotalChunkMsgSize in ConsumerImpl (#16797)
Posted by xy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d461948b7e5f468538b4ff6e4c337c88c2878e16
Author: Jiaqi Shen <50...@users.noreply.github.com>
AuthorDate: Fri Aug 5 15:26:01 2022 +0800
[fix][client] Remove redundant check for chunked message TotalChunkMsgSize in ConsumerImpl (#16797)
### Motivation
There is a incorrect out-of-order check for chunked message in `ConsumerImpl`. For the last check should compare the result of `chunkedMsgCtx.chunkedMsgBuffer.readableBytes() + compressedPayload.readableBytes()` with `TotalChunkMsgSize` instead of `ChunkId` with `TotalChunkMsgSize`.
### Modifications
Fix the out-of-order check for chunked message in `ConsumerImpl`.
(cherry picked from commit 7650612cd86ead01886300f9b31aba6946831736)
---
.../src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java | 8 +++-----
1 file changed, 3 insertions(+), 5 deletions(-)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index 974d27ddb3f..a6e79cf3136 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -1133,12 +1133,10 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
ChunkedMessageCtx chunkedMsgCtx = chunkedMessagesMap.get(msgMetadata.getUuid());
// discard message if chunk is out-of-order
if (chunkedMsgCtx == null || chunkedMsgCtx.chunkedMsgBuffer == null
- || msgMetadata.getChunkId() != (chunkedMsgCtx.lastChunkedMessageId + 1)
- || msgMetadata.getChunkId() >= msgMetadata.getTotalChunkMsgSize()) {
+ || msgMetadata.getChunkId() != (chunkedMsgCtx.lastChunkedMessageId + 1)) {
// means we lost the first chunk: should never happen
- log.info("Received unexpected chunk messageId {}, last-chunk-id{}, chunkId = {}, total-chunks {}", msgId,
- (chunkedMsgCtx != null ? chunkedMsgCtx.lastChunkedMessageId : null), msgMetadata.getChunkId(),
- msgMetadata.getTotalChunkMsgSize());
+ log.info("Received unexpected chunk messageId {}, last-chunk-id{}, chunkId = {}", msgId,
+ (chunkedMsgCtx != null ? chunkedMsgCtx.lastChunkedMessageId : null), msgMetadata.getChunkId());
if (chunkedMsgCtx != null) {
if (chunkedMsgCtx.chunkedMsgBuffer != null) {
ReferenceCountUtil.safeRelease(chunkedMsgCtx.chunkedMsgBuffer);
[pulsar] 08/08: Fix testProducerSemaphoreInvalidMessage by removing usages of mockStatic
Posted by xy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 6b3e46f20e7004890822d8b9e47ca7328097d3a0
Author: Yunze Xu <xy...@163.com>
AuthorDate: Sat Aug 6 02:51:19 2022 +0800
Fix testProducerSemaphoreInvalidMessage by removing usages of mockStatic
---
.../pulsar/client/impl/ProducerSemaphoreTest.java | 18 ++++++++----------
1 file changed, 8 insertions(+), 10 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
index 8e2876d0623..4d3c36de908 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
@@ -25,13 +25,12 @@ import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.util.FutureUtil;
-import org.mockito.MockedStatic;
-import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+import java.lang.reflect.Field;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
@@ -70,11 +69,13 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
.create();
this.stopBroker();
+
+ Field maxMessageSizeFiled = ClientCnx.class.getDeclaredField("maxMessageSize");
+ maxMessageSizeFiled.setAccessible(true);
+ maxMessageSizeFiled.set(null, 2);
+
try {
- try (MockedStatic<ClientCnx> mockedStatic = Mockito.mockStatic(ClientCnx.class)) {
- mockedStatic.when(ClientCnx::getMaxMessageSize).thenReturn(2);
- producer.send("semaphore-test".getBytes(StandardCharsets.UTF_8));
- }
+ producer.send("semaphore-test".getBytes(StandardCharsets.UTF_8));
throw new IllegalStateException("can not reach here");
} catch (PulsarClientException.InvalidMessageException ex) {
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
@@ -82,10 +83,7 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
producer.conf.setBatchingEnabled(false);
try {
- try (MockedStatic<ClientCnx> mockedStatic = Mockito.mockStatic(ClientCnx.class)) {
- mockedStatic.when(ClientCnx::getMaxMessageSize).thenReturn(2);
- producer.send("semaphore-test".getBytes(StandardCharsets.UTF_8));
- }
+ producer.send("semaphore-test".getBytes(StandardCharsets.UTF_8));
throw new IllegalStateException("can not reach here");
} catch (PulsarClientException.InvalidMessageException ex) {
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
[pulsar] 04/08: [improve][test] Verify the authentication data in the authorization provider (#16900)
Posted by xy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 05b16e24ba83cfd4b564f3ad33ad9c4067d6c381
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Thu Aug 4 15:44:28 2022 +0800
[improve][test] Verify the authentication data in the authorization provider (#16900)
Signed-off-by: Zixuan Liu <no...@gmail.com>
(cherry picked from commit cdec98a9a624d47125535e32e8eba2dd4bb4125f)
---
.../broker/auth/AuthorizationWithAuthDataTest.java | 291 +++++++++++++++++++++
1 file changed, 291 insertions(+)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationWithAuthDataTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationWithAuthDataTest.java
new file mode 100644
index 00000000000..ccb401f74b3
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationWithAuthDataTest.java
@@ -0,0 +1,291 @@
+/**
+ * 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.broker.auth;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import io.jsonwebtoken.Jwts;
+import io.jsonwebtoken.SignatureAlgorithm;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import javax.crypto.SecretKey;
+import lombok.Cleanup;
+import lombok.SneakyThrows;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
+import org.apache.pulsar.broker.authentication.AuthenticationProviderToken;
+import org.apache.pulsar.broker.authentication.utils.AuthTokenUtils;
+import org.apache.pulsar.broker.authorization.AuthorizationProvider;
+import org.apache.pulsar.client.admin.PulsarAdminBuilder;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.ClientBuilder;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.impl.auth.AuthenticationToken;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.AuthAction;
+import org.apache.pulsar.common.policies.data.NamespaceOperation;
+import org.apache.pulsar.common.policies.data.PolicyName;
+import org.apache.pulsar.common.policies.data.PolicyOperation;
+import org.apache.pulsar.common.policies.data.TenantInfo;
+import org.apache.pulsar.common.policies.data.TenantOperation;
+import org.apache.pulsar.common.policies.data.TopicOperation;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class AuthorizationWithAuthDataTest extends MockedPulsarServiceBaseTest {
+
+ private static final SecretKey SECRET_KEY = AuthTokenUtils.createSecretKey(SignatureAlgorithm.HS256);
+ private static final String ADMIN_ROLE = "admin";
+ private static final String ADMIN_TOKEN = Jwts.builder().setSubject(ADMIN_ROLE).signWith(SECRET_KEY).compact();
+
+ public static class MyAuthorizationProvider implements AuthorizationProvider {
+
+ public MyAuthorizationProvider() {
+ }
+
+ private void assertRoleAndAuthenticationData(String role, AuthenticationDataSource authenticationData) {
+ assertEquals(role, ADMIN_ROLE);
+ if (authenticationData.hasDataFromHttp()) {
+ String authorization = authenticationData.getHttpHeader("Authorization");
+ assertEquals(authorization, "Bearer " + ADMIN_TOKEN);
+ } else {
+ assertEquals(authenticationData.getCommandData(), ADMIN_TOKEN);
+ }
+ }
+
+ @Override
+ public CompletableFuture<Boolean> isSuperUser(String role, AuthenticationDataSource authenticationData,
+ ServiceConfiguration serviceConfiguration) {
+ assertRoleAndAuthenticationData(role, authenticationData);
+ return CompletableFuture.completedFuture(true);
+ }
+
+ @Override
+ public CompletableFuture<Boolean> isTenantAdmin(String tenant, String role, TenantInfo tenantInfo,
+ AuthenticationDataSource authenticationData) {
+ assertRoleAndAuthenticationData(role, authenticationData);
+ return CompletableFuture.completedFuture(true);
+ }
+
+ @Override
+ public CompletableFuture<Boolean> canProduceAsync(TopicName topicName, String role,
+ AuthenticationDataSource authenticationData) {
+ assertRoleAndAuthenticationData(role, authenticationData);
+ return CompletableFuture.completedFuture(true);
+ }
+
+ @Override
+ public CompletableFuture<Boolean> canConsumeAsync(TopicName topicName, String role,
+ AuthenticationDataSource authenticationData,
+ String subscription) {
+ assertRoleAndAuthenticationData(role, authenticationData);
+ return CompletableFuture.completedFuture(true);
+ }
+
+ @Override
+ public CompletableFuture<Boolean> canLookupAsync(TopicName topicName, String role,
+ AuthenticationDataSource authenticationData) {
+ assertRoleAndAuthenticationData(role, authenticationData);
+ return CompletableFuture.completedFuture(true);
+ }
+
+ @Override
+ public CompletableFuture<Boolean> allowFunctionOpsAsync(NamespaceName namespaceName, String role,
+ AuthenticationDataSource authenticationData) {
+ assertRoleAndAuthenticationData(role, authenticationData);
+ return CompletableFuture.completedFuture(true);
+ }
+
+ @Override
+ public CompletableFuture<Boolean> allowSourceOpsAsync(NamespaceName namespaceName, String role,
+ AuthenticationDataSource authenticationData) {
+ assertRoleAndAuthenticationData(role, authenticationData);
+ return CompletableFuture.completedFuture(true);
+ }
+
+ @Override
+ public CompletableFuture<Boolean> allowSinkOpsAsync(NamespaceName namespaceName, String role,
+ AuthenticationDataSource authenticationData) {
+ assertRoleAndAuthenticationData(role, authenticationData);
+ return CompletableFuture.completedFuture(true);
+ }
+
+ @Override
+ public CompletableFuture<Void> grantPermissionAsync(NamespaceName namespace, Set<AuthAction> actions,
+ String role, String authDataJson) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<Void> grantSubscriptionPermissionAsync(NamespaceName namespace,
+ String subscriptionName, Set<String> roles,
+ String authDataJson) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<Void> revokeSubscriptionPermissionAsync(NamespaceName namespace,
+ String subscriptionName, String role,
+ String authDataJson) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<Void> grantPermissionAsync(TopicName topicName, Set<AuthAction> actions, String role,
+ String authDataJson) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<Boolean> allowTenantOperationAsync(String tenantName, String role,
+ TenantOperation operation,
+ AuthenticationDataSource authData) {
+ assertRoleAndAuthenticationData(role, authData);
+ return CompletableFuture.completedFuture(true);
+ }
+
+ @Override
+ public CompletableFuture<Boolean> allowNamespaceOperationAsync(NamespaceName namespaceName, String role,
+ NamespaceOperation operation,
+ AuthenticationDataSource authData) {
+ assertRoleAndAuthenticationData(role, authData);
+ return CompletableFuture.completedFuture(true);
+ }
+
+ @Override
+ public CompletableFuture<Boolean> allowNamespacePolicyOperationAsync(NamespaceName namespaceName,
+ PolicyName policy,
+ PolicyOperation operation, String role,
+ AuthenticationDataSource authData) {
+ assertRoleAndAuthenticationData(role, authData);
+ return CompletableFuture.completedFuture(true);
+ }
+
+ @Override
+ public CompletableFuture<Boolean> allowTopicOperationAsync(TopicName topic, String role,
+ TopicOperation operation,
+ AuthenticationDataSource authData) {
+ assertRoleAndAuthenticationData(role, authData);
+ return CompletableFuture.completedFuture(true);
+ }
+
+ @Override
+ public CompletableFuture<Boolean> allowTopicPolicyOperationAsync(TopicName topic, String role,
+ PolicyName policy, PolicyOperation operation,
+ AuthenticationDataSource authData) {
+ assertRoleAndAuthenticationData(role, authData);
+ return CompletableFuture.completedFuture(true);
+ }
+
+ @Override
+ public void close() throws IOException {
+ // noop
+ }
+ }
+
+ @Override
+ protected void doInitConf() throws Exception {
+ super.doInitConf();
+ conf.setAuthenticationEnabled(true);
+ conf.getProperties().setProperty("tokenSecretKey", "data:;base64,"
+ + Base64.getEncoder().encodeToString(SECRET_KEY.getEncoded()));
+ Set<String> providers = new HashSet<>();
+ providers.add(AuthenticationProviderToken.class.getName());
+ conf.setAuthenticationProviders(providers);
+ Set<String> superUserRoles = new HashSet<>();
+ superUserRoles.add("admin");
+ conf.setSuperUserRoles(superUserRoles);
+
+ conf.setBrokerClientAuthenticationPlugin(AuthenticationToken.class.getName());
+ conf.setBrokerClientAuthenticationParameters(ADMIN_TOKEN);
+
+ conf.setAuthorizationEnabled(true);
+ conf.setAuthorizationProvider(MyAuthorizationProvider.class.getName());
+ }
+
+ @SneakyThrows
+ @Override
+ protected void customizeNewPulsarAdminBuilder(PulsarAdminBuilder pulsarAdminBuilder) {
+ super.customizeNewPulsarAdminBuilder(pulsarAdminBuilder);
+ pulsarAdminBuilder.authentication(AuthenticationToken.class.getName(), ADMIN_TOKEN);
+ }
+
+ @SneakyThrows
+ @Override
+ protected void customizeNewPulsarClientBuilder(ClientBuilder clientBuilder) {
+ super.customizeNewPulsarClientBuilder(clientBuilder);
+ clientBuilder.authentication(AuthenticationToken.class.getName(), ADMIN_TOKEN);
+ }
+
+ @BeforeClass(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ internalSetup();
+ setupDefaultTenantAndNamespace();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ internalCleanup();
+ }
+
+ @Test
+ public void testAdmin() throws PulsarAdminException {
+ admin.tenants().createTenant("test-tenant-1",
+ TenantInfo.builder().allowedClusters(Set.of(configClusterName)).build());
+ admin.namespaces().createNamespace("test-tenant-1/test-namespace-1");
+ String partitionedTopic = UUID.randomUUID().toString();
+ admin.topics().createPartitionedTopic(partitionedTopic,3);
+ String nonPartitionedTopic = UUID.randomUUID().toString();
+ admin.topics().createNonPartitionedTopic(nonPartitionedTopic);
+ admin.lookups().lookupPartitionedTopic(partitionedTopic);
+ admin.lookups().lookupTopic(nonPartitionedTopic);
+ }
+
+ @Test
+ public void testClient() throws PulsarClientException {
+ String topic = UUID.randomUUID().toString();
+ @Cleanup
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create();
+ byte[] msg = "Hello".getBytes(StandardCharsets.UTF_8);
+ producer.send(msg);
+
+ @Cleanup
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscriptionName("test").subscribe();
+ Message<byte[]> receive = consumer.receive(3, TimeUnit.SECONDS);
+ assertNotNull(receive);
+ assertEquals(receive.getData(), msg);
+ }
+}
[pulsar] 03/08: [improve][authentication] Improve get the basic authentication config (#16526)
Posted by xy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit acb4eba0ec4077f017ad5b9fb26d64ac224438a6
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Wed Aug 3 20:31:11 2022 +0800
[improve][authentication] Improve get the basic authentication config (#16526)
Signed-off-by: Zixuan Liu <no...@gmail.com>
(cherry picked from commit d32e1df89c4a88fe3f3a26e3ed9563d8c2a2ae21)
---
conf/broker.conf | 7 ++
conf/proxy.conf | 7 ++
conf/standalone.conf | 6 ++
.../AuthenticationProviderBasic.java | 49 ++++++++----
.../AuthenticationProviderBasicTest.java | 91 ++++++++++++++++++++++
.../test/resources/authentication/basic/.htpasswd | 2 +
6 files changed, 147 insertions(+), 15 deletions(-)
diff --git a/conf/broker.conf b/conf/broker.conf
index ea27c8022b3..f485e854bc1 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -689,6 +689,13 @@ athenzDomainNames=
# When this parameter is not empty, unauthenticated users perform as anonymousUserRole
anonymousUserRole=
+## Configure the datasource of basic authenticate, supports the file and Base64 format.
+# file:
+# basicAuthConf=/path/my/.htpasswd
+# use Base64 to encode the contents of .htpasswd:
+# basicAuthConf=YOUR-BASE64-DATA
+basicAuthConf=
+
### --- Token Authentication Provider --- ###
## Symmetric key
diff --git a/conf/proxy.conf b/conf/proxy.conf
index 3336a6166a5..bc55d1b0327 100644
--- a/conf/proxy.conf
+++ b/conf/proxy.conf
@@ -238,6 +238,13 @@ httpRequestsLimitEnabled=false
httpRequestsMaxPerSecond=100.0
+## Configure the datasource of basic authenticate, supports the file and Base64 format.
+# file:
+# basicAuthConf=/path/my/.htpasswd
+# use Base64 to encode the contents of .htpasswd:
+# basicAuthConf=YOUR-BASE64-DATA
+basicAuthConf=
+
### --- Token Authentication Provider --- ###
## Symmetric key
diff --git a/conf/standalone.conf b/conf/standalone.conf
index 8562f1aa8e0..1abf5aa7de9 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -462,6 +462,12 @@ athenzDomainNames=
# When this parameter is not empty, unauthenticated users perform as anonymousUserRole
anonymousUserRole=
+## Configure the datasource of basic authenticate, supports the file and Base64 format.
+# file:
+# basicAuthConf=/path/my/.htpasswd
+# use Base64 to encode the contents of .htpasswd:
+# basicAuthConf=YOUR-BASE64-DATA
+basicAuthConf=
### --- Token Authentication Provider --- ###
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderBasic.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderBasic.java
index b564f8a90f7..40ff8635e03 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderBasic.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderBasic.java
@@ -19,24 +19,29 @@
package org.apache.pulsar.broker.authentication;
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.StringReader;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import javax.naming.AuthenticationException;
+import lombok.Cleanup;
import org.apache.commons.codec.digest.Crypt;
import org.apache.commons.codec.digest.Md5Crypt;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.broker.ServiceConfiguration;
-
-import lombok.Cleanup;
import org.apache.pulsar.broker.authentication.metrics.AuthenticationMetrics;
-import javax.naming.AuthenticationException;
-import java.io.BufferedReader;
-import java.io.File;
-import java.io.FileReader;
-import java.io.IOException;
-import java.util.*;
-
public class AuthenticationProviderBasic implements AuthenticationProvider {
private static final String HTTP_HEADER_NAME = "Authorization";
private static final String CONF_SYSTEM_PROPERTY_KEY = "pulsar.auth.basic.conf";
+ private static final String CONF_PULSAR_PROPERTY_KEY = "basicAuthConf";
private Map<String, String> users;
@Override
@@ -46,14 +51,28 @@ public class AuthenticationProviderBasic implements AuthenticationProvider {
@Override
public void initialize(ServiceConfiguration config) throws IOException {
- File confFile = new File(System.getProperty(CONF_SYSTEM_PROPERTY_KEY));
- if (!confFile.exists()) {
- throw new IOException("The password auth conf file does not exist");
- } else if (!confFile.isFile()) {
- throw new IOException("The path is not a file");
+ String data = config.getProperties().getProperty(CONF_PULSAR_PROPERTY_KEY);
+ if (StringUtils.isEmpty(data)) {
+ data = System.getProperty(CONF_SYSTEM_PROPERTY_KEY);
+ }
+ if (StringUtils.isEmpty(data)) {
+ throw new IOException("No basic authentication config provided");
+ }
+
+ @Cleanup BufferedReader reader = null;
+ if (org.apache.commons.codec.binary.Base64.isBase64(data)) {
+ reader = new BufferedReader(new StringReader(new String(Base64.getDecoder().decode(data),
+ StandardCharsets.UTF_8)));
+ } else {
+ File confFile = new File(data);
+ if (!confFile.exists()) {
+ throw new IOException("The password auth conf file does not exist");
+ } else if (!confFile.isFile()) {
+ throw new IOException("The path is not a file");
+ }
+ reader = new BufferedReader(new FileReader(confFile));
}
- @Cleanup BufferedReader reader = new BufferedReader(new FileReader(confFile));
users = new HashMap<>();
for (String line : reader.lines().toArray(s -> new String[s])) {
List<String> splitLine = Arrays.asList(line.split(":"));
diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authentication/AuthenticationProviderBasicTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authentication/AuthenticationProviderBasicTest.java
new file mode 100644
index 00000000000..812e99a0abf
--- /dev/null
+++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authentication/AuthenticationProviderBasicTest.java
@@ -0,0 +1,91 @@
+/**
+ * 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.broker.authentication;
+
+import com.google.common.io.Resources;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.util.Base64;
+import java.util.Properties;
+import lombok.Cleanup;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.common.api.AuthData;
+import org.testng.annotations.Test;
+
+import javax.naming.AuthenticationException;
+
+public class AuthenticationProviderBasicTest {
+ private final String basicAuthConf = Resources.getResource("authentication/basic/.htpasswd").getPath();
+ private final String basicAuthConfBase64 = Base64.getEncoder().encodeToString(Files.readAllBytes(
+ FileSystems.getDefault().getPath(basicAuthConf)));
+
+ public AuthenticationProviderBasicTest() throws IOException {
+ }
+
+ private void testAuthenticate(AuthenticationProviderBasic provider) throws AuthenticationException {
+ AuthData authData = AuthData.of("superUser2:superpassword".getBytes(StandardCharsets.UTF_8));
+ provider.newAuthState(authData, null, null);
+ }
+
+ @Test
+ public void testLoadFileFromPulsarProperties() throws Exception {
+ @Cleanup
+ AuthenticationProviderBasic provider = new AuthenticationProviderBasic();
+ ServiceConfiguration serviceConfiguration = new ServiceConfiguration();
+ Properties properties = new Properties();
+ properties.setProperty("basicAuthConf", basicAuthConf);
+ serviceConfiguration.setProperties(properties);
+ provider.initialize(serviceConfiguration);
+ testAuthenticate(provider);
+ }
+
+ @Test
+ public void testLoadBase64FromPulsarProperties() throws Exception {
+ @Cleanup
+ AuthenticationProviderBasic provider = new AuthenticationProviderBasic();
+ ServiceConfiguration serviceConfiguration = new ServiceConfiguration();
+ Properties properties = new Properties();
+ properties.setProperty("basicAuthConf", basicAuthConfBase64);
+ serviceConfiguration.setProperties(properties);
+ provider.initialize(serviceConfiguration);
+ testAuthenticate(provider);
+ }
+
+ @Test
+ public void testLoadFileFromSystemProperties() throws Exception {
+ @Cleanup
+ AuthenticationProviderBasic provider = new AuthenticationProviderBasic();
+ ServiceConfiguration serviceConfiguration = new ServiceConfiguration();
+ System.setProperty("pulsar.auth.basic.conf", basicAuthConf);
+ provider.initialize(serviceConfiguration);
+ testAuthenticate(provider);
+ }
+
+ @Test
+ public void testLoadBase64FromSystemProperties() throws Exception {
+ @Cleanup
+ AuthenticationProviderBasic provider = new AuthenticationProviderBasic();
+ ServiceConfiguration serviceConfiguration = new ServiceConfiguration();
+ System.setProperty("pulsar.auth.basic.conf", basicAuthConfBase64);
+ provider.initialize(serviceConfiguration);
+ testAuthenticate(provider);
+ }
+}
diff --git a/pulsar-broker-common/src/test/resources/authentication/basic/.htpasswd b/pulsar-broker-common/src/test/resources/authentication/basic/.htpasswd
new file mode 100644
index 00000000000..b1a099a5f0e
--- /dev/null
+++ b/pulsar-broker-common/src/test/resources/authentication/basic/.htpasswd
@@ -0,0 +1,2 @@
+superUser:mQQQIsyvvKRtU
+superUser2:$apr1$foobarmq$kuSZlLgOITksCkRgl57ie/