This is an automated email from the ASF dual-hosted git repository.
vyommani pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ranger.git
The following commit(s) were added to refs/heads/master by this push:
new 97c53ffeb RANGER-5771: Improve Ranger Kafka Plugin to Prevent OOM
During SRM St… (#1265)
97c53ffeb is described below
commit 97c53ffeb33512d348b10e63d76d4026493a778d
Author: Vyom Mani Tiwari <[email protected]>
AuthorDate: Wed Sep 30 12:03:15 2026 +0530
RANGER-5771: Improve Ranger Kafka Plugin to Prevent OOM During SRM St…
(#1265)
* RANGER-5771: Improve Ranger Kafka Plugin to Prevent OOM During SRM Startup
* RANGER-5771: addressed review comments
* RANGER-5771: addressed review comments
* RANGER-5771: addressed review comments
---
.../ranger/plugin/service/RangerBasePlugin.java | 36 ++-
.../service/RangerDefaultRequestProcessor.java | 85 +++++--
.../plugin/util/RangerAccessRequestUtil.java | 23 ++
.../ranger/plugin/util/RangerBatchEvalContext.java | 142 +++++++++++
.../TestRangerBasePluginBatchEvalContext.java | 260 +++++++++++++++++++++
.../service/TestRangerDefaultRequestProcessor.java | 236 +++++++++++++++++++
.../TestRangerKafkaAuthorizerSharedRoles.java | 223 ++++++++++++++++++
7 files changed, 984 insertions(+), 21 deletions(-)
diff --git
a/agents-common/src/main/java/org/apache/ranger/plugin/service/RangerBasePlugin.java
b/agents-common/src/main/java/org/apache/ranger/plugin/service/RangerBasePlugin.java
index 33bf33949..f11f98265 100644
---
a/agents-common/src/main/java/org/apache/ranger/plugin/service/RangerBasePlugin.java
+++
b/agents-common/src/main/java/org/apache/ranger/plugin/service/RangerBasePlugin.java
@@ -62,6 +62,8 @@
import org.apache.ranger.plugin.util.GrantRevokeRoleRequest;
import org.apache.ranger.plugin.util.PerfDataRecorder;
import org.apache.ranger.plugin.util.PolicyRefresher;
+import org.apache.ranger.plugin.util.RangerAccessRequestUtil;
+import org.apache.ranger.plugin.util.RangerBatchEvalContext;
import org.apache.ranger.plugin.util.RangerPolicyDeltaUtil;
import org.apache.ranger.plugin.util.RangerRoles;
import org.apache.ranger.plugin.util.RangerRolesUtil;
@@ -712,11 +714,27 @@ public Collection<RangerAccessResult>
isAccessAllowed(Collection<RangerAccessReq
refreshPoliciesAndTags();
}
- Collection<RangerAccessResult> ret = null;
- RangerPolicyEngine policyEngine = this.policyEngine;
+ Collection<RangerAccessResult> ret = null;
+ RangerPolicyEngine policyEngine = this.policyEngine;
+ RangerBatchEvalContext batchEvalContext = null;
if (policyEngine != null) {
- ret = policyEngine.evaluatePolicies(requests,
RangerPolicy.POLICY_TYPE_ACCESS, null);
+ if (CollectionUtils.isNotEmpty(requests)) {
+ batchEvalContext = new RangerBatchEvalContext();
+
+ for (RangerAccessRequest request : requests) {
+ if (request == null) {
+ continue;
+ }
+
RangerAccessRequestUtil.setBatchEvalContext(request.getContext(),
batchEvalContext);
+ }
+ }
+
+ try {
+ ret = policyEngine.evaluatePolicies(requests,
RangerPolicy.POLICY_TYPE_ACCESS, null);
+ } finally {
+ clearBatchEvalContext(requests, batchEvalContext);
+ }
}
if (CollectionUtils.isNotEmpty(ret)) {
@@ -752,6 +770,18 @@ public Collection<RangerAccessResult>
isAccessAllowed(Collection<RangerAccessReq
return ret;
}
+ private static void clearBatchEvalContext(Collection<RangerAccessRequest>
requests, RangerBatchEvalContext batchEvalContext) {
+ if (batchEvalContext == null || CollectionUtils.isEmpty(requests)) {
+ return;
+ }
+
+ for (RangerAccessRequest request : requests) {
+ if (request != null) {
+
RangerAccessRequestUtil.removeBatchEvalContext(request.getContext());
+ }
+ }
+ }
+
public RangerAccessResult evalDataMaskPolicies(RangerAccessRequest
request, RangerAccessResultProcessor resultProcessor) {
if (this.synchronousPolicyRefresh) {
refreshPoliciesAndTags();
diff --git
a/agents-common/src/main/java/org/apache/ranger/plugin/service/RangerDefaultRequestProcessor.java
b/agents-common/src/main/java/org/apache/ranger/plugin/service/RangerDefaultRequestProcessor.java
index 6661e34e1..05c040f1f 100644
---
a/agents-common/src/main/java/org/apache/ranger/plugin/service/RangerDefaultRequestProcessor.java
+++
b/agents-common/src/main/java/org/apache/ranger/plugin/service/RangerDefaultRequestProcessor.java
@@ -31,6 +31,7 @@
import org.apache.ranger.plugin.policyengine.RangerMutableResource;
import org.apache.ranger.plugin.policyengine.RangerPluginContext;
import org.apache.ranger.plugin.util.RangerAccessRequestUtil;
+import org.apache.ranger.plugin.util.RangerBatchEvalContext;
import org.apache.ranger.plugin.util.RangerCommonConstants;
import org.apache.ranger.plugin.util.RangerPerfTracer;
import org.apache.ranger.plugin.util.RangerUserStoreUtil;
@@ -84,8 +85,11 @@ public void preProcess(RangerAccessRequest request) {
setResourceServiceDef(request);
- RangerPluginContext pluginContext =
policyEngine.getPluginContext();
- RangerAccessRequestImpl reqImpl = null;
+ RangerPluginContext pluginContext =
policyEngine.getPluginContext();
+ RangerAccessRequestImpl reqImpl = null;
+ RangerBatchEvalContext batchEvalContext =
RangerAccessRequestUtil.getBatchEvalContext(request.getContext());
+ String originalUser = request.getUser();
+ Set<String> originalGroups = request.getUserGroups();
if (request instanceof RangerAccessRequestImpl) {
reqImpl = (RangerAccessRequestImpl) request;
@@ -103,20 +107,7 @@ public void preProcess(RangerAccessRequest request) {
reqImpl.setClusterType(pluginContext.getClusterType());
}
- RangerPluginConfig config =
policyEngine.getPluginContext().getConfig();
-
- boolean isNameTransformationSupported =
config.getBoolean(config.getPropertyPrefix() +
RangerCommonConstants.PLUGIN_CONFIG_SUFFIX_NAME_TRANSFORMATION, false);
-
- LOG.debug("isNameTransformationSupported = {}",
isNameTransformationSupported);
-
- if (isNameTransformationSupported) {
- reqImpl.setUser(getTransformedUser(policyEngine, request));
- reqImpl.setUserGroups(getTransformedGroups(policyEngine,
request));
- }
-
- convertEmailToUsername(reqImpl);
-
- updateUserGroups(reqImpl);
+ normalizeUserAndGroups(reqImpl, pluginContext,
batchEvalContext, originalUser, originalGroups);
}
}
@@ -130,9 +121,20 @@ public void preProcess(RangerAccessRequest request) {
Set<String> roles = request.getUserRoles();
if (pluginContext != null && CollectionUtils.isEmpty(roles)) {
- roles =
pluginContext.getAuthContext().getRolesForUserAndGroups(request.getUser(),
request.getUserGroups());
+ if (batchEvalContext != null) {
+ roles = batchEvalContext.getMappedUserRoles(originalUser,
originalGroups);
+
+ if (roles == null) {
+ roles =
pluginContext.getAuthContext().getRolesForUserAndGroups(request.getUser(),
request.getUserGroups());
+ roles = readOnly(roles);
- if (reqImpl != null && roles != null && !roles.isEmpty()) {
+ batchEvalContext.setUserRolesMapping(originalUser,
originalGroups, roles);
+ }
+ } else {
+ roles =
pluginContext.getAuthContext().getRolesForUserAndGroups(request.getUser(),
request.getUserGroups());
+ }
+
+ if (reqImpl != null && CollectionUtils.isNotEmpty(roles)) {
reqImpl.setUserRoles(roles);
}
}
@@ -173,6 +175,53 @@ public void enrich(RangerAccessRequest request) {
}
}
+ private void normalizeUserAndGroups(RangerAccessRequestImpl reqImpl,
RangerPluginContext pluginContext, RangerBatchEvalContext batchEvalContext,
String originalUser, Set<String> originalGroups) {
+ if (batchEvalContext != null) {
+ String mappedUser =
batchEvalContext.getMappedUserName(originalUser);
+ Set<String> mappedUserGroups =
batchEvalContext.getMappedUserGroups(originalUser, originalGroups);
+ // A mapped user can be null, so a null user is a hit when the key
is present.
+ boolean userMissing = mappedUser == null &&
!batchEvalContext.hasMappingForUserName(originalUser);
+
+ if (userMissing || mappedUserGroups == null) {
+ mapUserAndGroups(reqImpl, pluginContext);
+
+ Set<String> groups = copyReadOnly(reqImpl.getUserGroups());
+
+ batchEvalContext.setUserNameMapping(originalUser,
reqImpl.getUser());
+ batchEvalContext.setUserGroupsMapping(originalUser,
originalGroups, groups);
+ reqImpl.setUserGroups(groups);
+ } else {
+ reqImpl.setUser(mappedUser);
+ reqImpl.setUserGroups(mappedUserGroups);
+ }
+ } else {
+ mapUserAndGroups(reqImpl, pluginContext);
+ }
+ }
+
+ private void mapUserAndGroups(RangerAccessRequestImpl req,
RangerPluginContext pluginContext) {
+ RangerPluginConfig config = pluginContext.getConfig();
+ boolean needsTransform =
config.getBoolean(config.getPropertyPrefix() +
RangerCommonConstants.PLUGIN_CONFIG_SUFFIX_NAME_TRANSFORMATION, false);
+
+ LOG.debug("isNameTransformationSupported = {}", needsTransform);
+
+ if (needsTransform) {
+ req.setUser(getTransformedUser(policyEngine, req));
+ req.setUserGroups(getTransformedGroups(policyEngine, req));
+ }
+
+ convertEmailToUsername(req);
+ updateUserGroups(req);
+ }
+
+ private static Set<String> readOnly(Set<String> values) {
+ return values == null || values.isEmpty() ? Collections.emptySet() :
Collections.unmodifiableSet(values);
+ }
+
+ private static Set<String> copyReadOnly(Set<String> values) {
+ return values == null || values.isEmpty() ? Collections.emptySet() :
Collections.unmodifiableSet(new HashSet<>(values));
+ }
+
private String getTransformedUser(PolicyEngine policyEngine,
RangerAccessRequest request) {
RangerAuthContext authContext =
policyEngine.getPluginContext().getAuthContext();
boolean toLowerCase =
authContext.getUserNameCaseConversion() ==
UgsyncCommonConstants.CaseConversion.TO_LOWER;
diff --git
a/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerAccessRequestUtil.java
b/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerAccessRequestUtil.java
index 52a643ead..8e39c5c50 100644
---
a/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerAccessRequestUtil.java
+++
b/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerAccessRequestUtil.java
@@ -57,6 +57,7 @@ public class RangerAccessRequestUtil {
public static final String KEY_CONTEXT_REQUEST =
"_REQUEST";
public static final String KEY_CONTEXT_GDS_RESULT =
"_GDS_RESULT";
public static final String KEY_CONTEXT_IS_REQUEST_PREPROCESSED =
"ISREQUESTPREPROCESSED";
+ public static final String KEY_CONTEXT_BATCH_EVAL_CONTEXT =
"BATCH_EVAL_CONTEXT";
public static final String KEY_CONTEXT_RESOURCE_ZONE_NAMES =
"RESOURCE_ZONE_NAMES";
public static final String KEY_CONTEXT_ACL_ENFORCER =
"_ACL_ENFORCER";
@@ -239,6 +240,28 @@ public static boolean getIsRequestPreprocessed(Map<String,
Object> context) {
return value != null && value;
}
+ public static void setBatchEvalContext(Map<String, Object> context,
RangerBatchEvalContext batchEvalContext) {
+ if (context != null && batchEvalContext != null) {
+ context.put(KEY_CONTEXT_BATCH_EVAL_CONTEXT, batchEvalContext);
+ }
+ }
+
+ public static RangerBatchEvalContext getBatchEvalContext(Map<String,
Object> context) {
+ if (context == null) {
+ return null;
+ }
+
+ Object value = context.get(KEY_CONTEXT_BATCH_EVAL_CONTEXT);
+
+ return value instanceof RangerBatchEvalContext ?
(RangerBatchEvalContext) value : null;
+ }
+
+ public static void removeBatchEvalContext(Map<String, Object> context) {
+ if (context != null) {
+ context.remove(KEY_CONTEXT_BATCH_EVAL_CONTEXT);
+ }
+ }
+
public static void setAllRequestedAccessTypes(Map<String, Object> context,
Set<String> accessTypes) {
context.put(KEY_CONTEXT_ALL_ACCESSTYPES, accessTypes);
}
diff --git
a/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerBatchEvalContext.java
b/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerBatchEvalContext.java
new file mode 100644
index 000000000..12d5f3195
--- /dev/null
+++
b/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerBatchEvalContext.java
@@ -0,0 +1,142 @@
+/*
+ * 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.plugin.util;
+
+import org.apache.commons.collections.CollectionUtils;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * Not thread-safe. One instance per evaluation call; do not mutate a groups
set during that call, because the hot key matches the same set instance.
+ */
+public class RangerBatchEvalContext {
+ private final Map<String, String> userNameMappings = new HashMap<>();
+ private final Map<IdentityKey, Set<String>> userGroupsMappings = new
HashMap<>();
+ private final Map<IdentityKey, Set<String>> userRolesMappings = new
HashMap<>();
+
+ private IdentityKey hotKey;
+ private String hotUser;
+ private Set<String> hotGroupsRef;
+
+ public boolean hasMappingForUserName(String user) {
+ return userNameMappings.containsKey(user);
+ }
+
+ public String getMappedUserName(String user) {
+ return userNameMappings.get(user);
+ }
+
+ public void setUserNameMapping(String user, String normalizedUser) {
+ userNameMappings.put(user, normalizedUser);
+ }
+
+ public boolean hasMappingForUserGroups(String user, Set<String> groups) {
+ return userGroupsMappings.containsKey(keyFor(user, groups));
+ }
+
+ public Set<String> getMappedUserGroups(String user, Set<String> groups) {
+ return userGroupsMappings.get(keyFor(user, groups));
+ }
+
+ public void setUserGroupsMapping(String user, Set<String> groups,
Set<String> normalizedGroups) {
+ userGroupsMappings.put(storedKey(user, groups), normalizedGroups);
+ }
+
+ public boolean hasMappingForUserRoles(String user, Set<String> groups) {
+ return userRolesMappings.containsKey(keyFor(user, groups));
+ }
+
+ public Set<String> getMappedUserRoles(String user, Set<String> groups) {
+ return userRolesMappings.get(keyFor(user, groups));
+ }
+
+ public void setUserRolesMapping(String user, Set<String> groups,
Set<String> roles) {
+ userRolesMappings.put(storedKey(user, groups), roles);
+ }
+
+ private IdentityKey keyFor(String user, Set<String> groups) {
+ if (hotKey != null && Objects.equals(user, hotUser) && groups ==
hotGroupsRef) {
+ return hotKey;
+ }
+
+ IdentityKey key = new IdentityKey(user, groups, false);
+
+ hotKey = key;
+ hotUser = user;
+ hotGroupsRef = groups;
+
+ return key;
+ }
+
+ private IdentityKey storedKey(String user, Set<String> groups) {
+ IdentityKey key = new IdentityKey(user, groups, true);
+
+ hotKey = key;
+ hotUser = user;
+ hotGroupsRef = groups;
+
+ return key;
+ }
+
+ private static final class IdentityKey {
+ private final String user;
+ private final Set<String> groups;
+ private final int hash;
+
+ private IdentityKey(String user, Set<String> groups, boolean copy) {
+ this.user = user;
+ this.groups = copy || CollectionUtils.isEmpty(groups) ?
copyGroups(groups) : groups;
+ this.hash = Objects.hash(user, this.groups);
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (this == other) {
+ return true;
+ }
+
+ if (!(other instanceof IdentityKey)) {
+ return false;
+ }
+
+ IdentityKey that = (IdentityKey) other;
+
+ return Objects.equals(user, that.user) &&
groups.equals(that.groups);
+ }
+
+ @Override
+ public int hashCode() {
+ return hash;
+ }
+
+ private static Set<String> copyGroups(Set<String> groups) {
+ if (CollectionUtils.isEmpty(groups)) {
+ return Collections.emptySet();
+ }
+
+ return Collections.unmodifiableSet(new HashSet<>(groups));
+ }
+ }
+}
diff --git
a/agents-common/src/test/java/org/apache/ranger/plugin/service/TestRangerBasePluginBatchEvalContext.java
b/agents-common/src/test/java/org/apache/ranger/plugin/service/TestRangerBasePluginBatchEvalContext.java
new file mode 100644
index 000000000..33557d6c2
--- /dev/null
+++
b/agents-common/src/test/java/org/apache/ranger/plugin/service/TestRangerBasePluginBatchEvalContext.java
@@ -0,0 +1,260 @@
+/*
+ * 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.plugin.service;
+
+import org.apache.ranger.authorization.hadoop.config.RangerPluginConfig;
+import org.apache.ranger.plugin.model.RangerPolicy;
+import org.apache.ranger.plugin.model.RangerRole;
+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.RangerPolicyEngineOptions;
+import org.apache.ranger.plugin.store.EmbeddedServiceDefsUtil;
+import org.apache.ranger.plugin.util.RangerAccessRequestUtil;
+import org.apache.ranger.plugin.util.RangerBatchEvalContext;
+import org.apache.ranger.plugin.util.RangerRoles;
+import org.apache.ranger.plugin.util.ServicePolicies;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestRangerBasePluginBatchEvalContext {
+ @Test
+ void collectionAccessAttachesOneBatchEvalContextAndRemovesIt() throws
Exception {
+ RangerBasePlugin plugin = new RangerBasePlugin("hbase", "hbase");
+ RangerPolicyEngine policyEngine =
Mockito.mock(RangerPolicyEngine.class);
+ setPolicyEngine(plugin, policyEngine);
+
+ RangerAccessRequestImpl first = new RangerAccessRequestImpl();
+ RangerAccessRequestImpl second = new RangerAccessRequestImpl();
+ RangerAccessRequest nullContext =
Mockito.mock(RangerAccessRequest.class);
+ Mockito.when(nullContext.getContext()).thenReturn(null);
+
+ Mockito.when(policyEngine.evaluatePolicies(Mockito.anyCollection(),
Mockito.eq(RangerPolicy.POLICY_TYPE_ACCESS),
Mockito.isNull())).thenAnswer(invocation -> {
+ Collection<RangerAccessRequest> seen = invocation.getArgument(0);
+ RangerBatchEvalContext shared = null;
+
+ for (RangerAccessRequest request : seen) {
+ if (request == null || request.getContext() == null) {
+ continue;
+ }
+
+ RangerBatchEvalContext batchEvalContext =
RangerAccessRequestUtil.getBatchEvalContext(request.getContext());
+
+ assertNotNull(batchEvalContext);
+
+ if (shared == null) {
+ shared = batchEvalContext;
+ } else {
+ assertSame(shared, batchEvalContext);
+ }
+ }
+
+ assertNotNull(shared);
+
+ return Collections.emptyList();
+ });
+
+ plugin.isAccessAllowed(Arrays.asList(first, null, second,
nullContext), null);
+
+
assertNull(RangerAccessRequestUtil.getBatchEvalContext(first.getContext()));
+
assertNull(RangerAccessRequestUtil.getBatchEvalContext(second.getContext()));
+ }
+
+ @Test
+ void collectionAccessRemovesBatchEvalContextWhenEvaluationThrows() throws
Exception {
+ RangerBasePlugin plugin = new RangerBasePlugin("hbase", "hbase");
+ RangerPolicyEngine policyEngine =
Mockito.mock(RangerPolicyEngine.class);
+ setPolicyEngine(plugin, policyEngine);
+
+ RangerAccessRequestImpl request = new RangerAccessRequestImpl();
+ Mockito.when(policyEngine.evaluatePolicies(Mockito.anyCollection(),
Mockito.eq(RangerPolicy.POLICY_TYPE_ACCESS), Mockito.isNull())).thenThrow(new
IllegalStateException("evaluation failed"));
+
+ assertThrows(IllegalStateException.class, () ->
plugin.isAccessAllowed(Collections.singletonList(request), null));
+
assertNull(RangerAccessRequestUtil.getBatchEvalContext(request.getContext()));
+ }
+
+ @Test
+ void singleAccessDoesNotAttachBatchEvalContext() throws Exception {
+ RangerBasePlugin plugin = new RangerBasePlugin("hbase", "hbase");
+ RangerPolicyEngine policyEngine =
Mockito.mock(RangerPolicyEngine.class);
+ setPolicyEngine(plugin, policyEngine);
+
+ RangerAccessRequestImpl request = new RangerAccessRequestImpl();
+
+ plugin.isAccessAllowed(request, null);
+
+
assertNull(RangerAccessRequestUtil.getBatchEvalContext(request.getContext()));
+ }
+
+ @Test
+ void bulkEvaluationMatchesSingleRequestEvaluation() throws Exception {
+ RangerBasePlugin plugin = newPlugin();
+ plugin.setPolicies(servicePolicies());
+ plugin.setRoles(roles());
+
+ List<RangerAccessRequest> bulkRequests = new ArrayList<>(Arrays.asList(
+ request("alice"),
+ request("alice"),
+ request("bob"),
+ request("carol"),
+ request("carol"),
+ request("dave"),
+ request(null)));
+ List<RangerAccessRequest> singleRequests = new
ArrayList<>(Arrays.asList(
+ request("alice"),
+ request("alice"),
+ request("bob"),
+ request("carol"),
+ request("carol"),
+ request("dave"),
+ request(null)));
+
+ Collection<RangerAccessResult> bulkResults =
plugin.isAccessAllowed(bulkRequests, null);
+ List<Boolean> bulkAllowed = new ArrayList<>();
+
+ for (RangerAccessResult result : bulkResults) {
+ bulkAllowed.add(result.getIsAllowed());
+ }
+
+ List<Boolean> singleAllowed = new ArrayList<>();
+
+ for (RangerAccessRequest singleRequest : singleRequests) {
+
singleAllowed.add(plugin.isAccessAllowed(singleRequest).getIsAllowed());
+ }
+
+ assertEquals(Arrays.asList(true, true, false, true, true, false,
false), singleAllowed);
+ assertEquals(singleAllowed, bulkAllowed);
+ assertSame(bulkRequests.get(3).getUserRoles(),
bulkRequests.get(4).getUserRoles());
+ assertTrue(bulkRequests.get(3).getUserRoles().contains("readers"));
+ assertFalse(bulkRequests.get(0).getUserRoles().contains("readers"));
+
+ for (RangerAccessRequest bulkRequest : bulkRequests) {
+
assertNull(RangerAccessRequestUtil.getBatchEvalContext(bulkRequest.getContext()));
+ }
+ }
+
+ private static void setPolicyEngine(RangerBasePlugin plugin,
RangerPolicyEngine policyEngine) throws Exception {
+ Field field = RangerBasePlugin.class.getDeclaredField("policyEngine");
+
+ field.setAccessible(true);
+ field.set(plugin, policyEngine);
+ }
+
+ private static RangerBasePlugin newPlugin() {
+ RangerPolicyEngineOptions options = new RangerPolicyEngineOptions();
+ options.disablePolicyRefresher = true;
+ options.disableTagRetriever = true;
+ options.disableUserStoreRetriever = true;
+ options.disableGdsInfoRetriever = true;
+
+ return new RangerBasePlugin(new RangerPluginConfig("hive", "svc",
"app", null, null, options));
+ }
+
+ private static ServicePolicies servicePolicies() throws Exception {
+ RangerPolicy.RangerPolicyItem allowAlice =
policyItem(Collections.singletonList("alice"), Collections.emptyList());
+ RangerPolicy.RangerPolicyItem allowReaders =
policyItem(Collections.emptyList(), Collections.singletonList("readers"));
+ RangerPolicy.RangerPolicyItem denyBob =
policyItem(Collections.singletonList("bob"), Collections.emptyList());
+
+ RangerPolicy policy = new RangerPolicy();
+ policy.setId(1L);
+ policy.setName("db1-select");
+ policy.setService("svc");
+ policy.setIsEnabled(true);
+ policy.setResources(resources());
+ policy.setPolicyItems(Arrays.asList(allowAlice, allowReaders));
+ policy.setDenyPolicyItems(Collections.singletonList(denyBob));
+
+ ServicePolicies policies = new ServicePolicies();
+ policies.setServiceName("svc");
+ policies.setPolicyVersion(1L);
+
policies.setServiceDef(EmbeddedServiceDefsUtil.instance().getEmbeddedServiceDef(EmbeddedServiceDefsUtil.EMBEDDED_SERVICEDEF_HIVE_NAME));
+ policies.setPolicies(Collections.singletonList(policy));
+
+ return policies;
+ }
+
+ private static Map<String, RangerPolicy.RangerPolicyResource> resources() {
+ Map<String, RangerPolicy.RangerPolicyResource> resources = new
HashMap<>();
+ resources.put("database", new
RangerPolicy.RangerPolicyResource("db1"));
+ resources.put("table", new RangerPolicy.RangerPolicyResource("tbl1"));
+ resources.put("column", new RangerPolicy.RangerPolicyResource("col1"));
+
+ return resources;
+ }
+
+ private static RangerPolicy.RangerPolicyItem policyItem(List<String>
users, List<String> roles) {
+ RangerPolicy.RangerPolicyItemAccess access = new
RangerPolicy.RangerPolicyItemAccess();
+ access.setType("select");
+ access.setIsAllowed(true);
+
+ RangerPolicy.RangerPolicyItem item = new
RangerPolicy.RangerPolicyItem();
+ item.setUsers(users);
+ item.setRoles(roles);
+ item.setAccesses(Collections.singletonList(access));
+
+ return item;
+ }
+
+ private static RangerRoles roles() {
+ RangerRole readers = new RangerRole();
+ readers.setName("readers");
+ readers.setUsers(Collections.singletonList(new
RangerRole.RoleMember("carol", false)));
+
+ RangerRoles roles = new RangerRoles();
+ roles.setRangerRoles(Collections.singleton(readers));
+
+ return roles;
+ }
+
+ private static RangerAccessRequestImpl request(String user) {
+ RangerAccessResourceImpl resource = new RangerAccessResourceImpl();
+ resource.setValue("database", "db1");
+ resource.setValue("table", "tbl1");
+ resource.setValue("column", "col1");
+
+ RangerAccessRequestImpl request = new RangerAccessRequestImpl();
+ request.setResource(resource);
+ request.setAccessType("select");
+ request.setUser(user);
+ request.setUserGroups(new HashSet<>(Collections.singleton("public")));
+
+ return request;
+ }
+}
diff --git
a/agents-common/src/test/java/org/apache/ranger/plugin/service/TestRangerDefaultRequestProcessor.java
b/agents-common/src/test/java/org/apache/ranger/plugin/service/TestRangerDefaultRequestProcessor.java
index 9b4411722..18c727532 100644
---
a/agents-common/src/test/java/org/apache/ranger/plugin/service/TestRangerDefaultRequestProcessor.java
+++
b/agents-common/src/test/java/org/apache/ranger/plugin/service/TestRangerDefaultRequestProcessor.java
@@ -29,6 +29,7 @@
import org.apache.ranger.plugin.policyengine.RangerPluginContext;
import org.apache.ranger.plugin.policyengine.RangerPolicyEngineOptions;
import org.apache.ranger.plugin.util.RangerAccessRequestUtil;
+import org.apache.ranger.plugin.util.RangerBatchEvalContext;
import org.apache.ranger.plugin.util.RangerCommonConstants;
import org.apache.ranger.plugin.util.RangerRoles;
import org.apache.ranger.plugin.util.RangerUserStore;
@@ -40,6 +41,7 @@
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -48,7 +50,11 @@
import java.util.Set;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
@@ -332,4 +338,234 @@ public void
test4_NameAndGroupCaseConversionWithoutMapper() {
assertTrue(req.getUserGroups().contains("GRP"));
assertTrue(req.getUserGroups().contains("GRP2"));
}
+
+ @Test
+ public void test5_BatchEvalContextSharesNormalizedUserGroupsAndRoles() {
+ RangerUserStore userStore = new RangerUserStore();
+ Map<String, Map<String, String>> userAttrs = new HashMap<>();
+ Map<String, String> attrs = new HashMap<>();
+ attrs.put(RangerCommonConstants.SCRIPT_FIELD__EMAIL_ADDRESS,
"[email protected]");
+ userAttrs.put("u2", attrs);
+ userStore.setUserAttrMapping(userAttrs);
+ Map<String, Set<String>> userGroups = new HashMap<>();
+ userGroups.put("u2", new HashSet<>(Collections.singletonList("rg9")));
+ userStore.setUserGroupMapping(userGroups);
+
+ RangerRole role = new RangerRole();
+ role.setName("roleX");
+ role.setUsers(Collections.singletonList(new
RangerRole.RoleMember("u2", false)));
+ ProcessorFixture fixture = newFixture(true, true, true, true,
"ranger.plugin.dummy", userStore, roles(role));
+
+ RangerAccessRequestImpl first = newRequest(fixture.serviceDef,
"[email protected]", Collections.singleton("g1"));
+ RangerAccessRequestImpl second = newRequest(fixture.serviceDef,
"[email protected]", Collections.singleton("g1"));
+ attach(fixture.batchEvalContext, first, second);
+
+ fixture.processor.preProcess(first);
+ fixture.processor.preProcess(second);
+
+ assertEquals("u2", first.getUser());
+ assertEquals("u2", second.getUser());
+ assertEquals("clA", first.getClusterName());
+ assertEquals("clA", second.getClusterName());
+ assertSame(first.getUserGroups(), second.getUserGroups());
+ assertTrue(first.getUserGroups().contains("rg9"));
+ assertSame(first.getUserRoles(), second.getUserRoles());
+ assertTrue(first.getUserRoles().contains("roleX"));
+ assertSame(first.getUserRoles(),
RangerAccessRequestUtil.getCurrentUserRolesFromContext(second.getContext()));
+ Mockito.verify(fixture.authContext,
Mockito.times(1)).getRolesForUserAndGroups(Mockito.eq("u2"), Mockito.any());
+ }
+
+ @Test
+ public void test6_BatchEvalContextKeepsSeparateRolesForDifferentGroups() {
+ RangerRole readers = new RangerRole();
+ readers.setName("readers");
+ readers.setGroups(Collections.singletonList(new
RangerRole.RoleMember("g1", false)));
+ RangerRole writers = new RangerRole();
+ writers.setName("writers");
+ writers.setGroups(Collections.singletonList(new
RangerRole.RoleMember("g2", false)));
+ ProcessorFixture fixture = newFixture(false, false, false, false,
"ranger.plugin.svc", new RangerUserStore(), roles(readers, writers));
+
+ RangerAccessRequestImpl readersRequest =
newRequest(fixture.serviceDef, "alice", Collections.singleton("g1"));
+ RangerAccessRequestImpl writersRequest =
newRequest(fixture.serviceDef, "alice", Collections.singleton("g2"));
+ RangerAccessRequestImpl readersAgain = newRequest(fixture.serviceDef,
"alice", Collections.singleton("g1"));
+ attach(fixture.batchEvalContext, readersRequest, writersRequest,
readersAgain);
+
+ fixture.processor.preProcess(readersRequest);
+ fixture.processor.preProcess(writersRequest);
+ fixture.processor.preProcess(readersAgain);
+
+ assertSame(readersRequest.getUserRoles(), readersAgain.getUserRoles());
+ assertNotSame(readersRequest.getUserRoles(),
writersRequest.getUserRoles());
+ assertSame(readersRequest.getUserGroups(),
readersAgain.getUserGroups());
+ assertTrue(readersRequest.getUserRoles().contains("readers"));
+ assertFalse(readersRequest.getUserRoles().contains("writers"));
+ assertTrue(writersRequest.getUserRoles().contains("writers"));
+ assertFalse(writersRequest.getUserRoles().contains("readers"));
+ Mockito.verify(fixture.authContext,
Mockito.times(2)).getRolesForUserAndGroups(Mockito.eq("alice"), Mockito.any());
+ }
+
+ @Test
+ public void test7_BatchEvalContextDoesNotReplacePresetRoles() {
+ RangerRole role = new RangerRole();
+ role.setName("roleX");
+ role.setUsers(Collections.singletonList(new
RangerRole.RoleMember("alice", false)));
+ ProcessorFixture fixture = newFixture(false, false, false, false,
"ranger.plugin.svc", new RangerUserStore(), roles(role));
+
+ Set<String> presetRoles = new
HashSet<>(Collections.singleton("preset"));
+ RangerAccessRequestImpl preset = newRequest(fixture.serviceDef,
"alice", Collections.singleton("g1"));
+ RangerAccessRequestImpl resolved = newRequest(fixture.serviceDef,
"alice", Collections.singleton("g1"));
+ RangerAccessRequestImpl resolvedAgain = newRequest(fixture.serviceDef,
"alice", Collections.singleton("g1"));
+ preset.setUserRoles(presetRoles);
+ attach(fixture.batchEvalContext, preset, resolved, resolvedAgain);
+
+ fixture.processor.preProcess(preset);
+ fixture.processor.preProcess(resolved);
+ fixture.processor.preProcess(resolvedAgain);
+
+ assertSame(presetRoles, preset.getUserRoles());
+ assertSame(resolved.getUserRoles(), resolvedAgain.getUserRoles());
+ assertTrue(resolved.getUserRoles().contains("roleX"));
+ assertFalse(resolved.getUserRoles().contains("preset"));
+ Mockito.verify(fixture.authContext,
Mockito.times(1)).getRolesForUserAndGroups(Mockito.eq("alice"), Mockito.any());
+ }
+
+ @Test
+ public void test8_BatchEvalContextDoesNotReplaceEmptyRoleSet() {
+ ProcessorFixture fixture = newFixture(false, false, false, false,
"ranger.plugin.svc", new RangerUserStore(), new RangerRoles());
+
+ RangerAccessRequestImpl first = newRequest(fixture.serviceDef,
"alice", Collections.singleton("g1"));
+ RangerAccessRequestImpl second = newRequest(fixture.serviceDef,
"alice", Collections.singleton("g1"));
+ Set<String> firstRoles = first.getUserRoles();
+ Set<String> secondRoles = second.getUserRoles();
+ attach(fixture.batchEvalContext, first, second);
+
+ fixture.processor.preProcess(first);
+ fixture.processor.preProcess(second);
+
+ assertTrue(first.getUserRoles().isEmpty());
+ assertSame(firstRoles, first.getUserRoles());
+ assertSame(secondRoles, second.getUserRoles());
+ assertNotSame(first.getUserRoles(), second.getUserRoles());
+ Mockito.verify(fixture.authContext,
Mockito.times(1)).getRolesForUserAndGroups(Mockito.eq("alice"), Mockito.any());
+ }
+
+ @Test
+ public void test9_BatchEvalContextKeepsSeparateRolesForDifferentUsers() {
+ RangerRole aliceRole = new RangerRole();
+ aliceRole.setName("aliceRole");
+ aliceRole.setUsers(Collections.singletonList(new
RangerRole.RoleMember("alice", false)));
+ RangerRole bobRole = new RangerRole();
+ bobRole.setName("bobRole");
+ bobRole.setUsers(Collections.singletonList(new
RangerRole.RoleMember("bob", false)));
+ ProcessorFixture fixture = newFixture(false, false, false, false,
"ranger.plugin.svc", new RangerUserStore(), roles(aliceRole, bobRole));
+
+ RangerAccessRequestImpl alice = newRequest(fixture.serviceDef,
"alice", Collections.singleton("g1"));
+ RangerAccessRequestImpl aliceAgain = newRequest(fixture.serviceDef,
"alice", Collections.singleton("g1"));
+ RangerAccessRequestImpl bob = newRequest(fixture.serviceDef, "bob",
Collections.singleton("g1"));
+ RangerAccessRequestImpl bobAgain = newRequest(fixture.serviceDef,
"bob", Collections.singleton("g1"));
+ attach(fixture.batchEvalContext, alice, bob, aliceAgain, bobAgain);
+
+ fixture.processor.preProcess(alice);
+ fixture.processor.preProcess(bob);
+ fixture.processor.preProcess(aliceAgain);
+ fixture.processor.preProcess(bobAgain);
+
+ assertSame(alice.getUserRoles(), aliceAgain.getUserRoles());
+ assertSame(bob.getUserRoles(), bobAgain.getUserRoles());
+ assertNotSame(alice.getUserRoles(), bob.getUserRoles());
+ assertTrue(alice.getUserRoles().contains("aliceRole"));
+ assertFalse(alice.getUserRoles().contains("bobRole"));
+ assertTrue(bob.getUserRoles().contains("bobRole"));
+ Mockito.verify(fixture.authContext,
Mockito.times(1)).getRolesForUserAndGroups(Mockito.eq("alice"), Mockito.any());
+ Mockito.verify(fixture.authContext,
Mockito.times(1)).getRolesForUserAndGroups(Mockito.eq("bob"), Mockito.any());
+ }
+
+ @Test
+ public void test10_BatchEvalContextAllowsNullUser() {
+ RangerRole groupRole = new RangerRole();
+ groupRole.setName("fromGroup");
+ groupRole.setGroups(Collections.singletonList(new
RangerRole.RoleMember("g1", false)));
+ ProcessorFixture fixture = newFixture(false, false, false, false,
"ranger.plugin.svc", new RangerUserStore(), roles(groupRole));
+
+ RangerAccessRequestImpl first = newRequest(fixture.serviceDef, null,
Collections.singleton("g1"));
+ RangerAccessRequestImpl second = newRequest(fixture.serviceDef, null,
Collections.singleton("g1"));
+ RangerAccessRequestImpl otherGroups = newRequest(fixture.serviceDef,
null, Collections.singleton("g2"));
+ attach(fixture.batchEvalContext, first, second, otherGroups);
+
+ fixture.processor.preProcess(first);
+ fixture.processor.preProcess(second);
+ fixture.processor.preProcess(otherGroups);
+
+ assertNull(first.getUser());
+ assertSame(first.getUserRoles(), second.getUserRoles());
+ assertTrue(first.getUserRoles().contains("fromGroup"));
+ assertFalse(otherGroups.getUserRoles().contains("fromGroup"));
+ assertNotSame(first.getUserRoles(), otherGroups.getUserRoles());
+ Mockito.verify(fixture.authContext,
Mockito.times(2)).getRolesForUserAndGroups(Mockito.isNull(), Mockito.any());
+ }
+
+ private static RangerAccessRequestImpl newRequest(RangerServiceDef
serviceDef, String user, Set<String> groups) {
+ RangerAccessRequestImpl request = new RangerAccessRequestImpl();
+ TestResource resource = new TestResource();
+ resource.setServiceDef(serviceDef);
+ request.setResource(resource);
+ request.setUser(user);
+ request.setUserGroups(new HashSet<>(groups));
+
+ return request;
+ }
+
+ private static void attach(RangerBatchEvalContext batchEvalContext,
RangerAccessRequestImpl... requests) {
+ for (RangerAccessRequestImpl request : requests) {
+ RangerAccessRequestUtil.setBatchEvalContext(request.getContext(),
batchEvalContext);
+ }
+ }
+
+ private static RangerRoles roles(RangerRole... roleArray) {
+ RangerRoles roles = new RangerRoles();
+ roles.setRangerRoles(new HashSet<>(Arrays.asList(roleArray)));
+
+ return roles;
+ }
+
+ private static ProcessorFixture newFixture(boolean useRangerGroups,
boolean useOnlyRangerGroups, boolean convertEmail, boolean nameTransformation,
String prefix, RangerUserStore userStore, RangerRoles roles) {
+ RangerPluginConfig config = Mockito.mock(RangerPluginConfig.class);
+ Mockito.when(config.isUseRangerGroups()).thenReturn(useRangerGroups);
+
Mockito.when(config.isUseOnlyRangerGroups()).thenReturn(useOnlyRangerGroups);
+
Mockito.when(config.isConvertEmailToUsername()).thenReturn(convertEmail);
+ Mockito.when(config.getPropertyPrefix()).thenReturn(prefix);
+ Mockito.when(config.getBoolean(prefix +
RangerCommonConstants.PLUGIN_CONFIG_SUFFIX_NAME_TRANSFORMATION,
false)).thenReturn(nameTransformation);
+
+ RangerPluginContext pluginContext =
Mockito.mock(RangerPluginContext.class);
+ Mockito.when(pluginContext.getConfig()).thenReturn(config);
+ Mockito.when(pluginContext.getClusterName()).thenReturn("clA");
+ Mockito.when(pluginContext.getClusterType()).thenReturn("typeB");
+
+ PolicyEngine policyEngine = Mockito.mock(PolicyEngine.class);
+
Mockito.when(policyEngine.getPluginContext()).thenReturn(pluginContext);
+
Mockito.when(policyEngine.getUseForwardedIPAddress()).thenReturn(false);
+ Mockito.when(policyEngine.getTrustedProxyAddresses()).thenReturn(null);
+
+ RangerAuthContext authContext = Mockito.spy(new RangerAuthContext(new
HashMap<>(), null, roles, userStore));
+ Mockito.when(pluginContext.getAuthContext()).thenReturn(authContext);
+
+ RangerServiceDef serviceDef = new RangerServiceDef();
+ serviceDef.setName("dummy");
+
+ return new ProcessorFixture(new
RangerDefaultRequestProcessor(policyEngine), authContext, serviceDef, new
RangerBatchEvalContext());
+ }
+
+ private static final class ProcessorFixture {
+ private final RangerDefaultRequestProcessor processor;
+ private final RangerAuthContext authContext;
+ private final RangerServiceDef serviceDef;
+ private final RangerBatchEvalContext batchEvalContext;
+
+ private ProcessorFixture(RangerDefaultRequestProcessor processor,
RangerAuthContext authContext, RangerServiceDef serviceDef,
RangerBatchEvalContext batchEvalContext) {
+ this.processor = processor;
+ this.authContext = authContext;
+ this.serviceDef = serviceDef;
+ this.batchEvalContext = batchEvalContext;
+ }
+ }
}
diff --git
a/plugin-kafka/src/test/java/org/apache/ranger/authorization/kafka/authorizer/TestRangerKafkaAuthorizerSharedRoles.java
b/plugin-kafka/src/test/java/org/apache/ranger/authorization/kafka/authorizer/TestRangerKafkaAuthorizerSharedRoles.java
new file mode 100644
index 000000000..9d9e9af6b
--- /dev/null
+++
b/plugin-kafka/src/test/java/org/apache/ranger/authorization/kafka/authorizer/TestRangerKafkaAuthorizerSharedRoles.java
@@ -0,0 +1,223 @@
+/*
+ * 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;
+
+import org.apache.kafka.common.acl.AclOperation;
+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.common.security.auth.KafkaPrincipal;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+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.authorization.hadoop.config.RangerPluginConfig;
+import org.apache.ranger.plugin.model.RangerPolicy;
+import org.apache.ranger.plugin.model.RangerRole;
+import org.apache.ranger.plugin.policyengine.RangerAccessRequest;
+import org.apache.ranger.plugin.policyengine.RangerAccessResult;
+import org.apache.ranger.plugin.policyengine.RangerPolicyEngineOptions;
+import org.apache.ranger.plugin.service.RangerBasePlugin;
+import org.apache.ranger.plugin.store.EmbeddedServiceDefsUtil;
+import org.apache.ranger.plugin.util.RangerAccessRequestUtil;
+import org.apache.ranger.plugin.util.RangerRoles;
+import org.apache.ranger.plugin.util.ServicePolicies;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestRangerKafkaAuthorizerSharedRoles {
+ @AfterEach
+ void clearPlugin() throws Exception {
+ setPlugin(null);
+ }
+
+ @Test
+ void authorizeSharesOneRoleSetAcrossActions() throws Exception {
+ RecordingPlugin plugin = new RecordingPlugin(pluginConfig());
+ plugin.setPolicies(servicePolicies());
+ plugin.setRoles(roles());
+ setPlugin(plugin);
+
+ AuthorizableRequestContext context = new RequestContext(new
KafkaPrincipal(KafkaPrincipal.USER_TYPE, "alice"));
+
+ List<Action> actions = new ArrayList<>();
+ actions.add(topicRead("orders"));
+ actions.add(topicRead("payments"));
+ actions.add(topicRead("shipping"));
+
+ List<AuthorizationResult> results = new
RangerKafkaAuthorizer().authorize(context, actions);
+
+ assertEquals(Collections.nCopies(actions.size(),
AuthorizationResult.ALLOWED), results);
+ assertEquals(actions.size(), plugin.captured.size());
+
+ Set<String> roles = null;
+
+ for (RangerAccessRequest request : plugin.captured) {
+ assertTrue(request.getUserRoles().contains("readers"));
+
assertNull(RangerAccessRequestUtil.getBatchEvalContext(request.getContext()));
+
+ if (roles == null) {
+ roles = request.getUserRoles();
+ } else {
+ assertSame(roles, request.getUserRoles());
+ }
+ }
+ }
+
+ private static RangerPluginConfig pluginConfig() {
+ RangerPolicyEngineOptions options = new RangerPolicyEngineOptions();
+ options.disablePolicyRefresher = true;
+ options.disableTagRetriever = true;
+ options.disableUserStoreRetriever = true;
+ options.disableGdsInfoRetriever = true;
+
+ return new RangerPluginConfig("kafka", "svc", "app", null, null,
options);
+ }
+
+ private static ServicePolicies servicePolicies() throws Exception {
+ RangerPolicy.RangerPolicyItemAccess access = new
RangerPolicy.RangerPolicyItemAccess();
+ access.setType(RangerKafkaAuthorizer.ACCESS_TYPE_READ);
+ access.setIsAllowed(true);
+
+ RangerPolicy.RangerPolicyItem item = new
RangerPolicy.RangerPolicyItem();
+ item.setUsers(Collections.singletonList("alice"));
+ item.setRoles(Collections.singletonList("readers"));
+ item.setAccesses(Collections.singletonList(access));
+
+ Map<String, RangerPolicy.RangerPolicyResource> resources = new
HashMap<>();
+ resources.put("topic", new RangerPolicy.RangerPolicyResource("*"));
+
+ RangerPolicy policy = new RangerPolicy();
+ policy.setId(1L);
+ policy.setName("all-topics");
+ policy.setService("svc");
+ policy.setIsEnabled(true);
+ policy.setResources(resources);
+ policy.setPolicyItems(Collections.singletonList(item));
+
+ ServicePolicies policies = new ServicePolicies();
+ policies.setServiceName("svc");
+ policies.setPolicyVersion(1L);
+
policies.setServiceDef(EmbeddedServiceDefsUtil.instance().getEmbeddedServiceDef(EmbeddedServiceDefsUtil.EMBEDDED_SERVICEDEF_KAFKA_NAME));
+ policies.setPolicies(Collections.singletonList(policy));
+
+ return policies;
+ }
+
+ private static RangerRoles roles() {
+ RangerRole readers = new RangerRole();
+ readers.setName("readers");
+ readers.setUsers(Collections.singletonList(new
RangerRole.RoleMember("alice", false)));
+
+ RangerRoles roles = new RangerRoles();
+ roles.setRangerRoles(Collections.singleton(readers));
+
+ return roles;
+ }
+
+ private static Action topicRead(String topic) {
+ return new Action(AclOperation.READ, new
ResourcePattern(ResourceType.TOPIC, topic, PatternType.LITERAL), 1, true, true);
+ }
+
+ private static void setPlugin(RangerBasePlugin plugin) throws Exception {
+ Field field =
RangerKafkaAuthorizer.class.getDeclaredField("rangerPlugin");
+
+ field.setAccessible(true);
+ field.set(null, plugin);
+ }
+
+ private static final class RequestContext implements
AuthorizableRequestContext {
+ private final KafkaPrincipal principal;
+
+ private RequestContext(KafkaPrincipal principal) {
+ this.principal = principal;
+ }
+
+ @Override
+ public String listenerName() {
+ return "SASL_PLAINTEXT";
+ }
+
+ @Override
+ public SecurityProtocol securityProtocol() {
+ return SecurityProtocol.SASL_PLAINTEXT;
+ }
+
+ @Override
+ public KafkaPrincipal principal() {
+ return principal;
+ }
+
+ @Override
+ public InetAddress clientAddress() {
+ return InetAddress.getLoopbackAddress();
+ }
+
+ @Override
+ public int requestType() {
+ return 0;
+ }
+
+ @Override
+ public int requestVersion() {
+ return 0;
+ }
+
+ @Override
+ public String clientId() {
+ return "cid";
+ }
+
+ @Override
+ public int correlationId() {
+ return 1;
+ }
+ }
+
+ private static final class RecordingPlugin extends RangerBasePlugin {
+ private Collection<RangerAccessRequest> captured;
+
+ private RecordingPlugin(RangerPluginConfig config) {
+ super(config);
+ }
+
+ @Override
+ public Collection<RangerAccessResult>
isAccessAllowed(Collection<RangerAccessRequest> requests) {
+ captured = requests;
+
+ return super.isAccessAllowed(requests);
+ }
+ }
+}
