This is an automated email from the ASF dual-hosted git repository. ramackri pushed a commit to branch ranger-kafka-dynamic-partition-plan in repository https://gitbox.apache.org/repos/asf/ranger.git
commit 86b9a9706b2cb10efeae09821a4d10723a361bc7 Author: ramk <[email protected]> AuthorDate: Sun Jun 21 10:28:50 2026 +0530 Compose auth_to_local rules from partition-plan allowlists in dynamic mode. Add AuthToLocalRuleCatalog and AuthToLocalRuleComposer to build KerberosName rules from the XML catalog and the union of services[].allowedUsers, applied on PartitionPlanHolder.install without storing rules in the plan JSON. Co-authored-by: Cursor <[email protected]> --- .../kafka/partition/AuthToLocalRuleCatalog.java | 168 +++++++++++++++++++++ .../kafka/partition/AuthToLocalRuleComposer.java | 155 +++++++++++++++++++ .../kafka/partition/PartitionPlanHolder.java | 2 + .../org/apache/ranger/audit/rest/AuditREST.java | 25 +-- .../partition/AuthToLocalRuleComposerTest.java | 144 ++++++++++++++++++ 5 files changed, 485 insertions(+), 9 deletions(-) diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/AuthToLocalRuleCatalog.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/AuthToLocalRuleCatalog.java new file mode 100644 index 000000000..f11be02f2 --- /dev/null +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/AuthToLocalRuleCatalog.java @@ -0,0 +1,168 @@ +/* + * 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.audit.producer.kafka.partition; + +import org.apache.commons.lang3.StringUtils; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import java.util.stream.Collectors; + +/** Parsed {@code ranger.audit.ingestor.auth.to.local} catalog for dynamic rule composition. */ +final class AuthToLocalRuleCatalog { + private static final Pattern SUBSTITUTION_TARGET = Pattern.compile("s/\\.\\*/([^/]+)/\\s*$"); + private static final String SIMPLE_RULE_TEMPLATE = "RULE:[2:$1/$2@$0](%s/.*@.*)s/.*/%s/"; + + private final List<CatalogEntry> primaryRulesInOrder; + private final List<String> tailRules; + + AuthToLocalRuleCatalog(List<CatalogEntry> primaryRulesInOrder, List<String> tailRules) { + this.primaryRulesInOrder = Collections.unmodifiableList(new ArrayList<>(primaryRulesInOrder)); + this.tailRules = Collections.unmodifiableList(new ArrayList<>(tailRules)); + } + + static AuthToLocalRuleCatalog parse(String rawRules) { + List<CatalogEntry> primary = new ArrayList<>(); + List<String> tail = new ArrayList<>(); + for (String token : tokenizeRules(rawRules)) { + if (isTailRule(token)) { + tail.add(token); + } else { + primary.add(new CatalogEntry(token, extractTargetShortName(token))); + } + } + return new AuthToLocalRuleCatalog(primary, tail); + } + + /** Full static rule set (all primary rules in catalog order + tail). */ + String composeFull() { + List<String> lines = new ArrayList<>(primaryRulesInOrder.size() + tailRules.size()); + for (CatalogEntry entry : primaryRulesInOrder) { + lines.add(entry.ruleLine); + } + lines.addAll(tailRules); + return joinRules(lines); + } + + /** + * Subset of catalog rules for the union of active short usernames from partition-plan allowlists. + * Unknown names get a generated {@code service/host@REALM -> service} rule before tail rules. + */ + String compose(Set<String> activeShortNames) { + if (activeShortNames == null || activeShortNames.isEmpty()) { + return composeFull(); + } + + Set<String> active = activeShortNames.stream() + .filter(StringUtils::isNotBlank) + .map(String::trim) + .collect(Collectors.toCollection(LinkedHashSet::new)); + if (active.isEmpty()) { + return composeFull(); + } + + List<String> lines = new ArrayList<>(); + Set<String> covered = new LinkedHashSet<>(); + Map<String, String> catalogByShortName = catalogRulesByShortName(); + + for (CatalogEntry entry : primaryRulesInOrder) { + if (entry.targetShortName != null && active.contains(entry.targetShortName)) { + lines.add(entry.ruleLine); + covered.add(entry.targetShortName); + } + } + + List<String> generated = active.stream() + .filter(name -> !covered.contains(name)) + .sorted() + .map(AuthToLocalRuleCatalog::simpleRuleForShortName) + .collect(Collectors.toList()); + lines.addAll(generated); + lines.addAll(tailRules); + return joinRules(lines); + } + + private Map<String, String> catalogRulesByShortName() { + Map<String, String> ret = new LinkedHashMap<>(); + for (CatalogEntry entry : primaryRulesInOrder) { + if (entry.targetShortName != null) { + ret.putIfAbsent(entry.targetShortName, entry.ruleLine); + } + } + return ret; + } + + private static List<String> tokenizeRules(String rawRules) { + if (StringUtils.isBlank(rawRules)) { + return Collections.emptyList(); + } + return Arrays.stream(rawRules.split("\\s+")) + .map(String::trim) + .filter(s -> s.startsWith("RULE:") || "DEFAULT".equals(s)) + .collect(Collectors.toList()); + } + + private static boolean isTailRule(String ruleLine) { + return "DEFAULT".equals(ruleLine) + || ruleLine.startsWith("RULE:[1:") + || ruleLine.contains("s/@.*//"); + } + + static String extractTargetShortName(String ruleLine) { + if (StringUtils.isBlank(ruleLine)) { + return null; + } + Matcher matcher = SUBSTITUTION_TARGET.matcher(ruleLine); + if (matcher.find()) { + return matcher.group(1); + } + return null; + } + + static String simpleRuleForShortName(String shortName) { + return String.format(SIMPLE_RULE_TEMPLATE, shortName, shortName); + } + + private static String joinRules(List<String> lines) { + return String.join("\n", lines); + } + + int getPrimaryRuleCount() { + return primaryRulesInOrder.size(); + } + + static final class CatalogEntry { + private final String ruleLine; + private final String targetShortName; + + CatalogEntry(String ruleLine, String targetShortName) { + this.ruleLine = ruleLine; + this.targetShortName = targetShortName; + } + } +} diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/AuthToLocalRuleComposer.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/AuthToLocalRuleComposer.java new file mode 100644 index 000000000..59ad9511b --- /dev/null +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/AuthToLocalRuleComposer.java @@ -0,0 +1,155 @@ +/* + * 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.audit.producer.kafka.partition; + +import org.apache.commons.lang3.StringUtils; +import org.apache.hadoop.security.authentication.util.KerberosName; +import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlan; +import org.apache.ranger.audit.producer.kafka.partition.model.ServiceAllowlistEntry; +import org.apache.ranger.audit.server.AuditServerConfig; +import org.apache.ranger.audit.server.AuditServerConstants; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.LinkedHashSet; +import java.util.Map; +import java.util.Properties; +import java.util.Set; + +/** + * Builds effective {@code auth_to_local} rules from the XML catalog and the union of partition-plan + * {@code services[].allowedUsers}. Rules are not stored in the partition-plan JSON document. + */ +public final class AuthToLocalRuleComposer { + private static final Logger LOG = LoggerFactory.getLogger(AuthToLocalRuleComposer.class); + + private static final AuthToLocalRuleComposer INSTANCE = new AuthToLocalRuleComposer(); + + private volatile AuthToLocalRuleCatalog catalog; + private volatile String lastAppliedRules; + private volatile int lastAppliedPlanVersion; + + private AuthToLocalRuleComposer() { + } + + public static AuthToLocalRuleComposer getInstance() { + return INSTANCE; + } + + /** Loads the rule catalog from {@code ranger.audit.ingestor.auth.to.local} site XML. */ + public synchronized void initializeFromConfig() { + Properties props = AuditServerConfig.getInstance().getProperties(); + initializeFromProperties(props); + } + + synchronized void initializeFromProperties(Properties props) { + String raw = props.getProperty(AuditServerConstants.PROP_AUTH_TO_LOCAL); + catalog = AuthToLocalRuleCatalog.parse(raw); + lastAppliedRules = null; + lastAppliedPlanVersion = 0; + LOG.debug("Loaded auth_to_local catalog with {} primary rules", catalog.getPrimaryRuleCount()); + } + + /** Applies the full XML catalog (non-dynamic / startup fallback). */ + public void applyStaticRules() { + AuthToLocalRuleCatalog loaded = requireCatalog(); + applyRules(loaded.composeFull(), 0); + } + + /** + * When dynamic partition-plan mode is enabled, compose rules from the union of allowlisted short + * names and install them before audit REST authorization runs. + */ + public void applyForPlan(PartitionPlan plan) { + Properties props = AuditServerConfig.getInstance().getProperties(); + if (!PartitionPlanKafkaConfig.isDynamicPartitionPlanEnabled(props, PartitionPlanService.INGESTOR_PROP_PREFIX)) { + return; + } + if (plan == null) { + return; + } + + AuthToLocalRuleCatalog loaded = requireCatalog(); + Set<String> active = unionAllowedUsers(plan); + String rules = loaded.compose(active); + applyRules(rules, plan.getVersion()); + LOG.info("Applied composed auth_to_local rules for plan version {} ({} active short names)", plan.getVersion(), active.size()); + } + + /** Visible for tests — composes without applying Kerberos rules. */ + String composeForActiveShortNames(Set<String> activeShortNames) { + return requireCatalog().compose(activeShortNames); + } + + /** Clears cached apply state between unit tests. */ + public synchronized void resetForTests() { + lastAppliedRules = null; + lastAppliedPlanVersion = 0; + } + + static Set<String> unionAllowedUsers(PartitionPlan plan) { + Set<String> active = new LinkedHashSet<>(); + if (plan == null || plan.getServices() == null) { + return active; + } + for (Map.Entry<String, ServiceAllowlistEntry> entry : plan.getServices().entrySet()) { + ServiceAllowlistEntry allowlist = entry.getValue(); + if (allowlist == null) { + continue; + } + for (String user : allowlist.getAllowedUsers()) { + if (StringUtils.isNotBlank(user)) { + active.add(user.trim()); + } + } + } + return active; + } + + private AuthToLocalRuleCatalog requireCatalog() { + AuthToLocalRuleCatalog loaded = catalog; + if (loaded == null) { + initializeFromConfig(); + loaded = catalog; + } + if (loaded == null) { + throw new IllegalStateException("auth_to_local catalog is not loaded"); + } + return loaded; + } + + private synchronized void applyRules(String rules, int planVersion) { + if (StringUtils.isBlank(rules)) { + LOG.warn("Skipping auth_to_local apply: composed rules are blank"); + return; + } + if (rules.equals(lastAppliedRules) && planVersion == lastAppliedPlanVersion) { + return; + } + try { + KerberosName.setRules(rules); + lastAppliedRules = rules; + lastAppliedPlanVersion = planVersion; + LOG.debug("KerberosName auth_to_local rules updated (planVersion={})", planVersion); + } catch (Exception e) { + LOG.error("Failed to apply composed auth_to_local rules for plan version {}: {}", planVersion, e.getMessage(), e); + } + } +} diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanHolder.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanHolder.java index 9af9b1d4e..1481db2c5 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanHolder.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanHolder.java @@ -54,6 +54,7 @@ public void install(PartitionPlan plan, Integer kafkaPartitionCount) { PartitionPlanValidator.validate(plan, kafkaPartitionCount); planRef.set(plan); lastInstalledVersion = plan.getVersion(); + AuthToLocalRuleComposer.getInstance().applyForPlan(plan); } /** @@ -82,5 +83,6 @@ public boolean hasServiceAllowlist() { public void resetForTests() { planRef.set(null); lastInstalledVersion = 0; + AuthToLocalRuleComposer.getInstance().resetForTests(); } } diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/rest/AuditREST.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/rest/AuditREST.java index 5c06dd6b7..1a2c223b7 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/rest/AuditREST.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/rest/AuditREST.java @@ -23,7 +23,9 @@ import org.apache.hadoop.security.authentication.util.KerberosName; import org.apache.ranger.audit.model.AuthzAuditEvent; import org.apache.ranger.audit.producer.AuditDestinationMgr; +import org.apache.ranger.audit.producer.kafka.partition.AuthToLocalRuleComposer; import org.apache.ranger.audit.producer.kafka.partition.PartitionPlanHolder; +import org.apache.ranger.audit.producer.kafka.partition.PartitionPlanKafkaConfig; import org.apache.ranger.audit.producer.kafka.partition.PartitionPlanService; import org.apache.ranger.audit.producer.kafka.partition.exception.PartitionPlanConflictException; import org.apache.ranger.audit.producer.kafka.partition.exception.PartitionPlanException; @@ -533,21 +535,26 @@ private String applyAuthToLocal(String principal) { } /** - * Rules are loaded from ranger.audit.ingestor.auth.to.local property in ranger-audit-ingestor-site.xml. + * Loads the auth_to_local catalog from ranger-audit-ingestor-site.xml. When dynamic partition-plan + * mode is disabled, applies the full catalog immediately. When enabled, rules are composed from + * partition-plan allowlists on {@link PartitionPlanHolder#install(PartitionPlan, Integer)}. */ private static void initializeAuthToLocal() { AuditServerConfig config = AuditServerConfig.getInstance(); String authToLocalRules = config.get(AuditServerConstants.PROP_AUTH_TO_LOCAL); - if (StringUtils.isNotEmpty(authToLocalRules)) { - try { - KerberosName.setRules(authToLocalRules); - LOG.debug("Auth_to_local rules: {}", authToLocalRules); - } catch (Exception e) { - LOG.error("Failed to set auth_to_local rules from configuration: {}", e.getMessage(), e); - } - } else { + if (StringUtils.isEmpty(authToLocalRules)) { LOG.warn("No auth_to_local rules configured. Kerberos principal mapping may not work correctly."); LOG.warn("Set property '{}' in ranger-audit-ingestor-site.xml", AuditServerConstants.PROP_AUTH_TO_LOCAL); + return; + } + + AuthToLocalRuleComposer composer = AuthToLocalRuleComposer.getInstance(); + composer.initializeFromConfig(); + if (PartitionPlanKafkaConfig.isDynamicPartitionPlanEnabled(config.getProperties(), AuditServerConstants.PROP_PREFIX_AUDIT_SERVER.substring(0, AuditServerConstants.PROP_PREFIX_AUDIT_SERVER.length() - 1))) { + LOG.info("Dynamic partition plan enabled; auth_to_local rules will be composed from allowlisted short names on plan install"); + } else { + composer.applyStaticRules(); + LOG.debug("Applied static auth_to_local catalog from site XML"); } } diff --git a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/AuthToLocalRuleComposerTest.java b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/AuthToLocalRuleComposerTest.java new file mode 100644 index 000000000..01b2bbc32 --- /dev/null +++ b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/AuthToLocalRuleComposerTest.java @@ -0,0 +1,144 @@ +/* + * 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.audit.producer.kafka.partition; + +import org.apache.hadoop.security.authentication.util.KerberosName; +import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlan; +import org.apache.ranger.audit.producer.kafka.partition.model.PluginPartitionAssignment; +import org.apache.ranger.audit.producer.kafka.partition.model.ServiceAllowlistEntry; +import org.apache.ranger.audit.server.AuditServerConstants; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Properties; +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.assertTrue; + +public class AuthToLocalRuleComposerTest { + private static final String SAMPLE_CATALOG = + "RULE:[2:$1/$2@$0]([ndj]n/.*@.*|hdfs/.*@.*)s/.*/hdfs/ " + + "RULE:[2:$1/$2@$0]([rn]m/.*@.*|yarn/.*@.*)s/.*/yarn/ " + + "RULE:[2:$1/$2@$0](jhs/.*@.*)s/.*/mapred/ " + + "RULE:[2:$1/$2@$0](hive/.*@.*)s/.*/hive/ " + + "RULE:[2:$1/$2@$0](rangerkms/.*@.*)s/.*/rangerkms/ " + + "RULE:[2:$1/$2@$0](om/.*@.*)s/.*/om/ " + + "RULE:[2:$1/$2@$0](ozone/.*@.*)s/.*/ozone/ " + + "RULE:[1:$1@$0](.*@.*)s/@.*// " + + "DEFAULT"; + + private AuthToLocalRuleComposer composer; + + @BeforeEach + public void setUp() { + composer = AuthToLocalRuleComposer.getInstance(); + composer.resetForTests(); + Properties props = new Properties(); + props.setProperty(AuditServerConstants.PROP_AUTH_TO_LOCAL, SAMPLE_CATALOG); + composer.initializeFromProperties(props); + } + + @AfterEach + public void tearDown() { + composer.resetForTests(); + PartitionPlanHolder.getInstance().resetForTests(); + } + + @Test + public void testComposeKmsOnlyIncludesRangerkmsRuleAndTail() { + String rules = composer.composeForActiveShortNames(Set.of("rangerkms")); + + assertTrue(rules.contains("(rangerkms/.*@.*)s/.*/rangerkms/")); + assertFalse(rules.contains("(hive/.*@.*)s/.*/hive/")); + assertTrue(rules.contains("RULE:[1:$1@$0](.*@.*)s/@.*//")); + assertTrue(rules.endsWith("DEFAULT")); + } + + @Test + public void testComposeHdfsUsesCatalogRegexNotSimpleRule() { + String rules = composer.composeForActiveShortNames(Set.of("hdfs")); + + assertTrue(rules.contains("([ndj]n/.*@.*|hdfs/.*@.*)s/.*/hdfs/")); + assertFalse(rules.contains("(hdfs/.*@.*)s/.*/hdfs/")); + } + + @Test + public void testComposeOzoneIncludesMultipleShortNames() { + String rules = composer.composeForActiveShortNames(Set.of("ozone", "om", "scm", "dn")); + + assertTrue(rules.contains("(ozone/.*@.*)s/.*/ozone/")); + assertTrue(rules.contains("(om/.*@.*)s/.*/om/")); + } + + @Test + public void testComposeUnknownShortNameGeneratesSimpleRule() { + String rules = composer.composeForActiveShortNames(Set.of("myapp")); + + assertTrue(rules.contains("(myapp/.*@.*)s/.*/myapp/")); + } + + @Test + public void testUnionAllowedUsersFromPlan() { + Map<String, ServiceAllowlistEntry> services = new LinkedHashMap<>(); + services.put("dev_kms", ServiceAllowlistEntry.ofUsers("rangerkms")); + services.put("dev_ozone", ServiceAllowlistEntry.ofUsers("om", "ozone")); + PartitionPlan plan = PartitionPlan.builder() + .topic("ranger_audits") + .version(2) + .topicPartitionCount(6) + .plugins(Map.of("kms", PluginPartitionAssignment.of(0, 1))) + .buffer(PluginPartitionAssignment.of(2, 3, 4, 5)) + .services(services) + .build(); + + Set<String> active = AuthToLocalRuleComposer.unionAllowedUsers(plan); + + assertEquals(Set.of("rangerkms", "om", "ozone"), active); + } + + @Test + public void testKerberosMappingForComposedKmsRules() throws Exception { + String rules = composer.composeForActiveShortNames(Set.of("rangerkms")); + KerberosName.setRules(rules); + + assertEquals("rangerkms", new KerberosName("rangerkms/[email protected]").getShortName()); + } + + @Test + public void testKerberosMappingForComposedHdfsRules() throws Exception { + String rules = composer.composeForActiveShortNames(Set.of("hdfs")); + KerberosName.setRules(rules); + + assertEquals("hdfs", new KerberosName("nn/[email protected]").getShortName()); + } + + @Test + public void testExtractTargetShortNameFromCatalogLines() { + assertEquals("hdfs", AuthToLocalRuleCatalog.extractTargetShortName( + "RULE:[2:$1/$2@$0]([ndj]n/.*@.*|hdfs/.*@.*)s/.*/hdfs/")); + assertEquals("rangerkms", AuthToLocalRuleCatalog.extractTargetShortName( + "RULE:[2:$1/$2@$0](rangerkms/.*@.*)s/.*/rangerkms/")); + assertEquals("mapred", AuthToLocalRuleCatalog.extractTargetShortName( + "RULE:[2:$1/$2@$0](jhs/.*@.*)s/.*/mapred/")); + } +}
