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 0a1f805915ecacd1f9714d1804d681d1d4f28ba8 Author: ramk <[email protected]> AuthorDate: Tue Jun 23 18:23:38 2026 +0530 Fix dynamic partition plan review items and Solr Kerberos config. Close the watcher startup gap by replaying the plan topic from the beginning before the background thread starts; distinguish missing-repo allowlist fallback from explicit empty allowlists; extract ServiceAllowlistResolver and PartitionPlanUpdateApplier with unit tests; apply RANGER-5654 Solr useTicketCache=false config-only fix and revert AbstractKerberosUser to master. Co-authored-by: Cursor <[email protected]> --- .../ranger/audit/utils/AbstractKerberosUser.java | 12 +-- .../conf/ranger-audit-dispatcher-solr-site.xml | 2 +- .../kafka/partition/AuthToLocalRuleCatalog.java | 22 ++++- .../kafka/partition/AuthToLocalRuleComposer.java | 43 ++++++-- .../partition/KafkaPartitionPlanRegistry.java | 2 +- .../kafka/partition/PartitionPlanBootstrap.java | 2 +- .../partition/PartitionPlanBootstrapConfig.java | 2 +- .../kafka/partition/PartitionPlanHolder.java | 14 ++- .../kafka/partition/PartitionPlanKafkaConfig.java | 2 +- .../partition/PartitionPlanRequestValidator.java | 2 +- .../partition/PartitionPlanUpdateApplier.java | 70 +++++++++++++ .../kafka/partition/PartitionPlanWatcher.java | 32 ++---- .../kafka/partition/ServiceAllowlistBootstrap.java | 2 +- .../kafka/partition/ServiceAllowlistResolver.java | 62 ++++++++++++ .../constants/PartitionPlanConstants.java | 2 +- .../org/apache/ranger/audit/rest/AuditREST.java | 38 +++---- .../kafka/partition/PartitionPlanHolderTest.java | 100 +++++++++++++++++++ .../partition/PartitionPlanUpdateApplierTest.java | 110 +++++++++++++++++++++ .../partition/ServiceAllowlistBootstrapTest.java | 21 ++-- .../partition/ServiceAllowlistResolverTest.java | 96 ++++++++++++++++++ .../ranger-audit-dispatcher-solr-site.xml | 63 +----------- .../scripts/audit/partition-plan-e2e-lib.sh | 1 + 22 files changed, 552 insertions(+), 148 deletions(-) diff --git a/agents-audit/core/src/main/java/org/apache/ranger/audit/utils/AbstractKerberosUser.java b/agents-audit/core/src/main/java/org/apache/ranger/audit/utils/AbstractKerberosUser.java index 1f990ec7b..47d16f31a 100644 --- a/agents-audit/core/src/main/java/org/apache/ranger/audit/utils/AbstractKerberosUser.java +++ b/agents-audit/core/src/main/java/org/apache/ranger/audit/utils/AbstractKerberosUser.java @@ -172,16 +172,8 @@ public synchronized boolean checkTGTAndRelogin() throws LoginException { LOG.debug("Performing relogin for {}", getPrincipal()); - try { - logout(); - login(); - } catch (LoginException e) { - LOG.warn("Relogin failed for {}, recreating JAAS login context: {}", getPrincipal(), e.getMessage()); - loginContext = null; - subject = new Subject(); - loggedIn.set(false); - login(); - } + logout(); + login(); return true; } diff --git a/audit-server/audit-dispatcher/dispatcher-solr/src/main/resources/conf/ranger-audit-dispatcher-solr-site.xml b/audit-server/audit-dispatcher/dispatcher-solr/src/main/resources/conf/ranger-audit-dispatcher-solr-site.xml index b82128bee..253ed9895 100644 --- a/audit-server/audit-dispatcher/dispatcher-solr/src/main/resources/conf/ranger-audit-dispatcher-solr-site.xml +++ b/audit-server/audit-dispatcher/dispatcher-solr/src/main/resources/conf/ranger-audit-dispatcher-solr-site.xml @@ -175,7 +175,7 @@ <property> <name>xasecure.audit.jaas.Client.option.useTicketCache</name> - <value>true</value> + <value>false</value> <description>Allow use of cached Kerberos tickets for Solr authentication</description> </property> 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 index 9d1daa021..a4314b7e1 100644 --- 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 @@ -31,7 +31,7 @@ import java.util.regex.Pattern; import java.util.stream.Collectors; -/** Parses and composes Kerberos {@code auth_to_local} rules from the ingestor site XML catalog. */ +/** Parses and composes Kerberos {@code auth_to_local} rules from {@code ranger.audit.ingestor.auth.to.local}. */ public class AuthToLocalRuleCatalog { private static final Pattern RULE_SUBSTITUTION_SHORT_NAME_PATTERN = Pattern.compile("s/\\.\\*/([^/]+)/\\s*$"); @@ -53,8 +53,7 @@ public static AuthToLocalRuleCatalog parse(String rawAuthToLocalRules) { if (isFallbackKerberosRule(ruleLine)) { fallbackRuleLines.add(ruleLine); } else { - primaryCatalogRulesInOrder.add( - new PrimaryCatalogRule(ruleLine, extractMappedShortName(ruleLine))); + primaryCatalogRulesInOrder.add(new PrimaryCatalogRule(ruleLine, extractMappedShortName(ruleLine))); } } return new AuthToLocalRuleCatalog(primaryCatalogRulesInOrder, fallbackRuleLines); @@ -71,8 +70,16 @@ public String composeFullCatalogRules() { } /** - * 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 fallback rules. + * Builds the active rule set for dynamic mode: catalog rules whose mapped short name is in the union, + * plus generated rules for union members not covered by the catalog, plus fallback tail. + * + * <p>Examples (union member to effective rule): + * <ul> + * <li>{@code hdfs} in union: full hdfs catalog line (nn/dn/jn/hdfs principals map to hdfs)</li> + * <li>{@code nn} in union but catalog mapped name is hdfs: generated nn rule plus hdfs catalog line</li> + * <li>{@code foo} in union, not in catalog: generated foo/host@REALM to foo rule</li> + * </ul> + * Empty union falls back to {@link #composeFullCatalogRules()}. */ public String composeRulesForAllowedShortNames(Set<String> allowedUserShortNames) { if (allowedUserShortNames == null || allowedUserShortNames.isEmpty()) { @@ -111,6 +118,10 @@ public int getPrimaryCatalogRuleCount() { return primaryCatalogRulesInOrder.size(); } + /** + * Mapped Kerberos short name from a catalog rule line (substitution suffix s/.star/<name>/). + * Example: hdfs catalog rule maps nn/dn/jn/hdfs principals to short name hdfs. + */ public static String extractMappedShortName(String ruleLine) { if (StringUtils.isBlank(ruleLine)) { return null; @@ -136,6 +147,7 @@ private static boolean isFallbackKerberosRule(String ruleLine) { return "DEFAULT".equals(ruleLine) || ruleLine.startsWith("RULE:[1:") || ruleLine.contains("s/@.*//"); } + /** Simple rule for allowlist short names absent from the XML catalog: {@code myapp/host@REALM -> myapp}. */ private static String buildGeneratedShortNameRule(String shortName) { return String.format(GENERATED_SHORT_NAME_RULE_TEMPLATE, shortName, shortName); } 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 index c98064d9a..95184a6f2 100644 --- 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 @@ -35,10 +35,16 @@ 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. + * Builds effective {@code auth_to_local} rules from the XML catalog and the global allowlist union + * ({@link #collectAllowedUserShortNames}). Rule text is not stored in the partition-plan Kafka JSON. + * + * <p>Two-step audit POST identity (see {@link ServiceAllowlistResolver}): + * <ol> + * <li><b>Kerberos mapping</b> — {@code KerberosName.getShortName()} uses composed rules (global union)</li> + * <li><b>Authorization</b> — per-repo {@code services[repo].allowedUsers} (or static XML fallback)</li> + * </ol> */ -public final class AuthToLocalRuleComposer { +public class AuthToLocalRuleComposer { private static final Logger LOG = LoggerFactory.getLogger(AuthToLocalRuleComposer.class); private static final AuthToLocalRuleComposer INSTANCE = new AuthToLocalRuleComposer(); @@ -61,7 +67,7 @@ public synchronized void initializeFromConfig() { initializeFromProperties(ingestorProperties); } - synchronized void initializeFromProperties(Properties ingestorProperties) { + public synchronized void initializeFromProperties(Properties ingestorProperties) { String rawAuthToLocalRules = ingestorProperties.getProperty(AuditServerConstants.PROP_AUTH_TO_LOCAL); ruleCatalog = AuthToLocalRuleCatalog.parse(rawAuthToLocalRules); lastAppliedRuleText = null; @@ -94,8 +100,20 @@ public void applyStartupRulesForDynamicMode(Properties ingestorProperties, Strin } /** - * When dynamic partition-plan mode is enabled, compose rules from the union of allowlisted short - * names and install them before audit REST authorization runs. + * When dynamic partition-plan mode is enabled, compose {@code auth_to_local} rules from the XML + * catalog plus the global allowlist union, then install via {@link KerberosName#setRules(String)}. + * + * <p>Called from {@link PartitionPlanHolder#install} on every plan version. Example plan: + * <pre>{@code + * services: { + * dev_hdfs: { allowedUsers: [hdfs, nn] }, + * dev_hive: { allowedUsers: [hive] }, + * dev_foo: { allowedUsers: [foo] } // new plugin, not in XML catalog + * } + * }</pre> + * Union of hdfs, nn, hive, foo activates hdfs catalog rule (nn/dn/jn/hdfs principals to hdfs), + * hive catalog rule, and a generated foo/host@REALM to foo rule. Per-repo POST checks still + * use each repo's own allowlist, not the union. */ public void applyForPlan(PartitionPlan partitionPlan) { Properties ingestorProperties = AuditServerConfig.getInstance().getProperties(); @@ -115,7 +133,7 @@ public void applyForPlan(PartitionPlan partitionPlan) { } /** Visible for tests — composes without applying Kerberos rules. */ - String composeKerberosRulesForAllowedShortNames(Set<String> allowedUserShortNames) { + public String composeKerberosRulesForAllowedShortNames(Set<String> allowedUserShortNames) { return requireRuleCatalog().composeRulesForAllowedShortNames(allowedUserShortNames); } @@ -127,7 +145,7 @@ public synchronized void resetForTests() { } /** When non-null, overrides {@link AuditMessageQueueUtils#partitionPlanTopicExists} for unit tests. */ - void setPartitionPlanTopicExistsTestOverride(Boolean topicExists) { + public void setPartitionPlanTopicExistsTestOverride(Boolean topicExists) { partitionPlanTopicExistsTestOverride = topicExists; } @@ -139,7 +157,14 @@ private boolean isPartitionPlanTopicPresent(Properties ingestorProperties, Strin return AuditMessageQueueUtils.partitionPlanTopicExists(ingestorProperties, ingestorPropertyPrefix); } - static Set<String> collectAllowedUserShortNames(PartitionPlan partitionPlan) { + /** + * Global allowlist union: all distinct {@code allowedUsers} short names across {@code plan.services}. + * Used only for {@code auth_to_local} composition (which mapping rules are active), not for POST authorization. + * + * <p>Example: dev_hdfs with hdfs and nn, dev_hive with hive, dev_ozone with om and ozone yields + * union hdfs, nn, hive, om, ozone. Repos not listed in services do not contribute. + */ + public static Set<String> collectAllowedUserShortNames(PartitionPlan partitionPlan) { Set<String> allowedUserShortNames = new LinkedHashSet<>(); if (partitionPlan == null || partitionPlan.getServices() == null) { return allowedUserShortNames; diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/KafkaPartitionPlanRegistry.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/KafkaPartitionPlanRegistry.java index 99f88586a..198219ae5 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/KafkaPartitionPlanRegistry.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/KafkaPartitionPlanRegistry.java @@ -38,7 +38,7 @@ import java.util.Properties; /** Kafka compacted topic implementation of {@link PartitionPlanRegistry}. */ -public final class KafkaPartitionPlanRegistry implements PartitionPlanRegistry { +public class KafkaPartitionPlanRegistry implements PartitionPlanRegistry { private static final Logger LOG = LoggerFactory.getLogger(KafkaPartitionPlanRegistry.class); private final Properties props; diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrap.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrap.java index 3dea4a95b..7ab09b8e8 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrap.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrap.java @@ -32,7 +32,7 @@ import java.util.Map; /** Builds the initial bootstrap plan from legacy XML and seeds the registry when the plan topic is empty. */ -public final class PartitionPlanBootstrap { +public class PartitionPlanBootstrap { private static final Logger LOG = LoggerFactory.getLogger(PartitionPlanBootstrap.class); private PartitionPlanBootstrap() { diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrapConfig.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrapConfig.java index 921fedc9e..349cf5def 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrapConfig.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrapConfig.java @@ -28,7 +28,7 @@ import java.util.Map; /** Inputs for building the first partition plan from legacy ingestor XML / producer config. */ -public final class PartitionPlanBootstrapConfig { +public class PartitionPlanBootstrapConfig { private final String auditTopic; private final String[] configuredPlugins; private final int defaultPartitionsPerPlugin; 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 cbf2501af..c89a391b4 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 @@ -28,7 +28,7 @@ import java.util.concurrent.atomic.AtomicReference; /** Hot-path in-memory plan for {@code AuditPartitioner} and the background watcher. */ -public final class PartitionPlanHolder { +public class PartitionPlanHolder { private static final PartitionPlanHolder INSTANCE = new PartitionPlanHolder(); private final AtomicReference<PartitionPlan> planRef = new AtomicReference<>(); @@ -59,7 +59,12 @@ public void install(PartitionPlan plan, Integer kafkaPartitionCount) { /** * Returns allowed short usernames for a service repo from the in-memory registry document. - * {@code null} when the plan has no {@code services} block (caller should fall back to static XML). + * {@code null} when the plan has no {@code services} block, or when the repo is not present + * in the plan (caller should fall back to static XML). + * Returns an empty set when the repo is present with an explicit empty allowlist (deny all). + * + * <p>Used by {@link ServiceAllowlistResolver} for per-repo POST authorization — not for the global + * allowlist union ({@link AuthToLocalRuleComposer#collectAllowedUserShortNames}). */ public Set<String> getAllowedUsersForService(String serviceName) { PartitionPlan plan = planRef.get(); @@ -67,7 +72,10 @@ public Set<String> getAllowedUsersForService(String serviceName) { return null; } ServiceAllowlistEntry entry = plan.getServices().get(serviceName); - if (entry == null || entry.getAllowedUsers().isEmpty()) { + if (entry == null) { + return null; + } + if (entry.getAllowedUsers().isEmpty()) { return Collections.emptySet(); } return Collections.unmodifiableSet(new LinkedHashSet<>(entry.getAllowedUsers())); diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanKafkaConfig.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanKafkaConfig.java index 9fb496302..486ed2ce5 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanKafkaConfig.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanKafkaConfig.java @@ -33,7 +33,7 @@ import java.util.Set; /** Kafka client settings for the partition-plan registry topic. */ -public final class PartitionPlanKafkaConfig { +public class PartitionPlanKafkaConfig { private PartitionPlanKafkaConfig() { } diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanRequestValidator.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanRequestValidator.java index 6229ff1ad..f4901e350 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanRequestValidator.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanRequestValidator.java @@ -30,7 +30,7 @@ import java.util.List; /** Validates partition-plan REST mutation request bodies before registry writes. */ -public final class PartitionPlanRequestValidator { +public class PartitionPlanRequestValidator { private PartitionPlanRequestValidator() { } diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanUpdateApplier.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanUpdateApplier.java new file mode 100644 index 000000000..0ec280ecc --- /dev/null +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanUpdateApplier.java @@ -0,0 +1,70 @@ +/* + * 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.kafka.clients.consumer.ConsumerRecord; +import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlan; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Properties; + +/** Applies compacted partition-plan Kafka records into {@link PartitionPlanHolder}. */ +public class PartitionPlanUpdateApplier { + private static final Logger LOG = LoggerFactory.getLogger(PartitionPlanUpdateApplier.class); + + @FunctionalInterface + public interface AuditTopicPartitionCountSupplier { + int getPartitionCount() throws Exception; + } + + private final Properties props; + private final String auditTopicKey; + private final PartitionPlanHolder holder; + private final AuditTopicPartitionCountSupplier partitionCountSupplier; + + public PartitionPlanUpdateApplier( + Properties props, + String auditTopicKey, + PartitionPlanHolder holder, + AuditTopicPartitionCountSupplier partitionCountSupplier) { + this.props = props; + this.auditTopicKey = auditTopicKey; + this.holder = holder != null ? holder : PartitionPlanHolder.getInstance(); + this.partitionCountSupplier = partitionCountSupplier; + } + + /** Installs a plan record when its version is newer than the in-memory copy. */ + public void applyRecordIfNewer(ConsumerRecord<String, String> record) { + if (!auditTopicKey.equals(record.key())) { + return; + } + try { + PartitionPlan plan = PartitionPlan.fromJson(record.value()); + plan = ServiceAllowlistBootstrap.mergeSiteXmlAllowlistsWhenPlanServicesMissing(plan, props); + if (plan.getVersion() <= holder.getLastInstalledVersion()) { + return; + } + holder.install(plan, partitionCountSupplier.getPartitionCount()); + LOG.info("Installed partition plan version {} from Kafka offset {}", plan.getVersion(), record.offset()); + } catch (Exception e) { + LOG.error("Ignoring invalid partition plan at offset {} for audit topic '{}'", record.offset(), auditTopicKey, e); + } + } +} diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanWatcher.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanWatcher.java index 17314c742..18ed91444 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanWatcher.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanWatcher.java @@ -39,7 +39,7 @@ import java.util.Properties; /** Background thread: load plan from Kafka (or XML-seeded initial bootstrap plan), then incrementally refresh in-memory plan. */ -public final class PartitionPlanWatcher implements Runnable { +public class PartitionPlanWatcher implements Runnable { private static final Logger LOG = LoggerFactory.getLogger(PartitionPlanWatcher.class); private final Properties props; @@ -52,6 +52,7 @@ public final class PartitionPlanWatcher implements Runnable { private PartitionPlanRegistry registry; private KafkaConsumer<String, String> consumer; + private PartitionPlanUpdateApplier planUpdateApplier; private Thread watcherThread; private volatile boolean running; @@ -63,6 +64,7 @@ public PartitionPlanWatcher(Properties props, String propPrefix, String auditTop this.refreshIntervalMs = PartitionPlanKafkaConfig.resolveRefreshIntervalMs(props, propPrefix); this.consumerPollTimeoutMs = PartitionPlanKafkaConfig.resolveConsumerPollTimeoutMs(props, propPrefix); this.planTopic = PartitionPlanKafkaConfig.resolvePlanTopicName(props, propPrefix); + this.planUpdateApplier = new PartitionPlanUpdateApplier(props, auditTopicKey, this.partitionPlanHolder, this::resolveAuditTopicPartitionCount); } /** Blocking startup: empty-registry bootstrap, install plan, then start background refresh. */ @@ -74,7 +76,8 @@ public void startBlocking() throws Exception { plan = ServiceAllowlistBootstrap.mergeSiteXmlAllowlistsWhenPlanServicesMissing(plan, props); int kafkaPartitionCount = resolveAuditTopicPartitionCount(); partitionPlanHolder.install(plan, kafkaPartitionCount); - openConsumerAtLogEnd(); + openConsumerAtBeginning(); + pollIncrementalUpdates(); running = true; watcherThread = new Thread(this, "PartitionPlanWatcher"); watcherThread.setDaemon(true); @@ -126,35 +129,16 @@ private void pollIncrementalUpdates() { do { records = consumer.poll(Duration.ofMillis(consumerPollTimeoutMs)); for (ConsumerRecord<String, String> record : records) { - applyRecordIfNewer(record); + planUpdateApplier.applyRecordIfNewer(record); } } while (!records.isEmpty()); } - /** Installs a plan record when its version is newer than the in-memory copy. */ - private void applyRecordIfNewer(ConsumerRecord<String, String> record) { - if (!auditTopicKey.equals(record.key())) { - return; - } - try { - PartitionPlan plan = PartitionPlan.fromJson(record.value()); - plan = ServiceAllowlistBootstrap.mergeSiteXmlAllowlistsWhenPlanServicesMissing(plan, props); - if (plan.getVersion() <= partitionPlanHolder.getLastInstalledVersion()) { - return; - } - int kafkaPartitionCount = resolveAuditTopicPartitionCount(); - partitionPlanHolder.install(plan, kafkaPartitionCount); - LOG.info("Installed partition plan version {} from Kafka offset {}", plan.getVersion(), record.offset()); - } catch (Exception e) { - LOG.error("Ignoring invalid partition plan at offset {} for audit topic '{}'", record.offset(), auditTopicKey, e); - } - } - - private void openConsumerAtLogEnd() throws Exception { + private void openConsumerAtBeginning() throws Exception { consumer = new KafkaConsumer<>(PartitionPlanKafkaConfig.consumerConfig(props, propPrefix, PartitionPlanConstants.PLAN_WATCHER_CONSUMER_GROUP)); TopicPartition partition = new TopicPartition(planTopic, 0); consumer.assign(Collections.singletonList(partition)); - consumer.seekToEnd(Collections.singletonList(partition)); + consumer.seekToBeginning(Collections.singletonList(partition)); } private void closeConsumer() { diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistBootstrap.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistBootstrap.java index 7f9307c34..5c57b9394 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistBootstrap.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistBootstrap.java @@ -34,7 +34,7 @@ import static org.apache.ranger.audit.server.AuditServerConstants.PROP_SUFFIX_ALLOWED_USERS; /** Loads service allowlist entries from ingestor site XML for registry bootstrap and brownfield merge. */ -public final class ServiceAllowlistBootstrap { +public class ServiceAllowlistBootstrap { private static final String ALLOWLIST_SOURCE_SITE_XML = "xml-bootstrap"; private ServiceAllowlistBootstrap() { diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistResolver.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistResolver.java new file mode 100644 index 000000000..8b0b09bce --- /dev/null +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistResolver.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.audit.producer.kafka.partition; + +import org.apache.commons.lang3.StringUtils; + +import java.util.Map; +import java.util.Set; + +/** + * Per-repo audit POST authorization after {@code auth_to_local} mapping. + * + * <p>Distinct from the global allowlist union in {@link AuthToLocalRuleComposer#collectAllowedUserShortNames} + * (which selects Kerberos mapping rules). Checks whether the mapped short name is allowed for the + * specific {@code serviceName} on {@code POST /api/audit/access?serviceName=...}. + * + * <p>Example (dynamic mode, plan has dev_hdfs with hdfs and nn only): + * <ul> + * <li>isAllowed(dev_hdfs, hdfs) returns true (repo in plan)</li> + * <li>isAllowed(dev_hive, hive) returns false unless static XML lists hive for dev_hive + * (repo not in plan -> registry returns null -> XML fallback)</li> + * </ul> + */ +public class ServiceAllowlistResolver { + private ServiceAllowlistResolver() { + } + + /** + * @param serviceName Ranger repo from {@code serviceName} query param (e.g. {@code dev_hdfs}) + * @param userName short name after {@code KerberosName.getShortName()} (e.g. {@code hdfs} from {@code nn/...}) + * @param holder partition-plan registry; per-repo {@code services[serviceName].allowedUsers} + */ + public static boolean isAllowedServiceUser(String serviceName, String userName, boolean dynamicPartitionPlanEnabled, PartitionPlanHolder holder, Map<String, Set<String>> staticAllowedUsersByService) { + if (StringUtils.isBlank(serviceName) || StringUtils.isBlank(userName)) { + return false; + } + if (dynamicPartitionPlanEnabled && holder != null) { + Set<String> registryUsers = holder.getAllowedUsersForService(serviceName); + if (registryUsers != null) { + return registryUsers.contains(userName); + } + } + Set<String> allowedUsers = staticAllowedUsersByService != null ? staticAllowedUsersByService.get(serviceName) : null; + return allowedUsers != null && allowedUsers.contains(userName); + } +} diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/constants/PartitionPlanConstants.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/constants/PartitionPlanConstants.java index 6ef3f6dd4..13436b344 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/constants/PartitionPlanConstants.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/constants/PartitionPlanConstants.java @@ -20,7 +20,7 @@ package org.apache.ranger.audit.producer.kafka.partition.constants; /** Shared constants for the dynamic Kafka partition-plan registry. */ -public final class PartitionPlanConstants { +public class PartitionPlanConstants { /** Version number of the first XML-seeded initial bootstrap plan in an empty registry. */ public static final int INITIAL_PLAN_VERSION = 1; /** {@code updatedBy} value on plans published by {@code PartitionPlanBootstrap}. */ 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 415e93ed2..3f9b98169 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 @@ -27,6 +27,8 @@ 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.PartitionPlanUpdateApplier; +import org.apache.ranger.audit.producer.kafka.partition.ServiceAllowlistResolver; import org.apache.ranger.audit.producer.kafka.partition.exception.PartitionPlanConflictException; import org.apache.ranger.audit.producer.kafka.partition.exception.PartitionPlanException; import org.apache.ranger.audit.producer.kafka.partition.model.OnboardService; @@ -298,8 +300,7 @@ public Response patchPartitionPlan(PartitionPlanReplacement partitionPlanUpdate, ret = authFailure; } else { try { - ret = toSuccessfulPartitionPlanResponse( - partitionPlanService.mergePartitionPlan(partitionPlanUpdate, resolveUpdatedBy(httpRequest))); + ret = toSuccessfulPartitionPlanResponse(partitionPlanService.mergePartitionPlan(partitionPlanUpdate, resolveUpdatedBy(httpRequest))); } catch (PartitionPlanConflictException e) { ret = toPartitionPlanConflictResponse("PATCH /partition-plan", e); } catch (PartitionPlanException e) { @@ -393,16 +394,14 @@ public Response createService(OnboardService request, @Context HttpServletReques ret = authFailure; } else { try { - ret = toSuccessfulPartitionPlanResponse( - partitionPlanService.onboardService(request, resolveUpdatedBy(httpRequest))); + ret = toSuccessfulPartitionPlanResponse(partitionPlanService.onboardService(request, resolveUpdatedBy(httpRequest))); } catch (PartitionPlanConflictException e) { ret = toPartitionPlanConflictResponse("POST /partition-plan/services", e); } catch (PartitionPlanException e) { ret = toPartitionPlanErrorResponse("POST /partition-plan/services", e); } catch (Exception e) { LOG.error("Unexpected error creating service in partition plan", e); - ret = Response.status(Response.Status.INTERNAL_SERVER_ERROR) - .entity(buildErrorResponse("Failed to create service in partition plan")).build(); + ret = Response.status(Response.Status.INTERNAL_SERVER_ERROR).entity(buildErrorResponse("Failed to create service in partition plan")).build(); } } } @@ -571,22 +570,17 @@ private static void initializeAuthToLocal() { * @return true if user is allowed, false otherwise */ private boolean isAllowedServiceUser(String serviceName, String userName) { - if (StringUtils.isBlank(serviceName) || StringUtils.isBlank(userName)) { - return false; - } - - if (partitionPlanService != null && partitionPlanService.isDynamicPartitionPlanEnabled()) { - Set<String> registryUsers = PartitionPlanHolder.getInstance().getAllowedUsersForService(serviceName); - if (registryUsers != null) { - boolean ret = registryUsers.contains(userName); - LOG.debug("isAllowedServiceUser(serviceName={}, userName={}) from registry: ret={}", serviceName, userName, ret); - return ret; - } - } - - Set<String> allowedUsers = allowedServiceUsers.get(serviceName); - boolean ret = allowedUsers != null && allowedUsers.contains(userName); - LOG.debug("isAllowedServiceUser(serviceName={}, userName={}) from static XML: ret={}", serviceName, userName, ret); + boolean dynamicEnabled = partitionPlanService != null + && partitionPlanService.isDynamicPartitionPlanEnabled(); + boolean ret = ServiceAllowlistResolver.isAllowedServiceUser( + serviceName, + userName, + dynamicEnabled, + PartitionPlanHolder.getInstance(), + allowedServiceUsers); + LOG.debug( + "isAllowedServiceUser(serviceName={}, userName={}, dynamic={}): ret={}", + serviceName, userName, dynamicEnabled, ret); return ret; } diff --git a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanHolderTest.java b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanHolderTest.java new file mode 100644 index 000000000..9aa4b0de8 --- /dev/null +++ b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanHolderTest.java @@ -0,0 +1,100 @@ +/* + * 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.ranger.audit.producer.kafka.partition.exception.PartitionPlanException; +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.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static org.junit.jupiter.api.Assertions.assertIterableEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class PartitionPlanHolderTest { + private static final int TOPIC_PARTITIONS = 6; + + @AfterEach + public void tearDown() { + PartitionPlanHolder.getInstance().resetForTests(); + } + + @Test + public void testGetAllowedUsersReturnsNullWhenNoPlanInstalled() { + assertNull(PartitionPlanHolder.getInstance().getAllowedUsersForService("dev_hdfs")); + } + + @Test + public void testGetAllowedUsersReturnsNullWhenServicesBlockMissing() { + PartitionPlan plan = basePlanBuilder() + .build(); + PartitionPlanHolder.getInstance().install(plan, TOPIC_PARTITIONS); + + assertNull(PartitionPlanHolder.getInstance().getAllowedUsersForService("dev_hdfs")); + assertNull(PartitionPlanHolder.getInstance().getAllowedUsersForService("dev_hive")); + } + + @Test + public void testGetAllowedUsersReturnsNullForRepoNotInPartialPlan() { + Map<String, ServiceAllowlistEntry> services = new LinkedHashMap<>(); + services.put("dev_hdfs", ServiceAllowlistEntry.ofUsers("hdfs", "nn")); + PartitionPlan plan = basePlanBuilder().services(services).build(); + PartitionPlanHolder.getInstance().install(plan, TOPIC_PARTITIONS); + + assertIterableEquals(List.of("hdfs", "nn"), PartitionPlanHolder.getInstance().getAllowedUsersForService("dev_hdfs")); + assertNull(PartitionPlanHolder.getInstance().getAllowedUsersForService("dev_hive")); + assertNull(PartitionPlanHolder.getInstance().getAllowedUsersForService("dev_trino")); + } + + @Test + public void testGetAllowedUsersReturnsRegistryUsersForOnboardedRepo() { + Map<String, ServiceAllowlistEntry> services = Map.of("dev_hive", ServiceAllowlistEntry.ofUsers("hive")); + PartitionPlan plan = basePlanBuilder().services(services).build(); + PartitionPlanHolder.getInstance().install(plan, TOPIC_PARTITIONS); + + Set<String> allowed = PartitionPlanHolder.getInstance().getAllowedUsersForService("dev_hive"); + + assertIterableEquals(List.of("hive"), allowed); + assertTrue(allowed != null && !allowed.contains("hdfs")); + } + + @Test + public void testInstallRejectsEmptyServiceAllowlist() { + Map<String, ServiceAllowlistEntry> services = Map.of("dev_hdfs", new ServiceAllowlistEntry(List.of(), "test", null)); + PartitionPlan plan = basePlanBuilder().services(services).build(); + + assertThrows(PartitionPlanException.class, () -> PartitionPlanHolder.getInstance().install(plan, TOPIC_PARTITIONS)); + } + + private static PartitionPlan.Builder basePlanBuilder() { + return PartitionPlan.builder() + .topic("ranger_audits") + .version(1) + .topicPartitionCount(TOPIC_PARTITIONS) + .plugins(Map.of("hdfs", PluginPartitionAssignment.of(0, 1, 2))) + .buffer(PluginPartitionAssignment.of(3, 4, 5)); + } +} diff --git a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanUpdateApplierTest.java b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanUpdateApplierTest.java new file mode 100644 index 000000000..09b264ce2 --- /dev/null +++ b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanUpdateApplierTest.java @@ -0,0 +1,110 @@ +/* + * 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.kafka.clients.consumer.ConsumerRecord; +import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlan; +import org.apache.ranger.audit.producer.kafka.partition.model.PluginPartitionAssignment; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.Map; +import java.util.Properties; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class PartitionPlanUpdateApplierTest { + private static final String AUDIT_TOPIC = "ranger_audits"; + private static final String PLAN_TOPIC = "ranger_audit_partition_plan"; + private static final int KAFKA_PARTITIONS = 6; + + private PartitionPlanHolder holder; + private PartitionPlanUpdateApplier applier; + + @BeforeEach + public void setUp() { + holder = PartitionPlanHolder.getInstance(); + holder.resetForTests(); + holder.install(planWithVersion(1), KAFKA_PARTITIONS); + applier = new PartitionPlanUpdateApplier(new Properties(), AUDIT_TOPIC, holder, () -> KAFKA_PARTITIONS); + } + + @AfterEach + public void tearDown() { + holder.resetForTests(); + } + + @Test + public void testIgnoresWrongRecordKey() { + applier.applyRecordIfNewer(record("other-topic", planWithVersion(2).toJson())); + + assertEquals(1, holder.getLastInstalledVersion()); + } + + @Test + public void testIgnoresSameVersion() { + applier.applyRecordIfNewer(record(AUDIT_TOPIC, planWithVersion(1).toJson())); + + assertEquals(1, holder.getLastInstalledVersion()); + } + + @Test + public void testIgnoresOlderVersion() { + applier.applyRecordIfNewer(record(AUDIT_TOPIC, planWithVersion(0).toJson())); + + assertEquals(1, holder.getLastInstalledVersion()); + } + + @Test + public void testInstallsNewerVersion() { + applier.applyRecordIfNewer(record(AUDIT_TOPIC, planWithVersion(2).toJson())); + + assertEquals(2, holder.getLastInstalledVersion()); + assertEquals(2, holder.getPlan().getVersion()); + } + + @Test + public void testIgnoresInvalidJson() { + applier.applyRecordIfNewer(record(AUDIT_TOPIC, "{not-json")); + + assertEquals(1, holder.getLastInstalledVersion()); + } + + @Test + public void testStartupSyncCanCatchPlanPublishedBeforeWatcherThread() { + applier.applyRecordIfNewer(record(AUDIT_TOPIC, planWithVersion(2).toJson())); + applier.applyRecordIfNewer(record(AUDIT_TOPIC, planWithVersion(3).toJson())); + + assertEquals(3, holder.getLastInstalledVersion()); + } + + private static PartitionPlan planWithVersion(int version) { + return PartitionPlan.builder() + .topic(AUDIT_TOPIC) + .version(version) + .topicPartitionCount(KAFKA_PARTITIONS) + .plugins(Map.of("hdfs", PluginPartitionAssignment.of(0, 1, 2))) + .buffer(PluginPartitionAssignment.of(3, 4, 5)) + .build(); + } + + private static ConsumerRecord<String, String> record(String key, String value) { + return new ConsumerRecord<>(PLAN_TOPIC, 0, 0L, key, value); + } +} diff --git a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistBootstrapTest.java b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistBootstrapTest.java index b019e33de..66298e93f 100644 --- a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistBootstrapTest.java +++ b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistBootstrapTest.java @@ -71,24 +71,31 @@ public void testEnrichServicesFromXmlIfMissingMergesInMemoryOnly() { assertEquals(3, enriched.getVersion()); assertNotNull(enriched.getServices().get("dev_hive")); - assertTrue(PartitionPlanHolder.getInstance().getAllowedUsersForService("dev_hive") == null); + PartitionPlanHolder.getInstance().install(enriched, 6); + assertIterableEquals(List.of("hive"), PartitionPlanHolder.getInstance().getAllowedUsersForService("dev_hive")); } @Test - public void testHolderReturnsRegistryUsersWhenServicesPresent() { + public void testMergeSkippedWhenPlanAlreadyHasPartialServices() { Map<String, ServiceAllowlistEntry> services = new LinkedHashMap<>(); - services.put("dev_hive", ServiceAllowlistEntry.ofUsers("hive")); + services.put("dev_hdfs", ServiceAllowlistEntry.ofUsers("hdfs")); PartitionPlan plan = PartitionPlan.builder() .topic("ranger_audits") - .version(1) + .version(2) .topicPartitionCount(6) .plugins(Map.of("hdfs", PluginPartitionAssignment.of(0, 1, 2))) .buffer(PluginPartitionAssignment.of(3, 4, 5)) .services(services) .build(); - PartitionPlanHolder.getInstance().install(plan, 6); - assertIterableEquals(List.of("hive"), PartitionPlanHolder.getInstance().getAllowedUsersForService("dev_hive")); - assertTrue(PartitionPlanHolder.getInstance().getAllowedUsersForService("dev_trino").isEmpty()); + Properties props = new Properties(); + props.setProperty(PROP_PREFIX_AUDIT_SERVER_SERVICE + "dev_hive" + PROP_SUFFIX_ALLOWED_USERS, "hive"); + + PartitionPlan merged = ServiceAllowlistBootstrap.mergeSiteXmlAllowlistsWhenPlanServicesMissing(plan, props); + + assertEquals(plan, merged); + assertEquals(1, merged.getServices().size()); + assertNotNull(merged.getServices().get("dev_hdfs")); + assertTrue(merged.getServices().get("dev_hive") == null); } } diff --git a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistResolverTest.java b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistResolverTest.java new file mode 100644 index 000000000..755a2a783 --- /dev/null +++ b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistResolverTest.java @@ -0,0 +1,96 @@ +/* + * 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.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.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Set; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class ServiceAllowlistResolverTest { + private static final Map<String, Set<String>> STATIC = Map.of( + "dev_hive", Set.of("hive"), + "dev_hdfs", Set.of("hdfs", "nn")); + + @AfterEach + public void tearDown() { + PartitionPlanHolder.getInstance().resetForTests(); + } + + @Test + public void testRejectsBlankServiceOrUser() { + assertFalse(ServiceAllowlistResolver.isAllowedServiceUser("", "hive", true, PartitionPlanHolder.getInstance(), STATIC)); + assertFalse(ServiceAllowlistResolver.isAllowedServiceUser("dev_hive", "", true, PartitionPlanHolder.getInstance(), STATIC)); + assertFalse(ServiceAllowlistResolver.isAllowedServiceUser(null, "hive", true, PartitionPlanHolder.getInstance(), STATIC)); + } + + @Test + public void testUsesStaticXmlWhenDynamicDisabled() { + assertTrue(ServiceAllowlistResolver.isAllowedServiceUser("dev_hive", "hive", false, PartitionPlanHolder.getInstance(), STATIC)); + assertFalse(ServiceAllowlistResolver.isAllowedServiceUser("dev_hive", "hdfs", false, PartitionPlanHolder.getInstance(), STATIC)); + } + + @Test + public void testUsesRegistryWhenRepoOnboarded() { + installPartialPlan("dev_hdfs", "hdfs"); + + assertTrue(ServiceAllowlistResolver.isAllowedServiceUser("dev_hdfs", "hdfs", true, PartitionPlanHolder.getInstance(), STATIC)); + assertFalse(ServiceAllowlistResolver.isAllowedServiceUser("dev_hdfs", "nn", true, PartitionPlanHolder.getInstance(), STATIC)); + } + + @Test + public void testFallsBackToStaticXmlWhenRepoNotInRegistry() { + installPartialPlan("dev_hdfs", "hdfs"); + + assertTrue(ServiceAllowlistResolver.isAllowedServiceUser("dev_hive", "hive", true, PartitionPlanHolder.getInstance(), STATIC)); + assertFalse(ServiceAllowlistResolver.isAllowedServiceUser("dev_trino", "trino", true, PartitionPlanHolder.getInstance(), STATIC)); + } + + @Test + public void testDeniesWhenUserNotInRegistryAllowlist() { + Map<String, ServiceAllowlistEntry> services = Map.of("dev_hdfs", ServiceAllowlistEntry.ofUsers("hdfs")); + PartitionPlan plan = basePlanBuilder().services(services).build(); + PartitionPlanHolder holder = PartitionPlanHolder.getInstance(); + holder.install(plan, 6); + + assertFalse(ServiceAllowlistResolver.isAllowedServiceUser("dev_hdfs", "nn", true, holder, STATIC)); + } + + private static void installPartialPlan(String repo, String user) { + Map<String, ServiceAllowlistEntry> services = new LinkedHashMap<>(); + services.put(repo, ServiceAllowlistEntry.ofUsers(user)); + PartitionPlanHolder.getInstance().install(basePlanBuilder().services(services).build(), 6); + } + + private static PartitionPlan.Builder basePlanBuilder() { + return PartitionPlan.builder() + .topic("ranger_audits") + .version(1) + .topicPartitionCount(6) + .plugins(Map.of("hdfs", PluginPartitionAssignment.of(0, 1, 2))) + .buffer(PluginPartitionAssignment.of(3, 4, 5)); + } +} diff --git a/dev-support/ranger-docker/scripts/audit-dispatcher/ranger-audit-dispatcher-solr-site.xml b/dev-support/ranger-docker/scripts/audit-dispatcher/ranger-audit-dispatcher-solr-site.xml index 39e5d820d..a5be58db5 100644 --- a/dev-support/ranger-docker/scripts/audit-dispatcher/ranger-audit-dispatcher-solr-site.xml +++ b/dev-support/ranger-docker/scripts/audit-dispatcher/ranger-audit-dispatcher-solr-site.xml @@ -1,30 +1,13 @@ -<!-- - Licensed 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. See accompanying LICENSE file. ---> <configuration> - <!-- SOLR DISPATCHER SERVICE CONFIGURATION --> <property> <name>ranger.audit.dispatcher.solr.class</name> <value>org.apache.ranger.audit.dispatcher.SolrDispatcherManager</value> </property> - <property> <name>log.dir</name> <value>${audit.dispatcher.solr.log.dir}</value> <description>Log directory for Solr dispatcher service</description> </property> - - <!-- User for KERBEROS authentication into solr --> <property> <name>ranger.audit.dispatcher.host</name> <value>ranger-audit-dispatcher-solr.rangernw</value> @@ -33,7 +16,6 @@ - In VM: Use the actual FQDN (e.g., ranger.example.com) </description> </property> - <property> <name>ranger.audit.dispatcher.service.kerberos.principal</name> <value>rangerauditserver/[email protected]</value> @@ -41,7 +23,6 @@ rangerauditserver user kerberos principal for authentication into kafka </description> </property> - <property> <name>ranger.audit.dispatcher.service.kerberos.keytab</name> <value>/etc/keytabs/rangerauditserver.keytab</value> @@ -49,61 +30,46 @@ keytab of the rangerauditserver principal </description> </property> - - <!-- KAFKA DISPATCHER CONFIGURATION --> <property> <name>ranger.audit.dispatcher.kafka.bootstrap.servers</name> <value>ranger-kafka.rangernw:9092</value> </property> - <property> <name>ranger.audit.dispatcher.kafka.topic.name</name> <value>ranger_audits</value> </property> - <property> <name>ranger.audit.dispatcher.kafka.security.protocol</name> <value>SASL_PLAINTEXT</value> </property> - <property> <name>ranger.audit.dispatcher.kafka.sasl.mechanism</name> <value>GSSAPI</value> </property> - - <!-- Solr Dispatcher Threading Configuration --> <property> <name>ranger.audit.dispatcher.thread.count</name> <value>5</value> <description>Number of Solr dispatcher worker threads (higher for indexing throughput)</description> </property> - - <!-- Offset Management --> <property> <name>ranger.audit.dispatcher.offset.commit.strategy</name> <value>batch</value> <description>batch or manual</description> </property> - <property> <name>ranger.audit.dispatcher.offset.commit.interval.ms</name> <value>30000</value> <description>Used only if strategy is 'manual'</description> </property> - <property> <name>ranger.audit.dispatcher.max.poll.records</name> <value>500</value> <description>Maximum records per poll for batch processing</description> </property> - - <!-- Solr Dispatcher Class --> <property> <name>ranger.audit.dispatcher.class</name> <value>org.apache.ranger.audit.dispatcher.kafka.AuditSolrDispatcher</value> </property> - - <!-- SOLR DESTINATION CONFIGURATION --> <property> <name>xasecure.audit.destination.solr.urls</name> <value>http://ranger-solr.rangernw:8983/solr/ranger_audits</value> @@ -113,120 +79,97 @@ configure http://ranger-solr:8983/solr/ranger_audits </description> </property> - <property> <name>xasecure.audit.destination.solr.zookeepers</name> - <value></value> + <value /> <description> Zookeepers for Solr audit destination (SolrCloud mode). example: zookeeper-host1:2181,zookeeper-host2:2181,zookeeper-host3:2181 In Docker, use: ranger-zk:2181/ranger_audits </description> </property> - <property> <name>xasecure.audit.destination.solr.batch.filespool.dir</name> <value>/var/log/audit-dispatcher-solr/spool</value> </property> - - <!-- Solr Kerberos Authentication --> <property> <name>xasecure.audit.destination.solr.force.use.inmemory.jaas.config</name> <value>true</value> </property> - <property> <name>xasecure.audit.jaas.Client.loginModuleName</name> <value>com.sun.security.auth.module.Krb5LoginModule</value> </property> - <property> <name>xasecure.audit.jaas.Client.loginModuleControlFlag</name> <value>required</value> </property> - <property> <name>xasecure.audit.jaas.Client.option.useKeyTab</name> <value>true</value> </property> - <property> <name>xasecure.audit.jaas.Client.option.storeKey</name> <value>true</value> </property> - <property> <name>xasecure.audit.jaas.Client.option.useTicketCache</name> <value>false</value> <description>Use keytab for Solr auth (ticket cache breaks relogin with "No key to store")</description> </property> - <property> <name>xasecure.audit.jaas.Client.option.serviceName</name> <value>solr</value> <description>Kerberos service name for Solr</description> </property> - <property> <name>xasecure.audit.jaas.Client.option.principal</name> <value>rangerauditserver/[email protected]</value> <description>Principal for Solr authentication</description> </property> - <property> <name>xasecure.audit.jaas.Client.option.keyTab</name> <value>/etc/keytabs/rangerauditserver.keytab</value> <description>Keytab for Solr authentication</description> </property> - - <!-- DISPATCHER TYPE SELECTION --> <property> <name>ranger.audit.dispatcher.type</name> <value>solr</value> </property> - <property> <name>ranger.audit.dispatcher.war.file</name> <value>ranger-audit-dispatcher.war</value> </property> - <property> <name>ranger.audit.dispatcher.launcher.class</name> <value>org.apache.ranger.audit.dispatcher.AuditDispatcherLauncher</value> </property> - <property> <name>ranger.audit.dispatcher.main.class</name> <value>org.apache.ranger.audit.dispatcher.AuditDispatcherApplication</value> </property> - <property> <name>ranger.audit.dispatcher.http.port</name> <value>7090</value> </property> - <property> <name>ranger.audit.dispatcher.contextName</name> <value>/</value> </property> - <property> <name>ranger.audit.dispatcher.kafka.group.id</name> - <value></value> + <value /> </property> - <property> <name>ranger.audit.dispatcher.kafka.sasl.kerberos.service.name</name> <value>kafka</value> </property> - <property> <name>ranger.audit.dispatcher.hadoop.security.authentication</name> <value>KERBEROS</value> </property> - <property> <name>ranger.audit.dispatcher.bind.address</name> <value>0.0.0.0</value> </property> -</configuration> +</configuration> \ No newline at end of file diff --git a/dev-support/ranger-docker/scripts/audit/partition-plan-e2e-lib.sh b/dev-support/ranger-docker/scripts/audit/partition-plan-e2e-lib.sh index 30ce6f059..f8bf38864 100755 --- a/dev-support/ranger-docker/scripts/audit/partition-plan-e2e-lib.sh +++ b/dev-support/ranger-docker/scripts/audit/partition-plan-e2e-lib.sh @@ -255,6 +255,7 @@ pp_set_dynamic_enabled() { done if [[ "${value}" == "true" ]]; then pp_ensure_audit_partitioner_for_dynamic "${container}" + pp_prepare_audit_topic_for_greenfield "${container}" fi if [[ "${changed}" == "true" && "${restart}" == "true" ]]; then echo " Restarting ${container} (dynamic=${value})..."
