| /** |
| * 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.authorization; |
| |
| import static java.util.concurrent.TimeUnit.SECONDS; |
| import java.net.SocketAddress; |
| import java.util.Set; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.ExecutionException; |
| import org.apache.commons.lang3.StringUtils; |
| import org.apache.pulsar.broker.PulsarServerException; |
| import org.apache.pulsar.broker.ServiceConfiguration; |
| import org.apache.pulsar.broker.authentication.AuthenticationDataSource; |
| import org.apache.pulsar.broker.cache.ConfigurationCacheService; |
| import org.apache.pulsar.broker.resources.PulsarResources; |
| 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.TenantInfoImpl; |
| 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.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import javax.ws.rs.core.Response; |
| import java.util.concurrent.TimeoutException; |
| |
| /** |
| * Authorization service that manages pluggable authorization provider and authorize requests accordingly. |
| * |
| */ |
| public class AuthorizationService { |
| private static final Logger log = LoggerFactory.getLogger(AuthorizationService.class); |
| |
| private AuthorizationProvider provider; |
| private final ServiceConfiguration conf; |
| |
| public AuthorizationService(ServiceConfiguration conf, PulsarResources pulsarResources) |
| throws PulsarServerException { |
| this.conf = conf; |
| try { |
| final String providerClassname = conf.getAuthorizationProvider(); |
| if (StringUtils.isNotBlank(providerClassname)) { |
| provider = (AuthorizationProvider) Class.forName(providerClassname) |
| .getDeclaredConstructor().newInstance(); |
| provider.initialize(conf, pulsarResources); |
| log.info("{} has been loaded.", providerClassname); |
| } else { |
| throw new PulsarServerException("No authorization providers are present."); |
| } |
| } catch (PulsarServerException e) { |
| throw e; |
| } catch (Throwable e) { |
| throw new PulsarServerException("Failed to load an authorization provider.", e); |
| } |
| } |
| |
| public CompletableFuture<Boolean> isSuperUser(String user, AuthenticationDataSource authenticationData) { |
| if (provider != null) { |
| return provider.isSuperUser(user, authenticationData, conf); |
| } |
| return FutureUtil.failedFuture(new IllegalStateException("No authorization provider configured")); |
| } |
| |
| public CompletableFuture<Boolean> isTenantAdmin(String tenant, String role, TenantInfo tenantInfo, |
| AuthenticationDataSource authenticationData) { |
| if (provider != null) { |
| return provider.isTenantAdmin(tenant, role, tenantInfo, authenticationData); |
| } |
| return FutureUtil.failedFuture(new IllegalStateException("No authorization provider configured")); |
| } |
| |
| /** |
| * |
| * Grant authorization-action permission on a namespace to the given client |
| * |
| * @param namespace |
| * @param actions |
| * @param role |
| * @param authDataJson |
| * additional authdata in json for targeted authorization provider |
| * @return |
| * @throws IllegalArgumentException |
| * when namespace not found |
| * @throws IllegalStateException |
| * when failed to grant permission |
| */ |
| public CompletableFuture<Void> grantPermissionAsync(NamespaceName namespace, Set<AuthAction> actions, String role, |
| String authDataJson) { |
| |
| if (provider != null) { |
| return provider.grantPermissionAsync(namespace, actions, role, authDataJson); |
| } |
| return FutureUtil.failedFuture(new IllegalStateException("No authorization provider configured")); |
| } |
| |
| /** |
| * Grant permission to roles that can access subscription-admin api |
| * |
| * @param namespace |
| * @param subscriptionName |
| * @param roles |
| * @param authDataJson |
| * additional authdata in json for targeted authorization provider |
| * @return |
| */ |
| public CompletableFuture<Void> grantSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName, |
| Set<String> roles, String authDataJson) { |
| |
| if (provider != null) { |
| return provider.grantSubscriptionPermissionAsync(namespace, subscriptionName, roles, authDataJson); |
| } |
| return FutureUtil.failedFuture(new IllegalStateException("No authorization provider configured")); |
| } |
| |
| /** |
| * Revoke subscription admin-api access for a role |
| * |
| * @param namespace |
| * @param subscriptionName |
| * @param role |
| * @return |
| */ |
| public CompletableFuture<Void> revokeSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName, |
| String role, String authDataJson) { |
| if (provider != null) { |
| return provider.revokeSubscriptionPermissionAsync(namespace, subscriptionName, role, authDataJson); |
| } |
| return FutureUtil.failedFuture(new IllegalStateException("No authorization provider configured")); |
| } |
| |
| /** |
| * Grant authorization-action permission on a topic to the given client |
| * |
| * @param topicname |
| * @param role |
| * @param authDataJson |
| * additional authdata in json for targeted authorization provider |
| * @return IllegalArgumentException when namespace not found |
| * @throws IllegalStateException |
| * when failed to grant permission |
| */ |
| public CompletableFuture<Void> grantPermissionAsync(TopicName topicname, Set<AuthAction> actions, String role, |
| String authDataJson) { |
| |
| if (provider != null) { |
| return provider.grantPermissionAsync(topicname, actions, role, authDataJson); |
| } |
| return FutureUtil.failedFuture(new IllegalStateException("No authorization provider configured")); |
| |
| } |
| |
| /** |
| * Check if the specified role has permission to send messages to the specified fully qualified topic name. |
| * |
| * @param topicName |
| * the fully qualified topic name associated with the topic. |
| * @param role |
| * the app id used to send messages to the topic. |
| */ |
| public CompletableFuture<Boolean> canProduceAsync(TopicName topicName, String role, |
| AuthenticationDataSource authenticationData) { |
| |
| if (!this.conf.isAuthorizationEnabled()) { |
| return CompletableFuture.completedFuture(true); |
| } |
| if (provider != null) { |
| return provider.isSuperUser(role, authenticationData, conf).thenComposeAsync(isSuperUser -> { |
| if (isSuperUser) { |
| return CompletableFuture.completedFuture(true); |
| } else { |
| return provider.canProduceAsync(topicName, role, authenticationData); |
| } |
| }); |
| } |
| return FutureUtil.failedFuture(new IllegalStateException("No authorization provider configured")); |
| } |
| |
| /** |
| * Check if the specified role has permission to receive messages from the specified fully qualified topic name. |
| * |
| * @param topicName |
| * the fully qualified topic name associated with the topic. |
| * @param role |
| * the app id used to receive messages from the topic. |
| * @param subscription |
| * the subscription name defined by the client |
| */ |
| public CompletableFuture<Boolean> canConsumeAsync(TopicName topicName, String role, |
| AuthenticationDataSource authenticationData, String subscription) { |
| if (!this.conf.isAuthorizationEnabled()) { |
| return CompletableFuture.completedFuture(true); |
| } |
| if (provider != null) { |
| return provider.isSuperUser(role, authenticationData, conf).thenComposeAsync(isSuperUser -> { |
| if (isSuperUser) { |
| return CompletableFuture.completedFuture(true); |
| } else { |
| return provider.canConsumeAsync(topicName, role, authenticationData, subscription); |
| } |
| }); |
| } |
| return FutureUtil.failedFuture(new IllegalStateException("No authorization provider configured")); |
| } |
| |
| public boolean canProduce(TopicName topicName, String role, AuthenticationDataSource authenticationData) |
| throws Exception { |
| try { |
| return canProduceAsync(topicName, role, authenticationData).get(conf.getZooKeeperOperationTimeoutSeconds(), |
| SECONDS); |
| } catch (InterruptedException e) { |
| log.warn("Time-out {} sec while checking authorization on {} ", conf.getZooKeeperOperationTimeoutSeconds(), |
| topicName); |
| throw e; |
| } catch (Exception e) { |
| log.warn("Producer-client with Role - {} failed to get permissions for topic - {}. {}", role, topicName, |
| e.getMessage()); |
| throw e; |
| } |
| } |
| |
| public boolean canConsume(TopicName topicName, String role, AuthenticationDataSource authenticationData, |
| String subscription) throws Exception { |
| try { |
| return canConsumeAsync(topicName, role, authenticationData, subscription) |
| .get(conf.getZooKeeperOperationTimeoutSeconds(), SECONDS); |
| } catch (InterruptedException e) { |
| log.warn("Time-out {} sec while checking authorization on {} ", conf.getZooKeeperOperationTimeoutSeconds(), |
| topicName); |
| throw e; |
| } catch (Exception e) { |
| log.warn("Consumer-client with Role - {} failed to get permissions for topic - {}. {}", role, topicName, |
| e.getMessage()); |
| throw e; |
| } |
| } |
| |
| /** |
| * Check whether the specified role can perform a lookup for the specified topic. |
| * |
| * For that the caller needs to have producer or consumer permission. |
| * |
| * @param topicName |
| * @param role |
| * @return |
| * @throws Exception |
| */ |
| public boolean canLookup(TopicName topicName, String role, AuthenticationDataSource authenticationData) |
| throws Exception { |
| try { |
| return canLookupAsync(topicName, role, authenticationData) |
| .get(conf.getZooKeeperOperationTimeoutSeconds(), SECONDS); |
| } catch (InterruptedException e) { |
| log.warn("Time-out {} sec while checking authorization on {} ", conf.getZooKeeperOperationTimeoutSeconds(), |
| topicName); |
| throw e; |
| } catch (Exception e) { |
| log.warn("Role - {} failed to get lookup permissions for topic - {}. {}", role, topicName, |
| e.getMessage()); |
| throw e; |
| } |
| } |
| |
| /** |
| * Check whether the specified role can perform a lookup for the specified topic. |
| * |
| * For that the caller needs to have producer or consumer permission. |
| * |
| * @param topicName |
| * @param role |
| * @return |
| * @throws Exception |
| */ |
| public CompletableFuture<Boolean> canLookupAsync(TopicName topicName, String role, |
| AuthenticationDataSource authenticationData) { |
| if (!this.conf.isAuthorizationEnabled()) { |
| return CompletableFuture.completedFuture(true); |
| } |
| if (provider != null) { |
| return provider.isSuperUser(role, authenticationData, conf).thenComposeAsync(isSuperUser -> { |
| if (isSuperUser) { |
| return CompletableFuture.completedFuture(true); |
| } else { |
| return provider.canLookupAsync(topicName, role, authenticationData); |
| } |
| }); |
| } |
| return FutureUtil.failedFuture(new IllegalStateException("No authorization provider configured")); |
| } |
| |
| public CompletableFuture<Boolean> allowFunctionOpsAsync(NamespaceName namespaceName, String role, |
| AuthenticationDataSource authenticationData) { |
| return provider.allowFunctionOpsAsync(namespaceName, role, authenticationData); |
| } |
| |
| public CompletableFuture<Boolean> allowSourceOpsAsync(NamespaceName namespaceName, String role, |
| AuthenticationDataSource authenticationData) { |
| return provider.allowSourceOpsAsync(namespaceName, role, authenticationData); |
| } |
| |
| public CompletableFuture<Boolean> allowSinkOpsAsync(NamespaceName namespaceName, String role, |
| AuthenticationDataSource authenticationData) { |
| return provider.allowSinkOpsAsync(namespaceName, role, authenticationData); |
| } |
| |
| /** |
| * Whether the authenticatedPrincipal and the originalPrincipal form a valid pair. This method assumes that |
| * authenticatedPrincipal and originalPrincipal can be equal, as long as they are not a proxy role. This use |
| * case is relvant for the admin server because of the way the proxy handles authentication. The binary protocol |
| * should not use this method. |
| * @return true when roles are a valid combination and false when roles are an invalid combination |
| */ |
| public boolean isValidOriginalPrincipal(String authenticatedPrincipal, |
| String originalPrincipal, |
| AuthenticationDataSource authDataSource) { |
| SocketAddress remoteAddress = authDataSource != null ? authDataSource.getPeerAddress() : null; |
| return isValidOriginalPrincipal(authenticatedPrincipal, originalPrincipal, remoteAddress, true); |
| } |
| |
| /** |
| * Validates that the authenticatedPrincipal and the originalPrincipal are a valid combination. |
| * Valid combinations fulfills the following rule: |
| * <p> |
| * The authenticatedPrincipal is in {@link ServiceConfiguration#getProxyRoles()}, if, and only if, |
| * the originalPrincipal is set to a role that is not also in {@link ServiceConfiguration#getProxyRoles()}. |
| * @return true when roles are a valid combination and false when roles are an invalid combination |
| */ |
| public boolean isValidOriginalPrincipal(String authenticatedPrincipal, |
| String originalPrincipal, |
| SocketAddress remoteAddress, |
| boolean allowNonProxyPrincipalsToBeEqual) { |
| String errorMsg = null; |
| if (conf.getProxyRoles().contains(authenticatedPrincipal)) { |
| if (StringUtils.isBlank(originalPrincipal)) { |
| errorMsg = "originalPrincipal must be provided when connecting with a proxy role."; |
| } else if (conf.getProxyRoles().contains(originalPrincipal)) { |
| errorMsg = "originalPrincipal cannot be a proxy role."; |
| } |
| } else if (StringUtils.isNotBlank(originalPrincipal) |
| && !(allowNonProxyPrincipalsToBeEqual && originalPrincipal.equals(authenticatedPrincipal))) { |
| log.warn("[{}] Non-proxy role [{}] passed originalPrincipal [{}]. This behavior will not " |
| + "be allowed in a future release. A proxy's role must be in the broker's proxyRoles " |
| + "configuration.", remoteAddress, authenticatedPrincipal, originalPrincipal); |
| } |
| if (errorMsg != null) { |
| log.warn("[{}] Illegal combination of role [{}] and originalPrincipal [{}]: {}", remoteAddress, |
| authenticatedPrincipal, originalPrincipal, errorMsg); |
| return false; |
| } else { |
| return true; |
| } |
| } |
| |
| private boolean isProxyRole(String role) { |
| return role != null && conf.getProxyRoles().contains(role); |
| } |
| |
| /** |
| * Grant authorization-action permission on a tenant to the given client |
| * |
| * @param tenantName tenant name |
| * @param operation tenant operation |
| * @param role role name |
| * @param authData |
| * additional authdata in json for targeted authorization provider |
| * @return IllegalArgumentException when tenant not found |
| * @throws IllegalStateException |
| * when failed to grant permission |
| */ |
| public CompletableFuture<Boolean> allowTenantOperationAsync(String tenantName, |
| TenantOperation operation, |
| String role, |
| AuthenticationDataSource authData) { |
| if (!this.conf.isAuthorizationEnabled()) { |
| return CompletableFuture.completedFuture(true); |
| } |
| |
| if (provider != null) { |
| return provider.allowTenantOperationAsync(tenantName, role, operation, authData); |
| } |
| |
| return FutureUtil.failedFuture(new IllegalStateException("No authorization provider configured for " + |
| "allowTenantOperationAsync")); |
| } |
| |
| public CompletableFuture<Boolean> allowTenantOperationAsync(String tenantName, |
| TenantOperation operation, |
| String originalRole, |
| String role, |
| AuthenticationDataSource authData) { |
| if (!isValidOriginalPrincipal(role, originalRole, authData)) { |
| return CompletableFuture.completedFuture(false); |
| } |
| if (isProxyRole(role) || StringUtils.isNotBlank(originalRole)) { |
| CompletableFuture<Boolean> isRoleAuthorizedFuture = allowTenantOperationAsync( |
| tenantName, operation, role, authData); |
| CompletableFuture<Boolean> isOriginalAuthorizedFuture = allowTenantOperationAsync( |
| tenantName, operation, originalRole, authData); |
| return isRoleAuthorizedFuture.thenCombine(isOriginalAuthorizedFuture, |
| (isRoleAuthorized, isOriginalAuthorized) -> isRoleAuthorized && isOriginalAuthorized); |
| } else { |
| return allowTenantOperationAsync(tenantName, operation, role, authData); |
| } |
| } |
| |
| public boolean allowTenantOperation(String tenantName, |
| TenantOperation operation, |
| String originalRole, |
| String role, |
| AuthenticationDataSource authData) { |
| try { |
| return allowTenantOperationAsync( |
| tenantName, operation, originalRole, role, authData).get( |
| conf.getZooKeeperOperationTimeoutSeconds(), SECONDS); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| throw new RestException(e); |
| } catch (ExecutionException e) { |
| throw new RestException(e.getCause()); |
| } catch (TimeoutException e) { |
| throw new RestException(e); |
| } |
| } |
| |
| /** |
| * Grant authorization-action permission on a namespace to the given client |
| * |
| * @param namespaceName |
| * @param operation |
| * @param role |
| * @param authData |
| * additional authdata in json for targeted authorization provider |
| * @return IllegalArgumentException when namespace not found |
| * @throws IllegalStateException |
| * when failed to grant permission |
| */ |
| public CompletableFuture<Boolean> allowNamespaceOperationAsync(NamespaceName namespaceName, |
| NamespaceOperation operation, |
| String role, |
| AuthenticationDataSource authData) { |
| if (!this.conf.isAuthorizationEnabled()) { |
| return CompletableFuture.completedFuture(true); |
| } |
| |
| if (provider != null) { |
| return provider.allowNamespaceOperationAsync(namespaceName, role, operation, authData); |
| } |
| |
| return FutureUtil.failedFuture(new IllegalStateException("No authorization provider configured for " + |
| "allowNamespaceOperationAsync")); |
| } |
| |
| public CompletableFuture<Boolean> allowNamespaceOperationAsync(NamespaceName namespaceName, |
| NamespaceOperation operation, |
| String originalRole, |
| String role, |
| AuthenticationDataSource authData) { |
| if (!isValidOriginalPrincipal(role, originalRole, authData)) { |
| return CompletableFuture.completedFuture(false); |
| } |
| if (isProxyRole(role) || StringUtils.isNotBlank(originalRole)) { |
| CompletableFuture<Boolean> isRoleAuthorizedFuture = allowNamespaceOperationAsync( |
| namespaceName, operation, role, authData); |
| CompletableFuture<Boolean> isOriginalAuthorizedFuture = allowNamespaceOperationAsync( |
| namespaceName, operation, originalRole, authData); |
| return isRoleAuthorizedFuture.thenCombine(isOriginalAuthorizedFuture, |
| (isRoleAuthorized, isOriginalAuthorized) -> isRoleAuthorized && isOriginalAuthorized); |
| } else { |
| return allowNamespaceOperationAsync(namespaceName, operation, role, authData); |
| } |
| } |
| |
| public boolean allowNamespaceOperation(NamespaceName namespaceName, |
| NamespaceOperation operation, |
| String originalRole, |
| String role, |
| AuthenticationDataSource authData) { |
| try { |
| return allowNamespaceOperationAsync( |
| namespaceName, operation, originalRole, role, authData).get(); |
| } catch (InterruptedException e) { |
| throw new RestException(e); |
| } catch (ExecutionException e) { |
| throw new RestException(e.getCause()); |
| } |
| } |
| |
| /** |
| * Grant authorization-action permission on a namespace to the given client |
| * |
| * @param namespaceName |
| * @param operation |
| * @param role |
| * @param authData |
| * additional authdata in json for targeted authorization provider |
| * @return IllegalArgumentException when namespace not found |
| * @throws IllegalStateException |
| * when failed to grant permission |
| */ |
| public CompletableFuture<Boolean> allowNamespacePolicyOperationAsync(NamespaceName namespaceName, |
| PolicyName policy, |
| PolicyOperation operation, |
| String role, |
| AuthenticationDataSource authData) { |
| if (!this.conf.isAuthorizationEnabled()) { |
| return CompletableFuture.completedFuture(true); |
| } |
| |
| if (provider != null) { |
| return provider.allowNamespacePolicyOperationAsync(namespaceName, policy, operation, role, authData); |
| } |
| |
| return FutureUtil.failedFuture(new IllegalStateException("No authorization provider configured for " + |
| "allowNamespacePolicyOperationAsync")); |
| } |
| |
| public CompletableFuture<Boolean> allowNamespacePolicyOperationAsync(NamespaceName namespaceName, |
| PolicyName policy, |
| PolicyOperation operation, |
| String originalRole, |
| String role, |
| AuthenticationDataSource authData) { |
| if (!isValidOriginalPrincipal(role, originalRole, authData)) { |
| return CompletableFuture.completedFuture(false); |
| } |
| if (isProxyRole(role) || StringUtils.isNotBlank(originalRole)) { |
| CompletableFuture<Boolean> isRoleAuthorizedFuture = allowNamespacePolicyOperationAsync( |
| namespaceName, policy, operation, role, authData); |
| CompletableFuture<Boolean> isOriginalAuthorizedFuture = allowNamespacePolicyOperationAsync( |
| namespaceName, policy, operation, originalRole, authData); |
| return isRoleAuthorizedFuture.thenCombine(isOriginalAuthorizedFuture, |
| (isRoleAuthorized, isOriginalAuthorized) -> isRoleAuthorized && isOriginalAuthorized); |
| } else { |
| return allowNamespacePolicyOperationAsync(namespaceName, policy, operation, role, authData); |
| } |
| } |
| |
| public boolean allowNamespacePolicyOperation(NamespaceName namespaceName, |
| PolicyName policy, |
| PolicyOperation operation, |
| String originalRole, |
| String role, |
| AuthenticationDataSource authData) { |
| try { |
| return allowNamespacePolicyOperationAsync( |
| namespaceName, policy, operation, originalRole, role, authData).get( |
| conf.getZooKeeperOperationTimeoutSeconds(), SECONDS); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| throw new RestException(e); |
| } catch (ExecutionException e) { |
| throw new RestException(e.getCause()); |
| } catch (TimeoutException e) { |
| throw new RestException(e); |
| } |
| } |
| |
| /** |
| * Grant authorization-action permission on a topic to the given client |
| * |
| * @param topicName |
| * @param policy |
| * @param operation |
| * @param role |
| * @param authData additional authdata in json for targeted authorization provider |
| * @throws IllegalStateException when failed to grant permission |
| */ |
| public CompletableFuture<Boolean> allowTopicPolicyOperationAsync(TopicName topicName, |
| PolicyName policy, |
| PolicyOperation operation, |
| String role, |
| AuthenticationDataSource authData) { |
| if (!this.conf.isAuthorizationEnabled()) { |
| return CompletableFuture.completedFuture(true); |
| } |
| |
| if (provider != null) { |
| return provider.allowTopicPolicyOperationAsync(topicName, role, policy, operation, authData); |
| } |
| |
| return FutureUtil.failedFuture(new IllegalStateException("No authorization provider configured for " + |
| "allowTopicPolicyOperationAsync")); |
| } |
| |
| public CompletableFuture<Boolean> allowTopicPolicyOperationAsync(TopicName topicName, |
| PolicyName policy, |
| PolicyOperation operation, |
| String originalRole, |
| String role, |
| AuthenticationDataSource authData) { |
| if (!isValidOriginalPrincipal(role, originalRole, authData)) { |
| return CompletableFuture.completedFuture(false); |
| } |
| if (isProxyRole(role) || StringUtils.isNotBlank(originalRole)) { |
| CompletableFuture<Boolean> isRoleAuthorizedFuture = allowTopicPolicyOperationAsync( |
| topicName, policy, operation, role, authData); |
| CompletableFuture<Boolean> isOriginalAuthorizedFuture = allowTopicPolicyOperationAsync( |
| topicName, policy, operation, originalRole, authData); |
| return isRoleAuthorizedFuture.thenCombine(isOriginalAuthorizedFuture, |
| (isRoleAuthorized, isOriginalAuthorized) -> isRoleAuthorized && isOriginalAuthorized); |
| } else { |
| return allowTopicPolicyOperationAsync(topicName, policy, operation, role, authData); |
| } |
| } |
| |
| |
| public Boolean allowTopicPolicyOperation(TopicName topicName, |
| PolicyName policy, |
| PolicyOperation operation, |
| String originalRole, |
| String role, |
| AuthenticationDataSource authData) { |
| try { |
| return allowTopicPolicyOperationAsync( |
| topicName, policy, operation, originalRole, role, authData).get( |
| conf.getZooKeeperOperationTimeoutSeconds(), SECONDS); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| throw new RestException(e); |
| } catch (ExecutionException e) { |
| throw new RestException(e.getCause()); |
| } catch (TimeoutException e) { |
| throw new RestException(e); |
| } |
| } |
| |
| /** |
| * Grant authorization-action permission on a topic to the given client |
| * |
| * @param topicName |
| * @param operation |
| * @param role |
| * @param authData |
| * additional authdata in json for targeted authorization provider |
| * @return IllegalArgumentException when namespace not found |
| * @throws IllegalStateException |
| * when failed to grant permission |
| */ |
| public CompletableFuture<Boolean> allowTopicOperationAsync(TopicName topicName, |
| TopicOperation operation, |
| String role, |
| AuthenticationDataSource authData) { |
| if (log.isDebugEnabled()) { |
| log.debug("Check if role {} is allowed to execute topic operation {} on topic {}", |
| role, operation, topicName); |
| } |
| if (!this.conf.isAuthorizationEnabled()) { |
| return CompletableFuture.completedFuture(true); |
| } |
| |
| if (provider != null) { |
| CompletableFuture<Boolean> allowFuture = |
| provider.allowTopicOperationAsync(topicName, role, operation, authData); |
| if (log.isDebugEnabled()) { |
| return allowFuture.whenComplete((allowed, exception) -> { |
| if (exception == null) { |
| if (allowed) { |
| log.debug("Topic operation {} on topic {} is allowed: role = {}", |
| operation, topicName, role); |
| } else { |
| log.debug("Topic operation {} on topic {} is NOT allowed: role = {}", |
| operation, topicName, role); |
| } |
| } else { |
| log.debug("Failed to check if topic operation {} on topic {} is allowed:" |
| + " role = {}", |
| operation, topicName, role, exception); |
| } |
| }); |
| } else { |
| return allowFuture; |
| } |
| } |
| |
| return FutureUtil.failedFuture(new IllegalStateException("No authorization provider configured for " + |
| "allowTopicOperationAsync")); |
| } |
| |
| public CompletableFuture<Boolean> allowTopicOperationAsync(TopicName topicName, |
| TopicOperation operation, |
| String originalRole, |
| String role, |
| AuthenticationDataSource authData) { |
| if (!isValidOriginalPrincipal(role, originalRole, authData)) { |
| return CompletableFuture.completedFuture(false); |
| } |
| if (isProxyRole(role) || StringUtils.isNotBlank(originalRole)) { |
| CompletableFuture<Boolean> isRoleAuthorizedFuture = allowTopicOperationAsync( |
| topicName, operation, role, authData); |
| CompletableFuture<Boolean> isOriginalAuthorizedFuture = allowTopicOperationAsync( |
| topicName, operation, originalRole, authData); |
| return isRoleAuthorizedFuture.thenCombine(isOriginalAuthorizedFuture, |
| (isRoleAuthorized, isOriginalAuthorized) -> isRoleAuthorized && isOriginalAuthorized); |
| } else { |
| return allowTopicOperationAsync(topicName, operation, role, authData); |
| } |
| } |
| |
| public Boolean allowTopicOperation(TopicName topicName, |
| TopicOperation operation, |
| String originalRole, |
| String role, |
| AuthenticationDataSource authData) throws Exception { |
| try { |
| return allowTopicOperationAsync(topicName, operation, originalRole, role, authData).get( |
| conf.getZooKeeperOperationTimeoutSeconds(), SECONDS); |
| } catch (InterruptedException e) { |
| throw new RestException(e); |
| } catch (ExecutionException e) { |
| throw new RestException(e.getCause()); |
| } |
| } |
| } |
