You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ranger.apache.org by rm...@apache.org on 2026/05/19 23:58:21 UTC
(ranger) branch RANGER-3939-patch created (now a2a86fdfb)
This is an automated email from the ASF dual-hosted git repository.
rameeshm pushed a change to branch RANGER-3939-patch
in repository https://gitbox.apache.org/repos/asf/ranger.git
at a2a86fdfb RANGER-3939:Implement acls, createAcls and deleteAcls in Kafka Authorizer
This branch includes the following new commits:
new a2a86fdfb RANGER-3939:Implement acls, createAcls and deleteAcls in Kafka Authorizer
The 1 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.
(ranger) 01/01: RANGER-3939:Implement acls, createAcls and deleteAcls in Kafka Authorizer
Posted by rm...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rameeshm pushed a commit to branch RANGER-3939-patch
in repository https://gitbox.apache.org/repos/asf/ranger.git
commit a2a86fdfb677cb2bba526dab64e3f6688abcc422
Author: Ramesh Mani <rm...@apache.org>
AuthorDate: Sun May 17 23:55:41 2026 -0700
RANGER-3939:Implement acls, createAcls and deleteAcls in Kafka Authorizer
Signed-off-by: Ramesh Mani <rm...@apache.org>
---
plugin-kafka/pom.xml | 8 +
.../kafka/authorizer/RangerKafkaAuditHandler.java | 5 +-
.../kafka/authorizer/RangerKafkaAuthorizer.java | 263 +++-------
.../authorizer/utils/RangerKafkaCheckAccess.java | 97 ++++
.../authorizer/utils/RangerKafkaGrantAccess.java | 182 +++++++
.../authorizer/utils/RangerKafkaListAccess.java | 62 +++
.../authorizer/utils/RangerKafkaRevokeAccess.java | 218 ++++++++
.../kafka/authorizer/utils/RangerKafkaUtils.java | 571 +++++++++++++++++++++
.../java/org/apache/ranger/rest/ServiceREST.java | 45 +-
9 files changed, 1244 insertions(+), 207 deletions(-)
diff --git a/plugin-kafka/pom.xml b/plugin-kafka/pom.xml
index a2eb04af8..3264719c9 100644
--- a/plugin-kafka/pom.xml
+++ b/plugin-kafka/pom.xml
@@ -91,6 +91,14 @@
<artifactId>ranger-plugins-common</artifactId>
<version>${project.version}</version>
<exclusions>
+ <exclusion>
+ <groupId>com.sun.jersey</groupId>
+ <artifactId>jersey-client</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.sun.jersey</groupId>
+ <artifactId>jersey-core</artifactId>
+ </exclusion>
<exclusion>
<groupId>javax.ws.rs</groupId>
<artifactId>jsr311-api</artifactId>
diff --git a/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/RangerKafkaAuditHandler.java b/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/RangerKafkaAuditHandler.java
index 5d6abe2bb..e91656c5c 100644
--- a/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/RangerKafkaAuditHandler.java
+++ b/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/RangerKafkaAuditHandler.java
@@ -20,6 +20,7 @@
package org.apache.ranger.authorization.kafka.authorizer;
import org.apache.ranger.audit.model.AuthzAuditEvent;
+import org.apache.ranger.authorization.kafka.authorizer.utils.RangerKafkaUtils;
import org.apache.ranger.plugin.audit.RangerDefaultAuditHandler;
import org.apache.ranger.plugin.policyengine.RangerAccessRequest;
import org.apache.ranger.plugin.policyengine.RangerAccessResourceImpl;
@@ -83,10 +84,10 @@ private boolean isAuditingNeeded(final RangerAccessResult result) {
boolean isAllowed = result.getIsAllowed();
RangerAccessRequest request = result.getAccessRequest();
RangerAccessResourceImpl resource = (RangerAccessResourceImpl) request.getResource();
- String resourceName = (String) resource.getValue(RangerKafkaAuthorizer.KEY_CLUSTER);
+ String resourceName = (String) resource.getValue(RangerKafkaUtils.KEY_CLUSTER);
if (resourceName != null) {
- if (request.getAccessType().equalsIgnoreCase(RangerKafkaAuthorizer.ACCESS_TYPE_CREATE) && !isAllowed) {
+ if (request.getAccessType().equalsIgnoreCase(RangerKafkaUtils.ACCESS_TYPE_CREATE) && !isAllowed) {
ret = false;
}
}
diff --git a/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/RangerKafkaAuthorizer.java b/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/RangerKafkaAuthorizer.java
index cbfc89a85..32bb8d25a 100644
--- a/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/RangerKafkaAuthorizer.java
+++ b/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/RangerKafkaAuthorizer.java
@@ -19,7 +19,6 @@
package org.apache.ranger.authorization.kafka.authorizer;
-import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.kafka.common.Endpoint;
@@ -27,6 +26,7 @@
import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.errors.ApiException;
import org.apache.kafka.common.network.ListenerName;
import org.apache.kafka.common.resource.ResourceType;
import org.apache.kafka.common.security.JaasContext;
@@ -41,19 +41,18 @@
import org.apache.kafka.server.authorizer.Authorizer;
import org.apache.kafka.server.authorizer.AuthorizerServerInfo;
import org.apache.ranger.audit.provider.MiscUtil;
-import org.apache.ranger.plugin.policyengine.RangerAccessRequest;
-import org.apache.ranger.plugin.policyengine.RangerAccessRequestImpl;
-import org.apache.ranger.plugin.policyengine.RangerAccessResourceImpl;
-import org.apache.ranger.plugin.policyengine.RangerAccessResult;
+import org.apache.ranger.authorization.kafka.authorizer.utils.RangerKafkaCheckAccess;
+import org.apache.ranger.authorization.kafka.authorizer.utils.RangerKafkaGrantAccess;
+import org.apache.ranger.authorization.kafka.authorizer.utils.RangerKafkaListAccess;
+import org.apache.ranger.authorization.kafka.authorizer.utils.RangerKafkaRevokeAccess;
+import org.apache.ranger.authorization.kafka.authorizer.utils.RangerKafkaUtils;
import org.apache.ranger.plugin.service.RangerBasePlugin;
import org.apache.ranger.plugin.util.RangerPerfTracer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.Collections;
-import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -66,27 +65,12 @@ public class RangerKafkaAuthorizer implements Authorizer {
private static final Logger logger = LoggerFactory.getLogger(RangerKafkaAuthorizer.class);
private static final Logger PERF_KAFKAAUTH_REQUEST_LOG = RangerPerfTracer.getPerfLogger("kafkaauth.request");
- public static final String ACCESS_TYPE_ALTER_CONFIGS = "alter_configs";
- public static final String KEY_TOPIC = "topic";
- public static final String KEY_CLUSTER = "cluster";
- public static final String KEY_CONSUMER_GROUP = "consumergroup";
- public static final String KEY_TRANSACTIONALID = "transactionalid";
- public static final String KEY_DELEGATIONTOKEN = "delegationtoken";
- public static final String ACCESS_TYPE_READ = "consume";
- public static final String ACCESS_TYPE_WRITE = "publish";
- public static final String ACCESS_TYPE_CREATE = "create";
- public static final String ACCESS_TYPE_DELETE = "delete";
- public static final String ACCESS_TYPE_CONFIGURE = "configure";
- public static final String ACCESS_TYPE_DESCRIBE = "describe";
- public static final String ACCESS_TYPE_DESCRIBE_CONFIGS = "describe_configs";
- public static final String ACCESS_TYPE_CLUSTER_ACTION = "cluster_action";
- public static final String ACCESS_TYPE_IDEMPOTENT_WRITE = "idempotent_write";
-
private static final String KAFKA_SUPER_USERS_PROP = "super.users";
private static volatile RangerBasePlugin rangerPlugin;
RangerKafkaAuditHandler auditHandler;
+ RangerKafkaUtils rangerKafkaUtils = new RangerKafkaUtils();
public RangerKafkaAuthorizer() {
}
@@ -162,7 +146,7 @@ public List<AuthorizationResult> authorize(AuthorizableRequestContext requestCon
if (rangerPlugin == null) {
MiscUtil.logErrorMessageByInterval(logger, "Authorizer is still not initialized");
- return denyAll(actions);
+ return RangerKafkaUtils.denyAll(actions);
}
RangerPerfTracer perf = null;
@@ -172,7 +156,7 @@ public List<AuthorizationResult> authorize(AuthorizableRequestContext requestCon
}
try {
- return wrappedAuthorization(requestContext, actions);
+ return rangerKafkaUtils.wrappedAuthorization(requestContext, actions, rangerPlugin, auditHandler);
} finally {
RangerPerfTracer.log(perf);
}
@@ -180,202 +164,103 @@ public List<AuthorizationResult> authorize(AuthorizableRequestContext requestCon
@Override
public List<? extends CompletionStage<AclCreateResult>> createAcls(AuthorizableRequestContext requestContext, List<AclBinding> aclBindings) {
- logger.error("createAcls is not supported by Ranger for Kafka");
+ List<? extends CompletionStage<AclCreateResult>> ret;
- return aclBindings.stream()
- .map(ab -> {
- CompletableFuture<AclCreateResult> completableFuture = new CompletableFuture<>();
- completableFuture.completeExceptionally(new UnsupportedOperationException("createAcls is not supported by Ranger for Kafka"));
- return completableFuture;
- })
- .collect(Collectors.toList());
- }
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> RangerKafkaAuthorizer.createAcls(): AuthorizableRequestContext: {} aclBindings: {}", requestContext, aclBindings);
+ }
- @Override
- public List<? extends CompletionStage<AclDeleteResult>> deleteAcls(AuthorizableRequestContext requestContext, List<AclBindingFilter> aclBindingFilters) {
- logger.error("deleteAcls is not supported by Ranger for Kafka");
+ RangerKafkaGrantAccess rangerKafkaGrantAccess = new RangerKafkaGrantAccess();
- return aclBindingFilters.stream()
- .map(ab -> {
- CompletableFuture<AclDeleteResult> completableFuture = new CompletableFuture<>();
- completableFuture.completeExceptionally(new UnsupportedOperationException("deleteAcls is not supported by Ranger for Kafka"));
+ ret = aclBindings.stream()
+ .map(aclBinding -> {
+ CompletableFuture<AclCreateResult> completableFuture = new CompletableFuture<>();
+ completableFuture.complete(rangerKafkaGrantAccess.grant(requestContext, aclBinding, rangerPlugin, auditHandler));
return completableFuture;
})
.collect(Collectors.toList());
- }
- @Override
- public Iterable<AclBinding> acls(AclBindingFilter filter) {
- logger.error("(getting) acls is not supported by Ranger for Kafka");
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== RangerKafkaAuthorizer.createAcls() result: {}", ret);
+ }
- throw new UnsupportedOperationException("(getting) acls is not supported by Ranger for Kafka");
+ return ret;
}
- // TODO: provide a real implementation (RANGER-3809)
- // Currently we return a dummy implementation because KAFKA-13598 makes producers idempotent by default and this causes
- // a failure in the InitProducerId API call on the broker side because of the missing acls() method implementation.
- // Overriding this with a dummy impl will make Kafka return an authorization error instead of an exception if the
- // IDEMPOTENT_WRITE permission wasn't set on the producer.
@Override
- public AuthorizationResult authorizeByResourceType(AuthorizableRequestContext requestContext, AclOperation op, ResourceType resourceType) {
- SecurityUtils.authorizeByResourceTypeCheckArgs(op, resourceType);
-
- logger.debug("authorizeByResourceType call is not supported by Ranger for Kafka yet");
-
- return AuthorizationResult.DENIED;
- }
-
- private static String mapToRangerAccessType(AclOperation operation) {
- switch (operation) {
- case READ:
- return ACCESS_TYPE_READ;
- case WRITE:
- return ACCESS_TYPE_WRITE;
- case ALTER:
- return ACCESS_TYPE_CONFIGURE;
- case DESCRIBE:
- return ACCESS_TYPE_DESCRIBE;
- case CLUSTER_ACTION:
- return ACCESS_TYPE_CLUSTER_ACTION;
- case CREATE:
- return ACCESS_TYPE_CREATE;
- case DELETE:
- return ACCESS_TYPE_DELETE;
- case DESCRIBE_CONFIGS:
- return ACCESS_TYPE_DESCRIBE_CONFIGS;
- case ALTER_CONFIGS:
- return ACCESS_TYPE_ALTER_CONFIGS;
- case IDEMPOTENT_WRITE:
- return ACCESS_TYPE_IDEMPOTENT_WRITE;
- case UNKNOWN:
- case ANY:
- case ALL:
- default:
- return null;
- }
- }
-
- private static String mapToResourceType(ResourceType resourceType) {
- switch (resourceType) {
- case TOPIC:
- return KEY_TOPIC;
- case CLUSTER:
- return KEY_CLUSTER;
- case GROUP:
- return KEY_CONSUMER_GROUP;
- case TRANSACTIONAL_ID:
- return KEY_TRANSACTIONALID;
- case DELEGATION_TOKEN:
- return KEY_DELEGATIONTOKEN;
- case ANY:
- case UNKNOWN:
- default:
- return null;
+ public List<? extends CompletionStage<AclDeleteResult>> deleteAcls(AuthorizableRequestContext requestContext, List<AclBindingFilter> aclBindingFilters) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> RangerKafkaAuthorizer.deleteAcls(): AuthorizableRequestContext: {} aclBindingFilters : {}", requestContext, aclBindingFilters);
}
- }
-
- private static RangerAccessResourceImpl createRangerAccessResource(String resourceTypeKey, String resourceName) {
- RangerAccessResourceImpl rangerResource = new RangerAccessResourceImpl();
-
- rangerResource.setValue(resourceTypeKey, resourceName);
-
- return rangerResource;
- }
-
- private static RangerAccessRequestImpl createRangerAccessRequest(String userName, Set<String> userGroups, String ip,
- Date eventTime, String resourceTypeKey, String resourceName, String accessType) {
- RangerAccessRequestImpl rangerRequest = new RangerAccessRequestImpl();
-
- rangerRequest.setResource(createRangerAccessResource(resourceTypeKey, resourceName));
- rangerRequest.setUser(userName);
- rangerRequest.setUserGroups(userGroups);
- rangerRequest.setClientIPAddress(ip);
- rangerRequest.setAccessTime(eventTime);
- rangerRequest.setAccessType(accessType);
- rangerRequest.setAction(accessType);
- rangerRequest.setRequestData(resourceName);
- return rangerRequest;
- }
+ List<? extends CompletionStage<AclDeleteResult>> ret;
+ List<CompletableFuture<AclDeleteResult>> completableFutures = new ArrayList<>();
- private static List<AuthorizationResult> denyAll(List<Action> actions) {
- return actions.stream().map(a -> AuthorizationResult.DENIED).collect(Collectors.toList());
- }
+ for (AclBindingFilter filter : aclBindingFilters) {
+ RangerKafkaRevokeAccess rangerKafkaRevokeAccess = new RangerKafkaRevokeAccess();
+ CompletableFuture<AclDeleteResult> completableFuture = new CompletableFuture<>();
+ try {
+ AclDeleteResult aclDeleteResult = rangerKafkaRevokeAccess.revoke(filter, rangerPlugin, auditHandler, requestContext);
+ completableFuture.complete(aclDeleteResult);
+ } catch (Exception e) {
+ completableFuture.completeExceptionally(new ApiException(e.getMessage()));
+ }
+ completableFutures.add(completableFuture);
+ }
- private static List<AuthorizationResult> mapResults(List<Action> actions, Collection<RangerAccessResult> results) {
- if (CollectionUtils.isEmpty(results)) {
- logger.error("Ranger Plugin returned null or empty. Returning Denied for all");
+ ret = completableFutures.stream().collect(Collectors.toList());
- return denyAll(actions);
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== RangerKafkaAuthorizer.deleteAcls() result: {}", ret);
}
- return results.stream()
- .map(r -> r != null && r.getIsAllowed() ? AuthorizationResult.ALLOWED : AuthorizationResult.DENIED)
- .collect(Collectors.toList());
- }
-
- private static String toString(AuthorizableRequestContext requestContext) {
- return requestContext == null ? null :
- String.format("AuthorizableRequestContext{principal=%s, clientAddress=%s, clientId=%s}",
- requestContext.principal(), requestContext.clientAddress(), requestContext.clientId());
+ return ret;
}
- private List<AuthorizationResult> wrappedAuthorization(AuthorizableRequestContext requestContext, List<Action> actions) {
- if (CollectionUtils.isEmpty(actions)) {
- return Collections.emptyList();
+ @Override
+ public Iterable<AclBinding> acls(AclBindingFilter filter) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> RangerKafkaAuthorizer.acls(): AclBindingFilter: {}", filter);
}
- String userName = requestContext.principal() == null ? null : requestContext.principal().getName();
- Set<String> userGroups = MiscUtil.getGroupsForRequestUser(userName);
- String hostAddress = requestContext.clientAddress() == null ? null : requestContext.clientAddress().getHostAddress();
- String ip = StringUtils.isNotEmpty(hostAddress) && hostAddress.charAt(0) == '/' ? hostAddress.substring(1) : hostAddress;
- Date eventTime = new Date();
-
- List<RangerAccessRequest> rangerRequests = new ArrayList<>();
-
- for (Action action : actions) {
- String accessType = mapToRangerAccessType(action.operation());
-
- if (accessType == null) {
- MiscUtil.logErrorMessageByInterval(logger, "Unsupported access type, requestContext=" + toString(requestContext) + ", actions=" + actions + ", operation=" + action.operation());
-
- return denyAll(actions);
- }
+ RangerKafkaListAccess rangerKafkaListAccess = new RangerKafkaListAccess();
+ Iterable<AclBinding> ret = rangerKafkaListAccess.getAclBindings(filter, rangerPlugin);
- String resourceTypeKey = mapToResourceType(action.resourcePattern().resourceType());
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== RangerKafkaAuthorizer.acls(): AclBindingFilter: {}", ret);
+ }
- if (resourceTypeKey == null) {
- MiscUtil.logErrorMessageByInterval(logger, "Unsupported resource type, requestContext=" + toString(requestContext) + ", actions=" + actions + ", resourceType=" + action.resourcePattern().resourceType());
+ return ret;
+ }
- return denyAll(actions);
- }
+ /**
+ *
+ * @param requestContext Request context including request resourceType, security protocol and listener name
+ * @param op The ACL operation to check
+ * @param resourceType The resource type to check
+ * @return Return {@link AuthorizationResult#ALLOWED} if the caller is authorized
+ * to perform the given ACL operation on at least one resource of the
+ * given type. Return {@link AuthorizationResult#DENIED} otherwise.
+ */
+ @Override
+ public AuthorizationResult authorizeByResourceType(AuthorizableRequestContext requestContext, AclOperation op, ResourceType resourceType) {
+ AuthorizationResult ret = AuthorizationResult.DENIED;
- RangerAccessRequestImpl rangerAccessRequest = createRangerAccessRequest(userName, userGroups, ip, eventTime, resourceTypeKey, action.resourcePattern().name(), accessType);
+ SecurityUtils.authorizeByResourceTypeCheckArgs(op, resourceType);
- rangerRequests.add(rangerAccessRequest);
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> RangerKafkaAuthorizer.authorizeByResourceType() requestContext: {} AclOperation : {} ResourceType : {}", requestContext, op, resourceType);
}
- Collection<RangerAccessResult> results = callRangerPlugin(rangerRequests);
+ RangerKafkaCheckAccess rangerKafkaCheckAccess = new RangerKafkaCheckAccess();
- List<AuthorizationResult> authorizationResults = mapResults(actions, results);
+ ret = rangerKafkaCheckAccess.authorizeByResourceType(requestContext, op, resourceType, rangerPlugin, auditHandler);
- logger.debug("rangerRequests={}, return={}", rangerRequests, authorizationResults);
-
- return authorizationResults;
- }
-
- private Collection<RangerAccessResult> callRangerPlugin(List<RangerAccessRequest> rangerRequests) {
- try {
- return rangerPlugin.isAccessAllowed(rangerRequests);
- } catch (Throwable t) {
- logger.error("Error while calling isAccessAllowed(). requests={}", rangerRequests, t);
-
- return null;
- } finally {
- if (auditHandler != null) {
- auditHandler.flushAudit();
- }
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== RangerKafkaAuthorizer.authorizeByResourceType() requestContext: {} AclOperation : {} ResourceType : {} AuthorizationResult : {}", requestContext, op, resourceType, ret);
}
+
+ return ret;
}
private Set<String> parseSuperUsersFromKafkaConfig(Map<String, ?> configs) {
diff --git a/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/utils/RangerKafkaCheckAccess.java b/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/utils/RangerKafkaCheckAccess.java
new file mode 100644
index 000000000..9a3dd8657
--- /dev/null
+++ b/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/utils/RangerKafkaCheckAccess.java
@@ -0,0 +1,97 @@
+/*
+ * 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.ranger.authorization.kafka.authorizer.utils;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.kafka.common.acl.AclOperation;
+import org.apache.kafka.common.resource.ResourceType;
+import org.apache.kafka.server.authorizer.AuthorizableRequestContext;
+import org.apache.kafka.server.authorizer.AuthorizationResult;
+import org.apache.ranger.authorization.kafka.authorizer.RangerKafkaAuditHandler;
+import org.apache.ranger.authorization.kafka.authorizer.RangerKafkaAuthorizer;
+import org.apache.ranger.plugin.policyengine.RangerAccessRequest;
+import org.apache.ranger.plugin.policyengine.RangerAccessRequestImpl;
+import org.apache.ranger.plugin.policyengine.RangerAccessResult;
+import org.apache.ranger.plugin.service.RangerBasePlugin;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+public class RangerKafkaCheckAccess {
+ private static final Logger logger = LoggerFactory.getLogger(RangerKafkaAuthorizer.class);
+
+ RangerKafkaUtils rangerKafkaUtils = new RangerKafkaUtils();
+
+ public AuthorizationResult authorizeByResourceType(AuthorizableRequestContext requestContext, AclOperation op, ResourceType resourceType, RangerBasePlugin rangerPlugin, RangerKafkaAuditHandler auditHandler) {
+ AuthorizationResult ret = null;
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> RangerKafkaCheckAccess.authorizeByResourceType() AuthorizableRequestContext : {} AclOperation : {} ResourceType : {}", requestContext, op, resourceType);
+ }
+
+ String principal = requestContext.principal().getName();
+ String userName = rangerKafkaUtils.getPolicyUser(principal);
+ String group = rangerKafkaUtils.getPolicyGroup(principal);
+ Set<String> userGroups = (group != null) ? new HashSet<>(Arrays.asList(group)) : null;
+ String resourceTypeKey = RangerKafkaUtils.mapToResourceType(resourceType);
+ String accessType = RangerKafkaUtils.mapToRangerAccessType(op);
+ String resourceName = StringUtils.EMPTY; // Set Resource to Empty as the access is for any resource of give request type
+ String hostAddress = requestContext.clientAddress() == null ? null : requestContext.clientAddress().getHostAddress();
+ String clientIPAddress = StringUtils.isNotEmpty(hostAddress) && hostAddress.charAt(0) == '/' ? hostAddress.substring(1) : hostAddress;
+
+ RangerAccessRequestImpl rangerAccessRequest = rangerKafkaUtils.createRangerAccessRequest(
+ userName,
+ userGroups,
+ clientIPAddress,
+ new Date(),
+ resourceTypeKey,
+ resourceName,
+ accessType);
+
+ // Resource Matching Scope is set to SELF_OR_PREFIX for this uses case
+ Map<String, RangerAccessRequest.ResourceElementMatchingScope> resourceElementMatchingScopes = rangerAccessRequest.getResourceElementMatchingScopes();
+ resourceElementMatchingScopes.put(resourceTypeKey, RangerAccessRequest.ResourceElementMatchingScope.SELF_OR_PREFIX);
+
+ RangerAccessResult result;
+ try {
+ result = rangerPlugin.isAccessAllowed(rangerAccessRequest);
+ } catch (Throwable t) {
+ logger.error("Error while calling isAccessAllowed(). requests=" + rangerAccessRequest, t);
+ return AuthorizationResult.DENIED;
+ } finally {
+ auditHandler.flushAudit();
+ }
+
+ if (result.getIsAllowed()) {
+ ret = AuthorizationResult.ALLOWED;
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== RangerKafkaCheckAccess.authorizeByResourceType() AuthorizableRequestContext : {} AclOperation : {} ResourceType : {} isAccessAllowed: {}", requestContext, op, resourceType, ret);
+ }
+
+ return ret;
+ }
+}
diff --git a/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/utils/RangerKafkaGrantAccess.java b/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/utils/RangerKafkaGrantAccess.java
new file mode 100644
index 000000000..895243447
--- /dev/null
+++ b/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/utils/RangerKafkaGrantAccess.java
@@ -0,0 +1,182 @@
+/*
+ * 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.ranger.authorization.kafka.authorizer.utils;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.security.AccessControlException;
+import org.apache.kafka.common.acl.AccessControlEntry;
+import org.apache.kafka.common.acl.AclBinding;
+import org.apache.kafka.common.acl.AclOperation;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.resource.PatternType;
+import org.apache.kafka.common.resource.ResourcePattern;
+import org.apache.kafka.common.resource.ResourceType;
+import org.apache.kafka.server.authorizer.AclCreateResult;
+import org.apache.kafka.server.authorizer.AuthorizableRequestContext;
+import org.apache.ranger.audit.provider.MiscUtil;
+import org.apache.ranger.authorization.kafka.authorizer.RangerKafkaAuditHandler;
+import org.apache.ranger.plugin.service.RangerBasePlugin;
+import org.apache.ranger.plugin.util.GrantRevokeRequest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+public class RangerKafkaGrantAccess {
+ private static final Logger logger = LoggerFactory.getLogger(RangerKafkaGrantAccess.class);
+
+ RangerKafkaUtils rangerKafkaUtils = new RangerKafkaUtils();
+
+ // Create GrantData for createACLs
+ public AclCreateResult grant(AuthorizableRequestContext requestContext, AclBinding aclBinding, RangerBasePlugin rangerPlugin, RangerKafkaAuditHandler auditHandler) {
+ AclCreateResult ret = null;
+ GrantRevokeRequest grantRevokeRequest = null;
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> RangerKafkaGrantAccess.grant() AuthorizableRequestContext : {} AclBinding : {}", requestContext, aclBinding);
+ }
+
+ try {
+ grantRevokeRequest = createGrantData(requestContext, aclBinding);
+ logger.info("GrantAccess: {}", grantRevokeRequest);
+ if (rangerPlugin != null) {
+ rangerPlugin.grantAccess(grantRevokeRequest, auditHandler);
+ ret = AclCreateResult.SUCCESS;
+ }
+ } catch (AccessControlException excp) {
+ logger.warn("grant() failed", excp);
+ ret = new AclCreateResult(new ApiException(excp.getMessage()));
+ } catch (IOException excp) {
+ logger.warn("grant() failed", excp);
+ ret = new AclCreateResult(new ApiException(excp.getMessage()));
+ } catch (Exception excp) {
+ logger.warn("grant() failed", excp);
+ ret = new AclCreateResult(new ApiException(excp.getMessage()));
+ } finally {
+ auditHandler.flushAudit();
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== RangerKafkaGrantAccess.grant() AuthorizableRequestContext : {} AclBinding : {} AclCreateResult: {}", requestContext, aclBinding, ret);
+ }
+
+ return ret;
+ }
+
+ private GrantRevokeRequest createGrantData(AuthorizableRequestContext requestContext, AclBinding aclBinding) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> RangerKafkaGrantAccess.createGrantData()");
+ }
+
+ GrantRevokeRequest ret = new GrantRevokeRequest();
+
+ // Grantor Info
+ String grantor = null;
+ Set<String> grantorGroups = null;
+ String principalType = requestContext.principal().getPrincipalType();
+ if ("User".equals(principalType)) {
+ grantor = requestContext.principal().getName();
+ grantorGroups = MiscUtil.getGroupsForRequestUser(grantor);
+ }
+
+ // Client details
+ String hostAddress = requestContext.clientAddress() == null ? null : requestContext.clientAddress().getHostAddress();
+ String clientIPAddress = StringUtils.isNotEmpty(hostAddress) && hostAddress.charAt(0) == '/' ? hostAddress.substring(1) : hostAddress;
+
+ // Kafka Resource
+ ResourcePattern resourcePattern = aclBinding.pattern();
+ ResourceType resourceType = resourcePattern.resourceType();
+ String resourceName = resourcePattern.name();
+ PatternType patternType = resourcePattern.patternType();
+
+ // Build resource
+ String resourceTypeKey = RangerKafkaUtils.mapToResourceType(resourceType);
+ Map<String, String> mapResource = new HashMap<>();
+ switch (resourceType) {
+ case TOPIC:
+ case CLUSTER:
+ case GROUP:
+ case DELEGATION_TOKEN:
+ case TRANSACTIONAL_ID:
+ mapResource.put(resourceTypeKey, rangerKafkaUtils.getResourceValue(patternType, resourceName));
+ break;
+ case ANY:
+ case UNKNOWN:
+ // TODO: throw exception?
+ break;
+ }
+
+ // Kafka Ranger Policy User / Group
+ AccessControlEntry entry = aclBinding.entry();
+ String policyUser = rangerKafkaUtils.getPolicyUser(entry.principal());
+ String policyGroup = rangerKafkaUtils.getPolicyGroup(entry.principal());
+
+ // Kafka Ranger Policy Permissions
+ AclOperation operation = entry.operation();
+
+ // Create GrantRevokeRequest
+ ret.setGrantor(grantor);
+ ret.setGrantorGroups(grantorGroups);
+ ret.setDelegateAdmin(Boolean.TRUE); // remove delegateAdmin privilege as well
+ ret.setEnableAudit(Boolean.TRUE);
+ ret.setReplaceExistingPermissions(Boolean.TRUE);
+ ret.setResource(mapResource);
+ ret.setClientIPAddress(clientIPAddress);
+ ret.setForwardedAddresses(null); // TODO: Need to check with Knox proxy how they handle forwarded address.
+ ret.setRemoteIPAddress(clientIPAddress);
+
+ StringBuilder sb = new StringBuilder("CreateACL: ");
+ sb.append("ResourceType: ").append(resourceTypeKey);
+ sb.append("resourceName: ").append(resourceName);
+ sb.append("operation: ").append(operation);
+ ret.setRequestData(sb.toString());
+
+ Set<String> users = new HashSet<>();
+ if (policyUser != null) {
+ users.add(policyUser);
+ ret.setUsers(users);
+ }
+
+ Set<String> groups = new HashSet<>();
+ if (policyGroup != null) {
+ groups.add(policyGroup);
+ ret.setGroups(groups);
+ }
+
+ String accessType = RangerKafkaUtils.mapToRangerAccessType(operation);
+ if (accessType == null) {
+ throw new IllegalArgumentException("Unsupported accessType in the request..");
+ }
+
+ Set<String> accessTypes = new HashSet<>();
+ accessTypes.add(accessType);
+ ret.setAccessTypes(accessTypes);
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== RangerKafkaGrantAccess.createGrantData(): {}", ret);
+ }
+
+ return ret;
+ }
+}
diff --git a/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/utils/RangerKafkaListAccess.java b/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/utils/RangerKafkaListAccess.java
new file mode 100644
index 000000000..84a8855d6
--- /dev/null
+++ b/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/utils/RangerKafkaListAccess.java
@@ -0,0 +1,62 @@
+/*
+ * 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.ranger.authorization.kafka.authorizer.utils;
+
+import org.apache.kafka.common.acl.AclBinding;
+import org.apache.kafka.common.acl.AclBindingFilter;
+import org.apache.ranger.plugin.policyengine.RangerAccessRequest;
+import org.apache.ranger.plugin.policyengine.RangerResourceACLs;
+import org.apache.ranger.plugin.service.RangerBasePlugin;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class RangerKafkaListAccess {
+ private static final Logger logger = LoggerFactory.getLogger(RangerKafkaListAccess.class);
+ RangerKafkaUtils rangerKafkaUtils = new RangerKafkaUtils();
+
+ public RangerKafkaListAccess() {
+ }
+
+ public Iterable<AclBinding> getAclBindings(AclBindingFilter filter, RangerBasePlugin rangerPlugin) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> RangerKafkaListAccess.getAclBindings() AclBindingFilter : {}", filter);
+ }
+
+ List<AclBinding> ret = new ArrayList<>();
+ try {
+ List<RangerAccessRequest> accessRequests = rangerKafkaUtils.createRangerRequests(filter);
+ for (RangerAccessRequest accessRequest : accessRequests) {
+ RangerResourceACLs rangerResourceACLs = rangerPlugin.getResourceACLs(accessRequest);
+ ret.addAll(rangerKafkaUtils.getKafkaAclBindings(rangerResourceACLs, filter, RangerKafkaUtils.AclType.LIST_ACCES));
+ }
+ } catch (Exception e) {
+ logger.error("Unable to fetch ACLs for {}", filter, e);
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== RangerKafkaListAccess.getAclBindings() AclBindingFilter : {}", ret);
+ }
+
+ return ret;
+ }
+}
diff --git a/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/utils/RangerKafkaRevokeAccess.java b/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/utils/RangerKafkaRevokeAccess.java
new file mode 100644
index 000000000..5175ec2f2
--- /dev/null
+++ b/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/utils/RangerKafkaRevokeAccess.java
@@ -0,0 +1,218 @@
+/*
+ * 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.ranger.authorization.kafka.authorizer.utils;
+
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.security.AccessControlException;
+import org.apache.kafka.common.acl.AccessControlEntry;
+import org.apache.kafka.common.acl.AclBinding;
+import org.apache.kafka.common.acl.AclBindingFilter;
+import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.resource.PatternType;
+import org.apache.kafka.common.resource.ResourcePattern;
+import org.apache.kafka.common.resource.ResourceType;
+import org.apache.kafka.server.authorizer.AclDeleteResult;
+import org.apache.kafka.server.authorizer.AuthorizableRequestContext;
+import org.apache.ranger.audit.provider.MiscUtil;
+import org.apache.ranger.authorization.kafka.authorizer.RangerKafkaAuditHandler;
+import org.apache.ranger.plugin.service.RangerBasePlugin;
+import org.apache.ranger.plugin.util.GrantRevokeRequest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+public class RangerKafkaRevokeAccess {
+ private static final Logger logger = LoggerFactory.getLogger(RangerKafkaRevokeAccess.class);
+
+ RangerKafkaUtils rangerKafkaUtils = new RangerKafkaUtils();
+
+ public AclDeleteResult revoke(AclBindingFilter filter, RangerBasePlugin rangerPlugin, RangerKafkaAuditHandler auditHandler, AuthorizableRequestContext requestContext) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> RangerKafkaRevokeAccess.revoke(): AuthorizableRequestContext: {} aclBindingFilter : {}", requestContext, filter);
+ }
+
+ Collection<AclDeleteResult.AclBindingDeleteResult> ret = new ArrayList<>();
+ Iterable<AclBinding> aclBindingsForUserOrGroup = rangerKafkaUtils.getAclBindingsForUserOrGroup(filter, rangerPlugin, RangerKafkaUtils.AclType.REVOKE_ACCESS);
+ if (aclBindingsForUserOrGroup != null && !aclBindingsForUserOrGroup.iterator().hasNext()) {
+ ret.add(new AclDeleteResult.AclBindingDeleteResult(rangerKafkaUtils.getAclBindingForFilter(filter)));
+ } else {
+ AclDeleteResult.AclBindingDeleteResult aclBindingDeleteResult = getAclDeleteResultForUserOrGroupAclBinding(aclBindingsForUserOrGroup, rangerPlugin, auditHandler, requestContext);
+ ret.add(aclBindingDeleteResult);
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== RangerKafkaRevokeAccess.revoke(): AclDeleteResult.AclBindingDeleteResults: {}", ret);
+ }
+
+ return new AclDeleteResult(ret);
+ }
+
+ public AclDeleteResult.AclBindingDeleteResult getAclDeleteResultForUserOrGroupAclBinding(Iterable<AclBinding> aclBindings, RangerBasePlugin rangerPlugin, RangerKafkaAuditHandler auditHandler, AuthorizableRequestContext requestContext) {
+ AclDeleteResult.AclBindingDeleteResult ret = null;
+ GrantRevokeRequest grantRevokeRequest = null;
+ AclBinding aclBinding = null;
+ try {
+ // Here aclBindings are for a single user/group hence fetch the first aclbinding
+ aclBinding = getAclBindingForUserOrGroup(aclBindings);
+ // Fetch all the operations to be revoked from the aclBindings
+ Set<String> rangerOperations = getRangerOperationsFromAclBindings(aclBindings);
+ if (CollectionUtils.isEmpty(rangerOperations)) {
+ throw new IllegalArgumentException("Unsupported accessType in the request..");
+ }
+ grantRevokeRequest = createRevokeData(requestContext, aclBinding, rangerOperations);
+ if (logger.isDebugEnabled()) {
+ logger.debug("RevokeAccessRequest: {}", grantRevokeRequest);
+ }
+ if (rangerPlugin != null) {
+ rangerPlugin.revokeAccess(grantRevokeRequest, auditHandler);
+ ret = new AclDeleteResult.AclBindingDeleteResult(aclBinding);
+ }
+ } catch (AccessControlException excp) {
+ logger.error("revoke() failed", excp);
+ ret = new AclDeleteResult.AclBindingDeleteResult(aclBinding, new ApiException(excp));
+ } catch (IOException excp) {
+ logger.error("revoke() failed", excp);
+ ret = new AclDeleteResult.AclBindingDeleteResult(aclBinding, new ApiException(excp));
+ } catch (Exception excp) {
+ logger.error("revoke() failed", excp);
+ ret = new AclDeleteResult.AclBindingDeleteResult(aclBinding, new ApiException(excp));
+ } finally {
+ auditHandler.flushAudit();
+ }
+ return ret;
+ }
+
+ public GrantRevokeRequest createRevokeData(AuthorizableRequestContext requestContext, AclBinding aclBinding, Set<String> operations) throws Exception {
+ GrantRevokeRequest ret = new GrantRevokeRequest();
+
+ // Revoker Info
+ String revoker = null;
+ Set<String> revokerGroups = null;
+ String principalType = requestContext.principal().getPrincipalType();
+ if ("User".equals(principalType)) {
+ revoker = requestContext.principal().getName();
+ revokerGroups = MiscUtil.getGroupsForRequestUser(revoker);
+ }
+
+ // Client details
+ String hostAddress = requestContext.clientAddress() == null ? null : requestContext.clientAddress().getHostAddress();
+ String clientIPAddress = StringUtils.isNotEmpty(hostAddress) && hostAddress.charAt(0) == '/' ? hostAddress.substring(1) : hostAddress;
+
+ // Kafka Resource
+ ResourcePattern resourcePattern = aclBinding.pattern();
+ ResourceType resourceType = resourcePattern.resourceType();
+ String resourceName = resourcePattern.name();
+ PatternType patternType = resourcePattern.patternType();
+
+ // Build resource
+ String resourceTypeKey = RangerKafkaUtils.mapToResourceType(resourceType);
+ Map<String, String> mapResource = new HashMap<>();
+ switch (resourceType) {
+ case TOPIC:
+ case CLUSTER:
+ case GROUP:
+ case DELEGATION_TOKEN:
+ case TRANSACTIONAL_ID:
+ mapResource.put(resourceTypeKey, rangerKafkaUtils.getResourceValue(patternType, resourceName));
+ break;
+ case ANY:
+ case UNKNOWN:
+ // TODO: throw exception?
+ break;
+ }
+
+ // Kafka Ranger Policy User / Group
+ AccessControlEntry accessControlEntry = aclBinding.entry();
+ String policyUser = rangerKafkaUtils.getPolicyUser(accessControlEntry.principal());
+ String policyGroup = rangerKafkaUtils.getPolicyGroup(accessControlEntry.principal());
+
+ // Kafka Ranger Policy Permissions
+ String operation = RangerKafkaUtils.mapToRangerAccessType(accessControlEntry.operation());
+ if (operation == null) {
+ throw new IllegalArgumentException("Unsupported accessType in the request..");
+ }
+
+ // Create GrantRevokeRequest
+ ret.setGrantor(revoker);
+ ret.setGrantorGroups(revokerGroups);
+ ret.setDelegateAdmin(Boolean.TRUE); // remove delegateAdmin privilege as well
+ ret.setEnableAudit(Boolean.TRUE);
+ ret.setReplaceExistingPermissions(Boolean.TRUE);
+ ret.setResource(mapResource);
+ ret.setClientIPAddress(clientIPAddress);
+ ret.setForwardedAddresses(null);
+ ret.setRemoteIPAddress(clientIPAddress);
+
+ StringBuilder sb = new StringBuilder("DeleteAcls:");
+ sb.append(" ResourceType: ").append(resourceTypeKey);
+ sb.append(" ResourceName: ").append(resourceName);
+ sb.append(" Operation: ").append(operations);
+ ret.setRequestData(sb.toString());
+
+ Set<String> users = new HashSet<>();
+ if (policyUser != null) {
+ users.add(policyUser);
+ ret.setUsers(users);
+ }
+
+ Set<String> groups = new HashSet<>();
+ if (policyGroup != null) {
+ groups.add(policyGroup);
+ ret.setGroups(groups);
+ }
+
+ Set<String> accessTypes = new HashSet<>();
+ accessTypes.addAll(operations);
+ ret.setAccessTypes(accessTypes);
+
+ ret.setDelegateAdmin(false);
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("CREATED REVOKE Request: {}", ret);
+ }
+
+ return ret;
+ }
+
+ private AclBinding getAclBindingForUserOrGroup(Iterable<AclBinding> aclBindings) {
+ AclBinding ret = null;
+ if (aclBindings.iterator().hasNext()) {
+ ret = aclBindings.iterator().next();
+ }
+ return ret;
+ }
+
+ private Set<String> getRangerOperationsFromAclBindings(Iterable<AclBinding> aclBindings) {
+ Set<String> ret = new HashSet<>();
+ for (AclBinding aclBinding : aclBindings) {
+ AccessControlEntry accessControlEntry = aclBinding.entry();
+ ret.add(RangerKafkaUtils.mapToRangerAccessType(accessControlEntry.operation()));
+ }
+ return ret;
+ }
+}
diff --git a/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/utils/RangerKafkaUtils.java b/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/utils/RangerKafkaUtils.java
new file mode 100644
index 000000000..900681b3d
--- /dev/null
+++ b/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/utils/RangerKafkaUtils.java
@@ -0,0 +1,571 @@
+/*
+ * 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.ranger.authorization.kafka.authorizer.utils;
+
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections.MapUtils;
+import org.apache.commons.lang.ArrayUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.kafka.common.acl.AccessControlEntry;
+import org.apache.kafka.common.acl.AccessControlEntryFilter;
+import org.apache.kafka.common.acl.AclBinding;
+import org.apache.kafka.common.acl.AclBindingFilter;
+import org.apache.kafka.common.acl.AclOperation;
+import org.apache.kafka.common.acl.AclPermissionType;
+import org.apache.kafka.common.resource.PatternType;
+import org.apache.kafka.common.resource.ResourcePattern;
+import org.apache.kafka.common.resource.ResourcePatternFilter;
+import org.apache.kafka.common.resource.ResourceType;
+import org.apache.kafka.server.authorizer.Action;
+import org.apache.kafka.server.authorizer.AuthorizableRequestContext;
+import org.apache.kafka.server.authorizer.AuthorizationResult;
+import org.apache.ranger.audit.provider.MiscUtil;
+import org.apache.ranger.authorization.kafka.authorizer.RangerKafkaAuditHandler;
+import org.apache.ranger.authorization.kafka.authorizer.RangerKafkaAuthorizer;
+import org.apache.ranger.plugin.policyengine.RangerAccessRequest;
+import org.apache.ranger.plugin.policyengine.RangerAccessRequestImpl;
+import org.apache.ranger.plugin.policyengine.RangerAccessResourceImpl;
+import org.apache.ranger.plugin.policyengine.RangerAccessResult;
+import org.apache.ranger.plugin.policyengine.RangerPolicyEngine;
+import org.apache.ranger.plugin.policyengine.RangerResourceACLs;
+import org.apache.ranger.plugin.service.RangerBasePlugin;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class RangerKafkaUtils {
+ private static final Logger logger = LoggerFactory.getLogger(RangerKafkaAuthorizer.class);
+
+ public static final String ACCESS_TYPE_ALTER_CONFIGS = "alter_configs";
+ public static final String KEY_TOPIC = "topic";
+ public static final String KEY_CLUSTER = "cluster";
+ public static final String KEY_CONSUMER_GROUP = "consumergroup";
+ public static final String KEY_TRANSACTIONALID = "transactionalid";
+ public static final String KEY_DELEGATIONTOKEN = "delegationtoken";
+ public static final String ACCESS_TYPE_READ = "consume";
+ public static final String ACCESS_TYPE_WRITE = "publish";
+ public static final String ACCESS_TYPE_CREATE = "create";
+ public static final String ACCESS_TYPE_ALTER = "alter";
+ public static final String ACCESS_TYPE_DELETE = "delete";
+ public static final String ACCESS_TYPE_CONFIGURE = "configure";
+ public static final String ACCESS_TYPE_DESCRIBE = "describe";
+ public static final String ACCESS_TYPE_DESCRIBE_CONFIGS = "describe_configs";
+ public static final String ACCESS_TYPE_CLUSTER_ACTION = "cluster_action";
+ public static final String ACCESS_TYPE_IDEMPOTENT_WRITE = "idempotent_write";
+ public static final String ACCESS_TYPE_ADMIN = "_admin";
+ public static final String ACCESS_TYPE_KAFKA_ADMIN = "kafka_admin";
+ public static final String ACCESS_TYPE_ANY = "any";
+ public static final String ACCESS_TYPE_UNKNOWN = "UNKNOWN";
+ public static final String WILDCARD_ASTERIX = "*";
+
+ public enum AclType { LIST_ACCES, CHECK_ACCESS, GRANT_ACCESS, REVOKE_ACCESS }
+
+ public static String mapToRangerAccessType(AclOperation operation) {
+ switch (operation) {
+ case READ:
+ return ACCESS_TYPE_READ;
+ case WRITE:
+ return ACCESS_TYPE_WRITE;
+ case ALTER:
+ return ACCESS_TYPE_CONFIGURE;
+ case DESCRIBE:
+ return ACCESS_TYPE_DESCRIBE;
+ case CLUSTER_ACTION:
+ return ACCESS_TYPE_CLUSTER_ACTION;
+ case CREATE:
+ return ACCESS_TYPE_CREATE;
+ case DELETE:
+ return ACCESS_TYPE_DELETE;
+ case DESCRIBE_CONFIGS:
+ return ACCESS_TYPE_DESCRIBE_CONFIGS;
+ case ALTER_CONFIGS:
+ return ACCESS_TYPE_ALTER_CONFIGS;
+ case IDEMPOTENT_WRITE:
+ return ACCESS_TYPE_IDEMPOTENT_WRITE;
+ case UNKNOWN:
+ case ANY:
+ case ALL:
+ default:
+ return null;
+ }
+ }
+
+ public static String mapToResourceType(ResourceType resourceType) {
+ switch (resourceType) {
+ case TOPIC:
+ return KEY_TOPIC;
+ case CLUSTER:
+ return KEY_CLUSTER;
+ case GROUP:
+ return KEY_CONSUMER_GROUP;
+ case TRANSACTIONAL_ID:
+ return KEY_TRANSACTIONALID;
+ case DELEGATION_TOKEN:
+ return KEY_DELEGATIONTOKEN;
+ case ANY:
+ case UNKNOWN:
+ default:
+ return null;
+ }
+ }
+
+ public Iterable<AclBinding> getAclBindingsForUserOrGroup(AclBindingFilter filter, RangerBasePlugin rangerPlugin, RangerKafkaUtils.AclType aclType) throws Exception {
+ List<AclBinding> ret = new ArrayList<>();
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> RangerKafkaListAccess.getAclBindingsForUserOrGroup() AclBindingFilter :{}", filter);
+ }
+
+ List<RangerAccessRequest> accessRequests = createRangerRequests(filter);
+ for (RangerAccessRequest accessRequest : accessRequests) {
+ RangerResourceACLs rangerResourceACLs = rangerPlugin.getResourceACLs(accessRequest);
+ ret.addAll(getKafkaAclBindings(rangerResourceACLs, filter, aclType));
+ }
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== RangerKafkaListAccess.getAclBindingsForUserOrGroup() AclBindingFilter :{}", ret);
+ }
+ return ret;
+ }
+
+ public List<RangerAccessRequest> createRangerRequests(AclBindingFilter aclBindingFilter) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> RangerKafkaUtils.createRangerRequests() : aclBindingFilter {}", aclBindingFilter);
+ }
+
+ List<RangerAccessRequest> ret = new ArrayList<>();
+
+ ResourcePatternFilter resourcePatternFilter = aclBindingFilter.patternFilter();
+ ResourceType resourceType = getResourceType(resourcePatternFilter);
+ PatternType patternType = getResourcePatternType(resourcePatternFilter);
+ String resourceName = getResourceName(resourcePatternFilter);
+
+ // Build Ranger resource
+ Set<ResourceType> resourceTypes = new HashSet<>();
+ if (RangerKafkaUtils.ACCESS_TYPE_ANY.equalsIgnoreCase(resourceType.name())) {
+ resourceTypes = new HashSet<>(Arrays.asList(ResourceType.values()));
+ } else if (!RangerKafkaUtils.ACCESS_TYPE_UNKNOWN.equalsIgnoreCase(resourceType.name())) {
+ resourceTypes.add(resourceType);
+ }
+
+ // Build request for each of the permission for the user
+ for (ResourceType resType : resourceTypes) {
+ String resourceTypeKey;
+ if (ResourceType.UNKNOWN.equals(resType) || ResourceType.ANY.equals(resType)) {
+ throw new Exception("Unsupported ResoureType " + resType + " in the request..");
+ }
+
+ Map<String, String> mapResource = new HashMap<>();
+ resourceTypeKey = RangerKafkaUtils.mapToResourceType(resType);
+ mapResource.put(resourceTypeKey, getResourceValue(patternType, resourceName));
+
+ AccessControlEntryFilter accessControlEntryFilter = aclBindingFilter.entryFilter();
+ AclOperation aclOperation = accessControlEntryFilter.operation();
+ String accessType = RangerKafkaUtils.mapToRangerAccessType(aclOperation);
+ if (StringUtils.isBlank(accessType)) {
+ accessType = "_any";
+ }
+
+ RangerAccessResourceImpl kafkaResource = createRangerAccessResource(resourceTypeKey, resourceName);
+ String user = getPolicyUser(accessControlEntryFilter.principal());
+ String group = getPolicyGroup(accessControlEntryFilter.principal());
+ String ip = accessControlEntryFilter.host();
+ RangerAccessRequestImpl request = new RangerAccessRequestImpl();
+ request.setResource(kafkaResource);
+ request.setAccessType(accessType);
+ request.setUser(user);
+ request.setUserGroups((group != null) ? new HashSet<>(Arrays.asList(group)) : null);
+ request.setAccessTime(new Date());
+ request.setClientIPAddress(ip);
+ ret.add(request);
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== RangerKafkaUtils.createRangerRequests() {}", ret);
+ }
+
+ return ret;
+ }
+
+ public RangerAccessResourceImpl createRangerAccessResource(String resourceTypeKey, String resourceName) {
+ RangerAccessResourceImpl rangerResource = new RangerAccessResourceImpl();
+ rangerResource.setValue(resourceTypeKey, resourceName);
+ return rangerResource;
+ }
+
+ public RangerAccessRequestImpl createRangerAccessRequest(String userName,
+ Set<String> userGroups,
+ String ip,
+ Date eventTime,
+ String resourceTypeKey,
+ String resourceName,
+ String accessType) {
+ RangerAccessRequestImpl rangerRequest = new RangerAccessRequestImpl();
+ rangerRequest.setResource(createRangerAccessResource(resourceTypeKey, resourceName));
+ rangerRequest.setUser(userName);
+ rangerRequest.setUserGroups(userGroups);
+ rangerRequest.setClientIPAddress(ip);
+ rangerRequest.setAccessTime(eventTime);
+ rangerRequest.setAccessType(accessType);
+ rangerRequest.setAction(accessType);
+ rangerRequest.setRequestData(resourceName);
+ return rangerRequest;
+ }
+
+ public static List<AuthorizationResult> mapResults(List<Action> actions, Collection<RangerAccessResult> results) {
+ if (CollectionUtils.isEmpty(results)) {
+ logger.error("Ranger Plugin returned null or empty. Returning Denied for all");
+ return denyAll(actions);
+ }
+ return results.stream()
+ .map(r -> r != null && r.getIsAllowed() ? AuthorizationResult.ALLOWED : AuthorizationResult.DENIED)
+ .collect(Collectors.toList());
+ }
+
+ public static List<AuthorizationResult> denyAll(List<Action> actions) {
+ return actions.stream().map(a -> AuthorizationResult.DENIED).collect(Collectors.toList());
+ }
+
+ public static String toString(AuthorizableRequestContext requestContext) {
+ return requestContext == null ? null :
+ String.format("AuthorizableRequestContext{principal=%s, clientAddress=%s, clientId=%s}",
+ requestContext.principal(), requestContext.clientAddress(), requestContext.clientId());
+ }
+
+ public String getPolicyUser(String principal) {
+ String ret = null;
+ if (logger.isDebugEnabled()) {
+ logger.debug("getPolicyUser(): Principal : {}", principal);
+ }
+
+ if (principal != null) {
+ if (principal.contains(":")) {
+ String[] userData = principal.split(":");
+ if (ArrayUtils.isNotEmpty(userData)) {
+ if ("User".equalsIgnoreCase(userData[0])) {
+ ret = userData[1];
+ }
+ }
+ } else if (principal.equals("<any>")) {
+ // Access check is for any user so returning public so group can be public
+ ret = "any";
+ }
+ } else {
+ return StringUtils.EMPTY;
+ }
+ return ret;
+ }
+
+ protected String getPolicyGroup(String principal) {
+ String ret = null;
+ if (logger.isDebugEnabled()) {
+ logger.debug("getPolicyGroup(): Principal : {}", principal);
+ }
+ logger.info("#### getPolicyGroup(): Principal : {}", principal);
+
+ if (principal != null && principal.contains(":")) {
+ String[] userData = principal.split(":");
+ if (ArrayUtils.isNotEmpty(userData)) {
+ if ("Group".equalsIgnoreCase(userData[0])) {
+ ret = userData[1];
+ }
+ }
+ }
+ return ret;
+ }
+
+ protected Set<String> getUserGroups(String userName) throws Exception {
+ return MiscUtil.getGroupsForRequestUser(userName);
+ }
+
+ protected String getAllowedHost(String host) {
+ String ret = host;
+ if ("*".equals(host)) {
+ ret = "";
+ }
+ return ret;
+ }
+
+ protected String getResourceValue(PatternType patternType, String resource) {
+ String ret = resource;
+ switch (patternType) {
+ case LITERAL:
+ ret = resource;
+ break;
+ case PREFIXED:
+ ret = resource + "*";
+ break;
+ case ANY:
+ case MATCH:
+ ret = resource + "*";
+ break;
+ }
+ return ret;
+ }
+
+ public List<AuthorizationResult> wrappedAuthorization(AuthorizableRequestContext requestContext, List<Action> actions, RangerBasePlugin rangerPlugin, RangerKafkaAuditHandler auditHandler) {
+ if (CollectionUtils.isEmpty(actions)) {
+ return Collections.emptyList();
+ }
+ String userName = requestContext.principal() == null ? null : requestContext.principal().getName();
+ Set<String> userGroups = MiscUtil.getGroupsForRequestUser(userName);
+ String hostAddress = requestContext.clientAddress() == null ? null : requestContext.clientAddress().getHostAddress();
+ String ip = StringUtils.isNotEmpty(hostAddress) && hostAddress.charAt(0) == '/' ? hostAddress.substring(1) : hostAddress;
+ Date eventTime = new Date();
+
+ List<RangerAccessRequest> rangerRequests = new ArrayList<>();
+ for (Action action : actions) {
+ String accessType = RangerKafkaUtils.mapToRangerAccessType(action.operation());
+ if (accessType == null) {
+ MiscUtil.logErrorMessageByInterval(logger, "Unsupported access type, requestContext=" + RangerKafkaUtils.toString(requestContext) +
+ ", actions=" + actions + ", operation=" + action.operation());
+ return RangerKafkaUtils.denyAll(actions);
+ }
+ String resourceTypeKey = RangerKafkaUtils.mapToResourceType(action.resourcePattern().resourceType());
+ if (resourceTypeKey == null) {
+ MiscUtil.logErrorMessageByInterval(logger, "Unsupported resource type, requestContext=" + RangerKafkaUtils.toString(requestContext) +
+ ", actions=" + actions + ", resourceType=" + action.resourcePattern().resourceType());
+ return RangerKafkaUtils.denyAll(actions);
+ }
+
+ RangerAccessRequestImpl rangerAccessRequest = createRangerAccessRequest(
+ userName,
+ userGroups,
+ ip,
+ eventTime,
+ resourceTypeKey,
+ action.resourcePattern().name(),
+ accessType);
+ rangerRequests.add(rangerAccessRequest);
+ }
+
+ Collection<RangerAccessResult> results = callRangerPlugin(rangerRequests, rangerPlugin, auditHandler);
+
+ List<AuthorizationResult> authorizationResults = RangerKafkaUtils.mapResults(actions, results);
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("rangerRequests={}, return={}", rangerRequests, authorizationResults);
+ }
+ return authorizationResults;
+ }
+
+ private Collection<RangerAccessResult> callRangerPlugin(List<RangerAccessRequest> rangerRequests, RangerBasePlugin rangerPlugin, RangerKafkaAuditHandler auditHandler) {
+ try {
+ return rangerPlugin.isAccessAllowed(rangerRequests);
+ } catch (Throwable t) {
+ logger.error("Error while calling isAccessAllowed(). requests={}", rangerRequests, t);
+ return null;
+ } finally {
+ auditHandler.flushAudit();
+ }
+ }
+
+ public List<AclBinding> getKafkaAclBindings(RangerResourceACLs rangerResourceACLs, AclBindingFilter filter, RangerKafkaUtils.AclType aclType) {
+ List<AclBinding> ret = new ArrayList<>();
+ if (rangerResourceACLs != null) {
+ Map<String, Map<String, RangerResourceACLs.AccessResult>> userRangerACLs = rangerResourceACLs.getUserACLs();
+ if (logger.isDebugEnabled()) {
+ logger.debug("RangerKafkaUtils.getKafkaAclBindings() userRangerAcls: {}", userRangerACLs);
+ }
+ List<AclBinding> userAclBindings = getAclBindingsForUserAcls(userRangerACLs, filter, aclType);
+ if (CollectionUtils.isNotEmpty(userAclBindings)) {
+ ret.addAll(userAclBindings);
+ }
+ Map<String, Map<String, RangerResourceACLs.AccessResult>> groupRangerACLs = rangerResourceACLs.getGroupACLs();
+ if (logger.isDebugEnabled()) {
+ logger.debug("RangerKafkaUtils.getKafkaAclBindings() groupRangerAcls: {}", groupRangerACLs);
+ }
+ List<AclBinding> groupAclBindins = getAclBindingsForGroupAcls(groupRangerACLs, filter, aclType);
+ if (CollectionUtils.isNotEmpty(groupAclBindins)) {
+ ret.addAll(groupAclBindins);
+ }
+ }
+ return ret;
+ }
+
+ public List<AclBinding> getAclBindingsForUserAcls(Map<String, Map<String, RangerResourceACLs.AccessResult>> userRangerACLs, AclBindingFilter filter, RangerKafkaUtils.AclType aclType) {
+ List<AclBinding> ret = new ArrayList<>();
+ String user = getPolicyUser(filter.entryFilter().principal());
+ if (StringUtils.isNotBlank(user) && MapUtils.isNotEmpty(userRangerACLs)) {
+ for (Map.Entry<String, Map<String, RangerResourceACLs.AccessResult>> entry : userRangerACLs.entrySet()) {
+ String userName = entry.getKey();
+ if (StringUtils.isNotBlank(user)) {
+ if (!"any".equals(user)
+ && !user.equals(userName) && !"*".equals(userName)) {
+ continue;
+ }
+ }
+ for (Map.Entry<String, RangerResourceACLs.AccessResult> privilege : entry.getValue().entrySet()) {
+ if (StringUtils.equals(RangerPolicyEngine.ADMIN_ACCESS, privilege.getKey())) {
+ continue;
+ }
+ ResourcePattern resourcePattern = getKafkaResourcePattern(filter);
+ AccessControlEntry accessControlEntry = getKafkaAccessControlEntry("USER:" + userName, filter.entryFilter().host(), privilege);
+ ret.add(new AclBinding(resourcePattern, accessControlEntry));
+ }
+ }
+ } else if (StringUtils.isBlank(user)
+ && MapUtils.isNotEmpty(userRangerACLs)
+ && aclType.equals(RangerKafkaUtils.AclType.LIST_ACCES)) {
+ for (Map.Entry<String, Map<String, RangerResourceACLs.AccessResult>> entry : userRangerACLs.entrySet()) {
+ String userName = entry.getKey();
+ for (Map.Entry<String, RangerResourceACLs.AccessResult> privilege : entry.getValue().entrySet()) {
+ if (StringUtils.equals(RangerPolicyEngine.ADMIN_ACCESS, privilege.getKey())) {
+ continue;
+ }
+ ResourcePattern resourcePattern = getKafkaResourcePattern(filter);
+ AccessControlEntry accessControlEntry = getKafkaAccessControlEntry("USER:" + userName, filter.entryFilter().host(), privilege);
+ ret.add(new AclBinding(resourcePattern, accessControlEntry));
+ }
+ }
+ }
+ return ret;
+ }
+
+ public List<AclBinding> getAclBindingsForGroupAcls(Map<String, Map<String, RangerResourceACLs.AccessResult>> groupRangerACLs, AclBindingFilter filter, RangerKafkaUtils.AclType aclType) {
+ List<AclBinding> ret = new ArrayList<>();
+ String filterGroup = getPolicyGroup(filter.entryFilter().principal());
+ logger.info("####GROUP: {}", filterGroup);
+ if (StringUtils.isNotBlank(filterGroup) && MapUtils.isNotEmpty(groupRangerACLs)) {
+ for (Map.Entry<String, Map<String, RangerResourceACLs.AccessResult>> entry : groupRangerACLs.entrySet()) {
+ String groupName = entry.getKey();
+ if (StringUtils.isNotBlank(filterGroup) && !"public".equals(groupName)
+ && !filterGroup.equals(groupName) && !"*".equals(groupName)) {
+ continue;
+ }
+ for (Map.Entry<String, RangerResourceACLs.AccessResult> privilege : entry.getValue().entrySet()) {
+ if (StringUtils.equals(RangerPolicyEngine.ADMIN_ACCESS, privilege.getKey())) {
+ continue;
+ }
+ ResourcePattern resourcePattern = getKafkaResourcePattern(filter);
+ AccessControlEntry accessControlEntry = getKafkaAccessControlEntry("GROUP:" + groupName, filter.entryFilter().host(), privilege);
+ ret.add(new AclBinding(resourcePattern, accessControlEntry));
+ }
+ }
+ } else if (StringUtils.isBlank(filterGroup)
+ && MapUtils.isNotEmpty(groupRangerACLs)
+ && aclType.equals(RangerKafkaUtils.AclType.LIST_ACCES)) {
+ for (Map.Entry<String, Map<String, RangerResourceACLs.AccessResult>> entry : groupRangerACLs.entrySet()) {
+ String groupName = entry.getKey();
+ for (Map.Entry<String, RangerResourceACLs.AccessResult> privilege : entry.getValue().entrySet()) {
+ if (StringUtils.equals(RangerPolicyEngine.ADMIN_ACCESS, privilege.getKey())) {
+ continue;
+ }
+ ResourcePattern resourcePattern = getKafkaResourcePattern(filter);
+ AccessControlEntry accessControlEntry = getKafkaAccessControlEntry("GROUP:" + groupName, filter.entryFilter().host(), privilege);
+ ret.add(new AclBinding(resourcePattern, accessControlEntry));
+ }
+ }
+ }
+ return ret;
+ }
+
+ public ResourcePattern getKafkaResourcePattern(AclBindingFilter filter) {
+ ResourcePatternFilter resourcePatternFilter = filter.patternFilter();
+ ResourceType resourceType = getResourceType(resourcePatternFilter);
+ String resourceName = getResourceName(resourcePatternFilter);
+ PatternType resourcePatternType = getResourcePatternType(resourcePatternFilter);
+ return new ResourcePattern(resourceType, resourceName, resourcePatternType);
+ }
+
+ public AccessControlEntry getKafkaAccessControlEntry(String userOrGroupName, String hostIP, Map.Entry<String, RangerResourceACLs.AccessResult> privilege) {
+ String principal = userOrGroupName;
+ String host = (hostIP != null) ? hostIP : ACCESS_TYPE_ANY;
+ AclOperation aclOperation = getKafkAclOperation(privilege.getKey());
+ AclPermissionType aclPermissionType = getKafkaAclPermissionType(privilege.getValue());
+ return new AccessControlEntry(principal, host, aclOperation, aclPermissionType);
+ }
+
+ public ResourceType getResourceType(ResourcePatternFilter resourcePatternFilter) {
+ return resourcePatternFilter.resourceType();
+ }
+
+ public String getResourceName(ResourcePatternFilter resourcePatternFilter) {
+ return resourcePatternFilter.name();
+ }
+
+ public PatternType getResourcePatternType(ResourcePatternFilter resourcePatternFilter) {
+ return resourcePatternFilter.patternType();
+ }
+
+ public AclOperation getKafkAclOperation(String rangerOperation) {
+ AclOperation ret = null;
+ switch (rangerOperation) {
+ case ACCESS_TYPE_READ:
+ ret = AclOperation.READ;
+ break;
+ case ACCESS_TYPE_WRITE:
+ ret = AclOperation.WRITE;
+ break;
+ case ACCESS_TYPE_CONFIGURE:
+ ret = AclOperation.DESCRIBE_CONFIGS;
+ break;
+ case ACCESS_TYPE_DESCRIBE:
+ ret = AclOperation.DESCRIBE;
+ break;
+ case ACCESS_TYPE_CLUSTER_ACTION:
+ ret = AclOperation.CLUSTER_ACTION;
+ break;
+ case ACCESS_TYPE_CREATE:
+ ret = AclOperation.CREATE;
+ break;
+ case ACCESS_TYPE_ALTER:
+ ret = AclOperation.ALTER;
+ break;
+ case ACCESS_TYPE_DELETE:
+ ret = AclOperation.DELETE;
+ break;
+ case ACCESS_TYPE_DESCRIBE_CONFIGS:
+ ret = AclOperation.DESCRIBE_CONFIGS;
+ break;
+ case ACCESS_TYPE_ALTER_CONFIGS:
+ ret = AclOperation.ALTER_CONFIGS;
+ break;
+ case ACCESS_TYPE_IDEMPOTENT_WRITE:
+ ret = AclOperation.IDEMPOTENT_WRITE;
+ break;
+ case ACCESS_TYPE_ADMIN:
+ case ACCESS_TYPE_KAFKA_ADMIN:
+ ret = AclOperation.ALL;
+ break;
+ }
+ return ret;
+ }
+
+ public AclPermissionType getKafkaAclPermissionType(RangerResourceACLs.AccessResult accessResult) {
+ return (accessResult.getResult() == 1) ? AclPermissionType.ALLOW : AclPermissionType.DENY;
+ }
+
+ public AclBinding getAclBindingForFilter(AclBindingFilter filter) {
+ ResourcePattern resourcePattern = getKafkaResourcePattern(filter);
+ AccessControlEntryFilter accessControlEntryFilter = filter.entryFilter();
+ AccessControlEntry accessControlEntry = new AccessControlEntry(accessControlEntryFilter.principal(), accessControlEntryFilter.host(), accessControlEntryFilter.operation(), accessControlEntryFilter.permissionType());
+ return new AclBinding(resourcePattern, accessControlEntry);
+ }
+}
diff --git a/security-admin/src/main/java/org/apache/ranger/rest/ServiceREST.java b/security-admin/src/main/java/org/apache/ranger/rest/ServiceREST.java
index 74468691c..db448ef5e 100644
--- a/security-admin/src/main/java/org/apache/ranger/rest/ServiceREST.java
+++ b/security-admin/src/main/java/org/apache/ranger/rest/ServiceREST.java
@@ -1169,7 +1169,10 @@ public RESTResponse grantAccess(@PathParam("serviceName") String serviceName, Gr
RangerService rangerService = svcStore.getServiceByName(serviceName);
String zoneName = getRangerAdminZoneName(serviceName, grantRequest);
- boolean isAdmin = bizUtil.isUserRangerAdmin(userName) || bizUtil.isUserServiceAdmin(rangerService, userName) || hasAdminAccess(serviceName, zoneName, userName, userGroups, resource, accessTypes);
+ boolean isAdmin = bizUtil.isUserRangerAdmin(userName)
+ || bizUtil.isUserServiceAdmin(rangerService, userName)
+ || bizUtil.isUserAllowed(rangerService, Allowed_User_List_For_Grant_Revoke)
+ || hasAdminAccess(serviceName, zoneName, userName, userGroups, resource, accessTypes);
if (!isAdmin) {
throw restErrorUtil.createGrantRevokeRESTException("User doesn't have necessary permission to grant access");
@@ -1288,7 +1291,10 @@ public RESTResponse secureGrantAccess(@PathParam("serviceName") String serviceNa
isAllowed = true;
}
} else {
- isAllowed = bizUtil.isUserRangerAdmin(userName) || bizUtil.isUserServiceAdmin(rangerService, userName) || hasAdminAccess(serviceName, zoneName, userName, userGroups, resource, accessTypes);
+ isAllowed = bizUtil.isUserRangerAdmin(userName)
+ || bizUtil.isUserServiceAdmin(rangerService, userName)
+ || bizUtil.isUserAllowed(rangerService, Allowed_User_List_For_Grant_Revoke)
+ || hasAdminAccess(serviceName, zoneName, userName, userGroups, resource, accessTypes);
}
if (isAllowed) {
@@ -1494,7 +1500,10 @@ public RESTResponse secureRevokeAccess(@PathParam("serviceName") String serviceN
isAllowed = true;
}
} else {
- isAllowed = bizUtil.isUserRangerAdmin(userName) || bizUtil.isUserServiceAdmin(rangerService, userName) || hasAdminAccess(serviceName, zoneName, userName, userGroups, resource, accessTypes);
+ isAllowed = bizUtil.isUserRangerAdmin(userName)
+ || bizUtil.isUserServiceAdmin(rangerService, userName)
+ || bizUtil.isUserAllowed(rangerService, Allowed_User_List_For_Grant_Revoke)
+ || hasAdminAccess(serviceName, zoneName, userName, userGroups, resource, accessTypes);
}
if (isAllowed) {
@@ -1535,7 +1544,7 @@ public RESTResponse secureRevokeAccess(@PathParam("serviceName") String serviceN
}
LOG.debug("<== ServiceREST.secureRevokeAccess({}, {}) :{}", serviceName, revokeRequest, ret);
-
+ LOG.info("<== ServiceREST.secureRevokeAccess({}, {}) :{}", serviceName, revokeRequest, ret);
return ret;
}
@@ -4355,29 +4364,33 @@ private void validateGrantor(String grantor) {
}
private void validateGrantees(Set<String> grantees) {
- for (String userName : grantees) {
- try {
- VXUser vxUser = xUserService.getXUserByUserName(userName);
+ if (CollectionUtils.isNotEmpty(grantees)) {
+ for (String userName : grantees) {
+ try {
+ VXUser vxUser = xUserService.getXUserByUserName(userName);
- if (vxUser == null) {
+ if (vxUser == null) {
+ throw restErrorUtil.createGrantRevokeRESTException("Grantee user " + userName + " doesn't exist");
+ }
+ } catch (Exception e) {
throw restErrorUtil.createGrantRevokeRESTException("Grantee user " + userName + " doesn't exist");
}
- } catch (Exception e) {
- throw restErrorUtil.createGrantRevokeRESTException("Grantee user " + userName + " doesn't exist");
}
}
}
private void validateGroups(Set<String> groups) {
- for (String groupName : groups) {
- try {
- VXGroup vxGroup = userMgr.getGroupByGroupName(groupName);
+ if (CollectionUtils.isNotEmpty(groups)) {
+ for (String groupName : groups) {
+ try {
+ VXGroup vxGroup = userMgr.getGroupByGroupName(groupName);
- if (vxGroup == null) {
+ if (vxGroup == null) {
+ throw restErrorUtil.createGrantRevokeRESTException("Grantee group " + groupName + " doesn't exist");
+ }
+ } catch (Exception e) {
throw restErrorUtil.createGrantRevokeRESTException("Grantee group " + groupName + " doesn't exist");
}
- } catch (Exception e) {
- throw restErrorUtil.createGrantRevokeRESTException("Grantee group " + groupName + " doesn't exist");
}
}
}