From a637a40b39826fa50642eb71130f7f55bc0b5671 Mon Sep 17 00:00:00 2001 From: liudezhi Date: Mon, 7 Feb 2022 20:33:33 +0800 Subject: [PATCH 1/9] make revokePermissionsOnTopic method async. --- .../admin/impl/PersistentTopicsBase.java | 64 +++++++++++-------- .../broker/admin/v1/PersistentTopics.java | 17 +++-- .../broker/admin/v2/PersistentTopics.java | 11 +++- .../apache/pulsar/broker/admin/AdminTest.java | 8 ++- .../broker/admin/PersistentTopicsTest.java | 13 +++- 5 files changed, 75 insertions(+), 38 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index 69e9bd52c67bc..a08075ae35cd4 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -327,49 +327,61 @@ protected void internalDeleteTopicForcefully(boolean authoritative, boolean dele } } - private void revokePermissions(String topicUri, String role) { + private CompletableFuture revokePermissions(String topicUri, String role) { Policies policies; try { policies = namespaceResources().getPolicies(namespaceName) .orElseThrow(() -> new RestException(Status.NOT_FOUND, "Namespace does not exist")); } catch (Exception e) { log.error("[{}] Failed to revoke permissions for topic {}", clientAppId(), topicUri, e); - throw new RestException(e); + return FutureUtil.failedFuture(new RestException(e)); } if (!policies.auth_policies.getTopicAuthentication().containsKey(topicUri) || !policies.auth_policies.getTopicAuthentication().get(topicUri).containsKey(role)) { log.warn("[{}] Failed to revoke permission from role {} on topic: Not set at topic level {}", clientAppId(), role, topicUri); - throw new RestException(Status.PRECONDITION_FAILED, "Permissions are not set at the topic level"); - } - try { - // Write the new policies to metadata store - namespaceResources().setPolicies(namespaceName, p -> { - p.auth_policies.getTopicAuthentication().get(topicUri).remove(role); - return p; - }); - log.info("[{}] Successfully revoke access for role {} - topic {}", clientAppId(), role, topicUri); - } catch (Exception e) { - log.error("[{}] Failed to revoke permissions for topic {}", clientAppId(), topicUri, e); - throw new RestException(e); + return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED, + "Permissions are not set at the topic level")); } + // Write the new policies to metadata store + return namespaceResources().setPoliciesAsync(namespaceName, p -> { + p.auth_policies.getTopicAuthentication().get(topicUri).remove(role); + return p; + }).thenAccept(__ -> log.info("[{}] Successfully revoke access for role {} - topic {}", clientAppId(), role, + topicUri) + + ).exceptionally(ex -> { + Throwable realCause = FutureUtil.unwrapCompletionException(ex); + log.error("[{}] Failed revoke access for role {} - topic {}", clientAppId(), role, topicUri, realCause); + throw new RestException(realCause); + + }); } - protected void internalRevokePermissionsOnTopic(String role) { + protected void internalRevokePermissionsOnTopic(AsyncResponse asyncResponse, String role) { // This operation should be reading from zookeeper and it should be allowed without having admin privileges validateAdminAccessForTenant(namespaceName.getTenant()); - validatePoliciesReadOnlyAccess(); - - PartitionedTopicMetadata meta = getPartitionedTopicMetadata(topicName, true, false); - int numPartitions = meta.partitions; - if (numPartitions > 0) { - for (int i = 0; i < numPartitions; i++) { - TopicName topicNamePartition = topicName.getPartition(i); - revokePermissions(topicNamePartition.toString(), role); - } - } - revokePermissions(topicName.toString(), role); + validatePoliciesReadOnlyAccessAsync().thenCompose(__ -> + getPartitionedTopicMetadataAsync(topicName, true, false) + .thenCompose(metadata -> { + int numPartitions = metadata.partitions; + CompletableFuture future = CompletableFuture.completedFuture(null); + if (numPartitions > 0) { + for (int i = 0; i < numPartitions; i++) { + TopicName topicNamePartition = topicName.getPartition(i); + future = future.thenComposeAsync(unused -> revokePermissions(topicNamePartition.toString(), + role)); + } + } + return future.thenComposeAsync(unused -> revokePermissions(topicName.toString(), role)) + .thenAccept(unused -> asyncResponse.resume(Response.noContent().build())); + })).exceptionally(ex -> { + Throwable realCause = FutureUtil.unwrapCompletionException(ex); + log.error("[{}] Failed to revoke permissions for topic {}", clientAppId(), topicName, realCause); + resumeAsyncResponseExceptionally(asyncResponse, realCause); + return null; + }); } protected void internalCreateNonPartitionedTopic(boolean authoritative, Map properties) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java index fc72ae4c40287..5ccb718022eb4 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java @@ -139,11 +139,18 @@ public void grantPermissionsOnTopic(@PathParam("property") String property, @ApiResponse(code = 403, message = "Don't have admin permission"), @ApiResponse(code = 404, message = "Namespace doesn't exist"), @ApiResponse(code = 412, message = "Permissions are not set at the topic level")}) - public void revokePermissionsOnTopic(@PathParam("property") String property, - @PathParam("cluster") String cluster, @PathParam("namespace") String namespace, - @PathParam("topic") @Encoded String encodedTopic, @PathParam("role") String role) { - validateTopicName(property, cluster, namespace, encodedTopic); - internalRevokePermissionsOnTopic(role); + public void revokePermissionsOnTopic(@Suspended final AsyncResponse asyncResponse, + @PathParam("property") String property, @PathParam("cluster") String cluster, + @PathParam("namespace") String namespace, @PathParam("topic") @Encoded String encodedTopic, + @PathParam("role") String role) { + try { + validateTopicName(property, cluster, namespace, encodedTopic); + internalRevokePermissionsOnTopic(asyncResponse, role); + } catch (WebApplicationException wae) { + asyncResponse.resume(wae); + } catch (Exception e) { + asyncResponse.resume(new RestException(e)); + } } @PUT diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java index 9449a6f8f7e68..6c58ca5e2ba4f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java @@ -190,6 +190,7 @@ public void grantPermissionsOnTopic( @ApiResponse(code = 412, message = "Permissions are not set at the topic level"), @ApiResponse(code = 500, message = "Internal server error")}) public void revokePermissionsOnTopic( + @Suspended final AsyncResponse asyncResponse, @ApiParam(value = "Specify the tenant", required = true) @PathParam("tenant") String tenant, @ApiParam(value = "Specify the namespace", required = true) @@ -198,8 +199,14 @@ public void revokePermissionsOnTopic( @PathParam("topic") @Encoded String encodedTopic, @ApiParam(value = "Client role to which grant permissions", required = true) @PathParam("role") String role) { - validateTopicName(tenant, namespace, encodedTopic); - internalRevokePermissionsOnTopic(role); + try { + validateTopicName(tenant, namespace, encodedTopic); + internalRevokePermissionsOnTopic(asyncResponse, role); + } catch (WebApplicationException wae) { + asyncResponse.resume(wae); + } catch (Exception e) { + asyncResponse.resume(new RestException(e)); + } } @PUT diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java index 6212c342cb1d4..8a45d3cf8f18a 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java @@ -781,14 +781,16 @@ public void persistentTopics() throws Exception { namespace, topic); assertEquals(permission.get(role), actions); // remove permission - persistentTopics.revokePermissionsOnTopic(property, cluster, namespace, topic, role); - + response = mock(AsyncResponse.class); + persistentTopics.revokePermissionsOnTopic(response, property, cluster, namespace, topic, role); + responseCaptor = ArgumentCaptor.forClass(Response.class); + verify(response, timeout(5000).times(1)).resume(responseCaptor.capture()); + Assert.assertEquals(responseCaptor.getValue().getStatus(), Response.Status.NO_CONTENT.getStatusCode()); // verify removed permission Awaitility.await().untilAsserted(() -> { Map> p = persistentTopics.getPermissionsOnTopic(property, cluster, namespace, topic); assertTrue(p.isEmpty()); }); - } @Test diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java index cda9653a0bb76..47f7c09df2f4e 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java @@ -640,7 +640,11 @@ public void testRevokeNonPartitionedTopic() { Set expectActions = new HashSet<>(); expectActions.add(AuthAction.produce); persistentTopics.grantPermissionsOnTopic(testTenant, testNamespace, topicName, role, expectActions); - persistentTopics.revokePermissionsOnTopic(testTenant, testNamespace, topicName, role); + AsyncResponse response = mock(AsyncResponse.class); + ArgumentCaptor responseCaptor = ArgumentCaptor.forClass(Response.class); + persistentTopics.revokePermissionsOnTopic(response, testTenant, testNamespace, topicName, role); + verify(response, timeout(5000).times(1)).resume(responseCaptor.capture()); + Assert.assertEquals(responseCaptor.getValue().getStatus(), Response.Status.NO_CONTENT.getStatusCode()); Map> permissions = persistentTopics.getPermissionsOnTopic(testTenant, testNamespace, topicName); Assert.assertEquals(permissions.get(role), null); } @@ -659,7 +663,12 @@ public void testRevokePartitionedTopic() { Set expectActions = new HashSet<>(); expectActions.add(AuthAction.produce); persistentTopics.grantPermissionsOnTopic(testTenant, testNamespace, partitionedTopicName, role, expectActions); - persistentTopics.revokePermissionsOnTopic(testTenant, testNamespace, partitionedTopicName, role); + response = mock(AsyncResponse.class); + persistentTopics.revokePermissionsOnTopic(response, testTenant, testNamespace, partitionedTopicName, role); + responseCaptor = ArgumentCaptor.forClass(Response.class); + verify(response, timeout(5000).times(1)).resume(responseCaptor.capture()); + Assert.assertEquals(responseCaptor.getValue().getStatus(), Response.Status.NO_CONTENT.getStatusCode()); + Map> permissions = persistentTopics.getPermissionsOnTopic(testTenant, testNamespace, partitionedTopicName); Assert.assertEquals(permissions.get(role), null); From d600a5243761c28252fae78feab2627af6c6858b Mon Sep 17 00:00:00 2001 From: liudezhi Date: Sun, 20 Mar 2022 18:50:15 +0800 Subject: [PATCH 2/9] chang validateAdminAccessForTenant for Async --- .../admin/impl/PersistentTopicsBase.java | 78 ++++---- .../pulsar/broker/web/PulsarWebResource.java | 169 ++++++++++++------ 2 files changed, 156 insertions(+), 91 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index d0779b996a886..26e47e9534b29 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.broker.admin.impl; +import static java.util.concurrent.TimeUnit.SECONDS; import static org.apache.pulsar.broker.PulsarService.isTransactionInternalName; import static org.apache.pulsar.broker.resources.PulsarResources.DEFAULT_OPERATION_TIMEOUT_SEC; import static org.apache.pulsar.common.events.EventsTopicNames.checkTopicIsTransactionCoordinatorAssign; @@ -321,41 +322,40 @@ protected void internalDeleteTopicForcefully(boolean authoritative, boolean dele } private CompletableFuture revokePermissions(String topicUri, String role) { - Policies policies; - try { - policies = namespaceResources().getPolicies(namespaceName) - .orElseThrow(() -> new RestException(Status.NOT_FOUND, "Namespace does not exist")); - } catch (Exception e) { - log.error("[{}] Failed to revoke permissions for topic {}", clientAppId(), topicUri, e); - return FutureUtil.failedFuture(new RestException(e)); - } - if (!policies.auth_policies.getTopicAuthentication().containsKey(topicUri) - || !policies.auth_policies.getTopicAuthentication().get(topicUri).containsKey(role)) { - log.warn("[{}] Failed to revoke permission from role {} on topic: Not set at topic level {}", clientAppId(), - role, topicUri); - return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED, - "Permissions are not set at the topic level")); - } - - // Write the new policies to metadata store - return namespaceResources().setPoliciesAsync(namespaceName, p -> { - p.auth_policies.getTopicAuthentication().get(topicUri).remove(role); - return p; - }).thenAccept(__ -> log.info("[{}] Successfully revoke access for role {} - topic {}", clientAppId(), role, - topicUri) - - ).exceptionally(ex -> { - Throwable realCause = FutureUtil.unwrapCompletionException(ex); - log.error("[{}] Failed revoke access for role {} - topic {}", clientAppId(), role, topicUri, realCause); - throw new RestException(realCause); - - }); + return namespaceResources().getPoliciesAsync(namespaceName).thenCompose( + policiesOptional -> { + Policies policies = policiesOptional.orElseThrow(() -> + new RestException(Status.NOT_FOUND, "Namespace does not exist")); + if (!policies.auth_policies.getTopicAuthentication().containsKey(topicUri) + || !policies.auth_policies.getTopicAuthentication().get(topicUri).containsKey(role)) { + log.warn("[{}] Failed to revoke permission from role {} on topic: Not set at topic level {}", + clientAppId(), role, topicUri); + return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED, + "Permissions are not set at the topic level")); + } + if (!policies.auth_policies.getTopicAuthentication().containsKey(topicUri) + || !policies.auth_policies.getTopicAuthentication().get(topicUri).containsKey(role)) { + log.warn("[{}] Failed to revoke permission from role {} on topic: Not set at topic level {}", + clientAppId(), role, topicUri); + return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED, + "Permissions are not set at the topic level")); + } + // Write the new policies to metadata store + return namespaceResources().setPoliciesAsync(namespaceName, p -> { + p.auth_policies.getTopicAuthentication().get(topicUri).remove(role); + return p; + }).thenAccept(__ -> + log.info("[{}] Successfully revoke access for role {} - topic {}", clientAppId(), role, + topicUri) + ); + } + ); } protected void internalRevokePermissionsOnTopic(AsyncResponse asyncResponse, String role) { // This operation should be reading from zookeeper and it should be allowed without having admin privileges - validateAdminAccessForTenant(namespaceName.getTenant()); - validatePoliciesReadOnlyAccessAsync().thenCompose(__ -> + validateAdminAccessForTenantAsync(namespaceName.getTenant()) + .thenAccept(__ -> validatePoliciesReadOnlyAccessAsync().thenCompose(unused1 -> getPartitionedTopicMetadataAsync(topicName, true, false) .thenCompose(metadata -> { int numPartitions = metadata.partitions; @@ -369,12 +369,13 @@ protected void internalRevokePermissionsOnTopic(AsyncResponse asyncResponse, Str } return future.thenComposeAsync(unused -> revokePermissions(topicName.toString(), role)) .thenAccept(unused -> asyncResponse.resume(Response.noContent().build())); - })).exceptionally(ex -> { - Throwable realCause = FutureUtil.unwrapCompletionException(ex); - log.error("[{}] Failed to revoke permissions for topic {}", clientAppId(), topicName, realCause); - resumeAsyncResponseExceptionally(asyncResponse, realCause); - return null; - }); + })) + ).exceptionally(ex -> { + Throwable realCause = FutureUtil.unwrapCompletionException(ex); + log.error("[{}] Failed to revoke permissions for topic {}", clientAppId(), topicName, realCause); + resumeAsyncResponseExceptionally(asyncResponse, realCause); + return null; + }); } protected void internalCreateNonPartitionedTopic(boolean authoritative, Map properties) { @@ -3878,7 +3879,8 @@ public static CompletableFuture getPartitionedTopicMet } catch (RestException e) { try { validateAdminAccessForTenant(pulsar, - clientAppId, originalPrincipal, topicName.getTenant(), authenticationData); + clientAppId, originalPrincipal, topicName.getTenant(), authenticationData, + pulsar.getConfiguration().getMetadataStoreOperationTimeoutSeconds(), SECONDS); } catch (RestException authException) { log.warn("Failed to authorize {} on topic {}", clientAppId, topicName); throw new PulsarClientException(String.format("Authorization failed %s on topic %s with error %s", diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java index 2d6f44ce1c815..5d950cf3d2b84 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java @@ -35,7 +35,9 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.function.BiFunction; import javax.servlet.ServletContext; import javax.servlet.http.HttpServletRequest; import javax.ws.rs.WebApplicationException; @@ -257,7 +259,8 @@ public void validateSuperUserAccess() { */ protected void validateAdminAccessForTenant(String tenant) { try { - validateAdminAccessForTenant(pulsar(), clientAppId(), originalPrincipal(), tenant, clientAuthData()); + validateAdminAccessForTenant(pulsar(), clientAppId(), originalPrincipal(), tenant, clientAuthData(), + config().getMetadataStoreOperationTimeoutSeconds(), SECONDS); } catch (RestException e) { throw e; } catch (Exception e) { @@ -267,65 +270,125 @@ protected void validateAdminAccessForTenant(String tenant) { } protected static void validateAdminAccessForTenant(PulsarService pulsar, String clientAppId, - String originalPrincipal, String tenant, - AuthenticationDataSource authenticationData) - throws Exception { + String originalPrincipal, String tenant, + AuthenticationDataSource authenticationData, + long timeout, TimeUnit unit) { + try { + validateAdminAccessForTenantAsync(pulsar, clientAppId, originalPrincipal, tenant, authenticationData) + .get(timeout, unit); + } catch (InterruptedException | ExecutionException | TimeoutException e) { + Throwable realCause = FutureUtil.unwrapCompletionException(e); + if (realCause instanceof WebApplicationException) { + throw (WebApplicationException) realCause; + } else { + throw new RestException(realCause); + } + } + } + + /** + * Checks that the http client role has admin access to the specified tenant async. + * + * @param tenant the tenant id + */ + protected CompletableFuture validateAdminAccessForTenantAsync(String tenant) { + return validateAdminAccessForTenantAsync(pulsar(), clientAppId(), originalPrincipal(), tenant, + clientAuthData()); + } + + protected static CompletableFuture validateAdminAccessForTenantAsync( + PulsarService pulsar, String clientAppId, + String originalPrincipal, String tenant, + AuthenticationDataSource authenticationData) { + CompletableFuture future = new CompletableFuture<>(); if (log.isDebugEnabled()) { log.debug("check admin access on tenant: {} - Authenticated: {} -- role: {}", tenant, (isClientAuthenticated(clientAppId)), clientAppId); } - TenantInfo tenantInfo = pulsar.getPulsarResources().getTenantResources().getTenant(tenant) - .orElseThrow(() -> new RestException(Status.NOT_FOUND, "Tenant does not exist")); - - if (pulsar.getConfiguration().isAuthenticationEnabled() && pulsar.getConfiguration().isAuthorizationEnabled()) { - if (!isClientAuthenticated(clientAppId)) { - throw new RestException(Status.FORBIDDEN, "Need to authenticate to perform the request"); - } - - validateOriginalPrincipal(pulsar.getConfiguration().getProxyRoles(), clientAppId, originalPrincipal); - - if (pulsar.getConfiguration().getProxyRoles().contains(clientAppId)) { - CompletableFuture isProxySuperUserFuture; - CompletableFuture isOriginalPrincipalSuperUserFuture; - try { - AuthorizationService authorizationService = pulsar.getBrokerService().getAuthorizationService(); - isProxySuperUserFuture = authorizationService.isSuperUser(clientAppId, authenticationData); - - isOriginalPrincipalSuperUserFuture = - authorizationService.isSuperUser(originalPrincipal, authenticationData); - - boolean proxyAuthorized = isProxySuperUserFuture.get() - || authorizationService.isTenantAdmin(tenant, clientAppId, - tenantInfo, authenticationData).get(); - boolean originalPrincipalAuthorized = - isOriginalPrincipalSuperUserFuture.get() || authorizationService.isTenantAdmin(tenant, - originalPrincipal, tenantInfo, authenticationData).get(); - if (!proxyAuthorized || !originalPrincipalAuthorized) { - throw new RestException(Status.UNAUTHORIZED, - String.format("Proxy not authorized to access resource (proxy:%s,original:%s)", - clientAppId, originalPrincipal)); + pulsar.getPulsarResources().getTenantResources().getTenantAsync(tenant) + .thenCompose(tenantInfoOptional -> { + if (!tenantInfoOptional.isPresent()) { + throw new RestException(Status.NOT_FOUND, "Tenant does not exist"); } - } catch (InterruptedException | ExecutionException e) { - throw new RestException(Status.INTERNAL_SERVER_ERROR, e.getMessage()); - } - log.debug("Successfully authorized {} (proxied by {}) on tenant {}", - originalPrincipal, clientAppId, tenant); - } else { - if (!pulsar.getBrokerService() - .getAuthorizationService() - .isSuperUser(clientAppId, authenticationData) - .join()) { - if (!pulsar.getBrokerService().getAuthorizationService() - .isTenantAdmin(tenant, clientAppId, tenantInfo, authenticationData).get()) { - throw new RestException(Status.UNAUTHORIZED, - "Don't have permission to administrate resources on this tenant"); + TenantInfo tenantInfo = tenantInfoOptional.get(); + if (pulsar.getConfiguration().isAuthenticationEnabled() && pulsar.getConfiguration() + .isAuthorizationEnabled()) { + if (!isClientAuthenticated(clientAppId)) { + throw new RestException(Status.FORBIDDEN, "Need to authenticate to perform the request"); + } + validateOriginalPrincipal(pulsar.getConfiguration().getProxyRoles(), clientAppId, + originalPrincipal); + if (pulsar.getConfiguration().getProxyRoles().contains(clientAppId)) { + AuthorizationService authorizationService = + pulsar.getBrokerService().getAuthorizationService(); + return authorizationService.isTenantAdmin(tenant, clientAppId, tenantInfo, + authenticationData) + .thenCompose(isTenantAdmin -> { + String debugMsg = "Successfully authorized {} (proxied by {}) on tenant {}"; + if (!isTenantAdmin) { + return authorizationService.isSuperUser(clientAppId, authenticationData) + .thenCombine(authorizationService.isSuperUser(originalPrincipal, + authenticationData), + new BiFunction() { + @Override + public Boolean apply(Boolean proxyAuthorized, + Boolean originalPrincipalAuthorized) { + if (!proxyAuthorized || !originalPrincipalAuthorized) { + throw new RestException(Status.UNAUTHORIZED, + String.format("Proxy not authorized to access " + + "resource (proxy:%s,original:%s)", + clientAppId, originalPrincipal)); + } else { + if (log.isDebugEnabled()) { + log.debug(debugMsg, originalPrincipal, clientAppId, + tenant); + } + future.complete(null); + return true; + } + } + }); + } else { + if (log.isDebugEnabled()) { + log.debug(debugMsg, originalPrincipal, clientAppId, tenant); + } + future.complete(null); + return CompletableFuture.completedFuture(null); + } + }); + } else { + return pulsar.getBrokerService() + .getAuthorizationService() + .isSuperUser(clientAppId, authenticationData) + .thenCompose(isSuperUser -> { + if (!isSuperUser) { + return pulsar.getBrokerService().getAuthorizationService() + .isTenantAdmin(tenant, clientAppId, tenantInfo, authenticationData); + } else { + return CompletableFuture.completedFuture(true); + } + }).thenCompose(authorized -> { + if (!authorized) { + throw new RestException(Status.UNAUTHORIZED, + "Don't have permission to administrate resources on this tenant"); + } else { + log.debug("Successfully authorized {} on tenant {}", clientAppId, tenant); + future.complete(null); + return CompletableFuture.completedFuture(null); + } + }); + } + } else { + future.complete(null); + return CompletableFuture.completedFuture(null); } - } - - log.debug("Successfully authorized {} on tenant {}", clientAppId, tenant); - } - } + }) + .exceptionally(ex -> { + future.completeExceptionally(new RestException(Status.INTERNAL_SERVER_ERROR, ex.getMessage())); + return null; + }); + return future; } /** From 0e90b6179b6fff7f7bed21b990060ed732649ce1 Mon Sep 17 00:00:00 2001 From: liudezhi Date: Tue, 12 Apr 2022 18:03:46 +0800 Subject: [PATCH 3/9] hand conflict --- .../java/org/apache/pulsar/broker/web/PulsarWebResource.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java index 5d950cf3d2b84..ededdbbb9ee25 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java @@ -37,7 +37,7 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; -import java.util.function.BiFunction; +import java.util.function.Supplier; import javax.servlet.ServletContext; import javax.servlet.http.HttpServletRequest; import javax.ws.rs.WebApplicationException; From a713838fef3c075c1c1ca316fbfc4b188e150986 Mon Sep 17 00:00:00 2001 From: liudezhi Date: Tue, 12 Apr 2022 18:06:52 +0800 Subject: [PATCH 4/9] hand conflict --- .../java/org/apache/pulsar/broker/web/PulsarWebResource.java | 1 + 1 file changed, 1 insertion(+) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java index b0526ec7c4d57..3185043f9ecdf 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java @@ -36,6 +36,7 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.function.BiFunction; import java.util.function.Supplier; import javax.servlet.ServletContext; import javax.servlet.http.HttpServletRequest; From b511627d8647ea1f7f8d9f10bea9740451d6f93a Mon Sep 17 00:00:00 2001 From: liudezhi Date: Tue, 12 Apr 2022 20:59:32 +0800 Subject: [PATCH 5/9] accept suggestion --- .../apache/pulsar/broker/admin/impl/PersistentTopicsBase.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index 26e47e9534b29..3ce162c38b9a5 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -355,7 +355,7 @@ private CompletableFuture revokePermissions(String topicUri, String role) protected void internalRevokePermissionsOnTopic(AsyncResponse asyncResponse, String role) { // This operation should be reading from zookeeper and it should be allowed without having admin privileges validateAdminAccessForTenantAsync(namespaceName.getTenant()) - .thenAccept(__ -> validatePoliciesReadOnlyAccessAsync().thenCompose(unused1 -> + .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync().thenCompose(unused1 -> getPartitionedTopicMetadataAsync(topicName, true, false) .thenCompose(metadata -> { int numPartitions = metadata.partitions; From a7758662cdffd02981832017aac775bbc035a847 Mon Sep 17 00:00:00 2001 From: liudezhi Date: Fri, 15 Apr 2022 19:07:22 +0800 Subject: [PATCH 6/9] remove duplicate code --- .../pulsar/broker/admin/impl/PersistentTopicsBase.java | 7 ------- 1 file changed, 7 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index 3ce162c38b9a5..8949045674cd2 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -333,13 +333,6 @@ private CompletableFuture revokePermissions(String topicUri, String role) return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED, "Permissions are not set at the topic level")); } - if (!policies.auth_policies.getTopicAuthentication().containsKey(topicUri) - || !policies.auth_policies.getTopicAuthentication().get(topicUri).containsKey(role)) { - log.warn("[{}] Failed to revoke permission from role {} on topic: Not set at topic level {}", - clientAppId(), role, topicUri); - return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED, - "Permissions are not set at the topic level")); - } // Write the new policies to metadata store return namespaceResources().setPoliciesAsync(namespaceName, p -> { p.auth_policies.getTopicAuthentication().get(topicUri).remove(role); From 42721df4d257a4709cf4e47af344ac102ed348d4 Mon Sep 17 00:00:00 2001 From: liudezhi Date: Mon, 18 Apr 2022 12:20:31 +0800 Subject: [PATCH 7/9] accept suggestion --- .../admin/impl/PersistentTopicsBase.java | 8 +-- .../pulsar/broker/web/PulsarWebResource.java | 59 ++++++++----------- 2 files changed, 28 insertions(+), 39 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index 8949045674cd2..eb8d9bcc513c0 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -321,7 +321,7 @@ protected void internalDeleteTopicForcefully(boolean authoritative, boolean dele } } - private CompletableFuture revokePermissions(String topicUri, String role) { + private CompletableFuture revokePermissionsAsync(String topicUri, String role) { return namespaceResources().getPoliciesAsync(namespaceName).thenCompose( policiesOptional -> { Policies policies = policiesOptional.orElseThrow(() -> @@ -356,11 +356,11 @@ protected void internalRevokePermissionsOnTopic(AsyncResponse asyncResponse, Str if (numPartitions > 0) { for (int i = 0; i < numPartitions; i++) { TopicName topicNamePartition = topicName.getPartition(i); - future = future.thenComposeAsync(unused -> revokePermissions(topicNamePartition.toString(), - role)); + future = future.thenComposeAsync(unused -> + revokePermissionsAsync(topicNamePartition.toString(), role)); } } - return future.thenComposeAsync(unused -> revokePermissions(topicName.toString(), role)) + return future.thenComposeAsync(unused -> revokePermissionsAsync(topicName.toString(), role)) .thenAccept(unused -> asyncResponse.resume(Response.noContent().build())); })) ).exceptionally(ex -> { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java index 3185043f9ecdf..fa8882383006b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java @@ -290,13 +290,11 @@ protected static CompletableFuture validateAdminAccessForTenantAsync( PulsarService pulsar, String clientAppId, String originalPrincipal, String tenant, AuthenticationDataSource authenticationData) { - CompletableFuture future = new CompletableFuture<>(); if (log.isDebugEnabled()) { log.debug("check admin access on tenant: {} - Authenticated: {} -- role: {}", tenant, (isClientAuthenticated(clientAppId)), clientAppId); } - - pulsar.getPulsarResources().getTenantResources().getTenantAsync(tenant) + return pulsar.getPulsarResources().getTenantResources().getTenantAsync(tenant) .thenCompose(tenantInfoOptional -> { if (!tenantInfoOptional.isPresent()) { throw new RestException(Status.NOT_FOUND, "Tenant does not exist"); @@ -317,36 +315,30 @@ protected static CompletableFuture validateAdminAccessForTenantAsync( .thenCompose(isTenantAdmin -> { String debugMsg = "Successfully authorized {} (proxied by {}) on tenant {}"; if (!isTenantAdmin) { - return authorizationService.isSuperUser(clientAppId, authenticationData) - .thenCombine(authorizationService.isSuperUser(originalPrincipal, - authenticationData), - new BiFunction() { - @Override - public Boolean apply(Boolean proxyAuthorized, - Boolean originalPrincipalAuthorized) { - if (!proxyAuthorized || !originalPrincipalAuthorized) { - throw new RestException(Status.UNAUTHORIZED, - String.format("Proxy not authorized to access " - + "resource (proxy:%s,original:%s)", - clientAppId, originalPrincipal)); - } else { - if (log.isDebugEnabled()) { - log.debug(debugMsg, originalPrincipal, clientAppId, - tenant); - } - future.complete(null); - return true; - } - } - }); - } else { + return authorizationService.isSuperUser(clientAppId, authenticationData) + .thenCombine(authorizationService.isSuperUser(originalPrincipal, + authenticationData), + (proxyAuthorized, originalPrincipalAuthorized) -> { + if (!proxyAuthorized || !originalPrincipalAuthorized) { + throw new RestException(Status.UNAUTHORIZED, + String.format("Proxy not authorized to access " + + "resource (proxy:%s,original:%s)" + , clientAppId, originalPrincipal)); + } else { + if (log.isDebugEnabled()) { + log.debug(debugMsg, originalPrincipal, clientAppId, + tenant); + } + return true; + } + }); + } else { if (log.isDebugEnabled()) { log.debug(debugMsg, originalPrincipal, clientAppId, tenant); } - future.complete(null); return CompletableFuture.completedFuture(null); } - }); + }); } else { return pulsar.getBrokerService() .getAuthorizationService() @@ -364,21 +356,18 @@ public Boolean apply(Boolean proxyAuthorized, "Don't have permission to administrate resources on this tenant"); } else { log.debug("Successfully authorized {} on tenant {}", clientAppId, tenant); - future.complete(null); return CompletableFuture.completedFuture(null); } }); } } else { - future.complete(null); return CompletableFuture.completedFuture(null); } - }) - .exceptionally(ex -> { - future.completeExceptionally(new RestException(Status.INTERNAL_SERVER_ERROR, ex.getMessage())); - return null; + }).exceptionally(ex -> { + throw new RestException(Status.INTERNAL_SERVER_ERROR, ex.getMessage()); + }).thenCompose(__ -> { + return CompletableFuture.completedFuture(null); }); - return future; } /** From 2014bc21c70913ae8eea4589105bf83c8ad38be4 Mon Sep 17 00:00:00 2001 From: liudezhi Date: Mon, 18 Apr 2022 14:06:08 +0800 Subject: [PATCH 8/9] fix Checkstyle violation --- .../java/org/apache/pulsar/broker/web/PulsarWebResource.java | 1 - 1 file changed, 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java index fa8882383006b..14bbf2154edef 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java @@ -36,7 +36,6 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; -import java.util.function.BiFunction; import java.util.function.Supplier; import javax.servlet.ServletContext; import javax.servlet.http.HttpServletRequest; From cb18853f70bbbcd2a107e7bfe5d9be44c0bcc0d5 Mon Sep 17 00:00:00 2001 From: liudezhi Date: Mon, 18 Apr 2022 18:27:12 +0800 Subject: [PATCH 9/9] accept suggestion --- .../org/apache/pulsar/broker/web/PulsarWebResource.java | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java index 14bbf2154edef..baa5d3fe332de 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java @@ -328,7 +328,7 @@ protected static CompletableFuture validateAdminAccessForTenantAsync( log.debug(debugMsg, originalPrincipal, clientAppId, tenant); } - return true; + return null; } }); } else { @@ -349,23 +349,18 @@ protected static CompletableFuture validateAdminAccessForTenantAsync( } else { return CompletableFuture.completedFuture(true); } - }).thenCompose(authorized -> { + }).thenAccept(authorized -> { if (!authorized) { throw new RestException(Status.UNAUTHORIZED, "Don't have permission to administrate resources on this tenant"); } else { log.debug("Successfully authorized {} on tenant {}", clientAppId, tenant); - return CompletableFuture.completedFuture(null); } }); } } else { return CompletableFuture.completedFuture(null); } - }).exceptionally(ex -> { - throw new RestException(Status.INTERNAL_SERVER_ERROR, ex.getMessage()); - }).thenCompose(__ -> { - return CompletableFuture.completedFuture(null); }); }