This is an automated email from the ASF dual-hosted git repository.
kumaab 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 a1f647b0a RANGER-5827: Add shared Kafka client authentication module
with support for both SASL/OAUTHBEARER and SASL/GSSAPI (#1297)
a1f647b0a is described below
commit a1f647b0ad1794c0aca5f6227696c8ad35ab19fb
Author: Abhishek Kumar <[email protected]>
AuthorDate: Fri Oct 16 14:24:41 2026 -0700
RANGER-5827: Add shared Kafka client authentication module with support for
both SASL/OAUTHBEARER and SASL/GSSAPI (#1297)
---
audit-server/audit-common/pom.xml | 5 +
.../ranger/audit/server/AuditServerConstants.java | 11 -
.../ranger/audit/utils/AuditMessageQueueUtils.java | 84 +-----
.../dispatcher/kafka/AuditDispatcherBase.java | 13 +-
.../conf/ranger-audit-dispatcher-hdfs-site.xml | 25 +-
.../ranger-audit-dispatcher-opensearch-site.xml | 25 +-
.../conf/ranger-audit-dispatcher-solr-site.xml | 25 +-
.../ranger/audit/producer/kafka/AuditProducer.java | 13 +-
.../ranger/audit/server/AuditServerConfig.java | 11 +
.../resources/conf/ranger-audit-ingestor-site.xml | 25 +-
.../audit/producer/kafka/AuditProducerTest.java | 65 ++++-
.../ranger-docker/Dockerfile.ranger-audit-ingestor | 7 +-
dev-support/ranger-docker/Dockerfile.ranger-kafka | 2 +-
.../ranger-docker/README-KAFKA-CLIENT-AUTH.md | 52 ++++
dev-support/ranger-docker/README.md | 4 +
.../docker-compose.ranger-audit-service.yml | 7 +
.../ranger-docker/docker-compose.ranger-kafka.yml | 5 +
.../docker-compose.ranger-tagsync.yml | 4 +
.../ranger-audit-dispatcher-hdfs-site.xml | 25 +-
.../ranger-audit-dispatcher-opensearch-site.xml | 25 +-
.../ranger-audit-dispatcher-solr-site.xml | 25 +-
.../audit-ingestor/audit-ingestor-config.py | 105 +++++++
.../configs/ranger-audit-ingestor-site.yaml | 130 +++++++++
.../audit-ingestor/ranger-audit-ingestor.sh | 7 +
.../scripts/kafka/kafka-oauth-token-minter.sh | 65 +++++
.../scripts/kafka/ranger-kafka-setup.sh | 8 +-
.../ranger-docker/scripts/kafka/ranger-kafka.sh | 2 +
.../tagsync/ranger-tagsync-install.properties | 3 +
.../scripts/tagsync/ranger-tagsync.sh | 12 +
distro/src/main/assembly/tagsync.xml | 1 +
pom.xml | 4 +
.../pom.xml | 40 +--
.../auth/RangerKafkaClientSecurityConfig.java | 320 +++++++++++++++++++++
...RangerKafkaOAuthBearerLoginCallbackHandler.java | 225 +++++++++++++++
.../ranger/kafka/auth/RangerOAuthBearerToken.java | 69 +++++
.../auth/RangerKafkaClientSecurityConfigTest.java | 301 +++++++++++++++++++
...erKafkaOAuthBearerLoginCallbackHandlerTest.java | 231 +++++++++++++++
tagsync/conf/templates/installprop2xml.properties | 2 +
tagsync/pom.xml | 5 +
tagsync/scripts/install.properties | 4 +
tagsync/scripts/setup.py | 44 ++-
41 files changed, 1887 insertions(+), 149 deletions(-)
diff --git a/audit-server/audit-common/pom.xml
b/audit-server/audit-common/pom.xml
index 5266db1c2..150203062 100644
--- a/audit-server/audit-common/pom.xml
+++ b/audit-server/audit-common/pom.xml
@@ -36,6 +36,11 @@
<artifactId>kafka-clients</artifactId>
<version>${kafka.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.ranger</groupId>
+ <artifactId>ranger-kafka-client-auth</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.ranger</groupId>
<artifactId>ranger-plugins-common</artifactId>
diff --git
a/audit-server/audit-common/src/main/java/org/apache/ranger/audit/server/AuditServerConstants.java
b/audit-server/audit-common/src/main/java/org/apache/ranger/audit/server/AuditServerConstants.java
index 58b136b85..bcd3c955f 100644
---
a/audit-server/audit-common/src/main/java/org/apache/ranger/audit/server/AuditServerConstants.java
+++
b/audit-server/audit-common/src/main/java/org/apache/ranger/audit/server/AuditServerConstants.java
@@ -34,13 +34,6 @@ private AuditServerConstants() {}
AUDIT-SERVER INGESTOR Configuration
**************************************/
// kafka configuration for audit ingestor
- public static final String JAAS_KRB5_MODULE =
"com.sun.security.auth.module.Krb5LoginModule required";
- public static final String JAAS_USE_KEYTAB =
"useKeyTab=true";
- public static final String JAAS_KEYTAB =
"keyTab=\"";
- public static final String JAAS_STOKE_KEY =
"storeKey=true";
- public static final String JAAS_SERVICE_NAME =
"serviceName=kafka";
- public static final String JAAS_USER_TICKET_CACHE =
"useTicketCache=false";
- public static final String JAAS_PRINCIPAL =
"principal=\"";
public static final String PROP_KAFKA_PROP_PREFIX =
"xasecure.audit.destination.kafka";
public static final String PROP_BOOTSTRAP_SERVERS =
"kafka.bootstrap.servers";
public static final String PROP_TOPIC_NAME =
"kafka.topic.name";
@@ -51,7 +44,6 @@ private AuditServerConstants() {}
public static final String PROP_REQ_TIMEOUT_MS =
"kafka.request.timeout.ms";
public static final String PROP_CONN_MAX_IDEAL_MS =
"kafka.connections.max.idle.ms";
public static final String PROP_SASL_JAAS_CONFIG =
"sasl.jaas.config";
- public static final String PROP_SASL_KERBEROS_SERVICE_NAME =
"sasl.kerberos.service.name";
// kafka topic - ranger_audits configuration
public static final String PROP_TOPIC_PARTITIONS =
"kafka.topic.partitions";
@@ -96,10 +88,7 @@ private AuditServerConstants() {}
// Kafka Topic defaults
public static final String DEFAULT_TOPIC =
"ranger_audits";
- public static final String DEFAULT_SASL_MECHANISM =
"PLAIN";
- public static final String DEFAULT_SECURITY_PROTOCOL =
"PLAINTEXT";
public static final String DEFAULT_SERVICE_NAME =
"kafka";
- public static final String PROP_SECURITY_PROTOCOL_VALUE =
"SASL";
// kafka Offset commit strategies
public static final String PROP_OFFSET_COMMIT_STRATEGY_MANUAL =
"manual";
diff --git
a/audit-server/audit-common/src/main/java/org/apache/ranger/audit/utils/AuditMessageQueueUtils.java
b/audit-server/audit-common/src/main/java/org/apache/ranger/audit/utils/AuditMessageQueueUtils.java
index bfba27260..80512619a 100644
---
a/audit-server/audit-common/src/main/java/org/apache/ranger/audit/utils/AuditMessageQueueUtils.java
+++
b/audit-server/audit-common/src/main/java/org/apache/ranger/audit/utils/AuditMessageQueueUtils.java
@@ -19,7 +19,6 @@
package org.apache.ranger.audit.utils;
-import org.apache.hadoop.security.SecureClientLogin;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.CreatePartitionsResult;
@@ -29,10 +28,10 @@
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.ranger.audit.provider.MiscUtil;
import org.apache.ranger.audit.server.AuditServerConstants;
+import org.apache.ranger.kafka.auth.RangerKafkaClientSecurityConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.File;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
@@ -52,8 +51,6 @@ public static String createAuditsTopicIfNotExists(Properties
props, String propP
String ret = null;
String topicName = MiscUtil.getStringProperty(props,
propPrefix + "." + AuditServerConstants.PROP_TOPIC_NAME,
AuditServerConstants.DEFAULT_TOPIC);
String bootstrapServers = MiscUtil.getStringProperty(props,
propPrefix + "." + AuditServerConstants.PROP_BOOTSTRAP_SERVERS);
- String securityProtocol = MiscUtil.getStringProperty(props,
propPrefix + "." + AuditServerConstants.PROP_SECURITY_PROTOCOL,
AuditServerConstants.DEFAULT_SECURITY_PROTOCOL);
- String saslMechanism = MiscUtil.getStringProperty(props,
propPrefix + "." + AuditServerConstants.PROP_SASL_MECHANISM,
AuditServerConstants.DEFAULT_SASL_MECHANISM);
int connMaxIdleTimeoutMS = MiscUtil.getIntProperty(props,
propPrefix + "." + AuditServerConstants.PROP_CONN_MAX_IDEAL_MS, 10000);
int partitions = getPartitions(props, propPrefix);
short replicationFactor = (short) MiscUtil.getIntProperty(props,
propPrefix + "." + AuditServerConstants.PROP_REPLICATION_FACTOR,
AuditServerConstants.DEFAULT_REPLICATION_FACTOR);
@@ -64,13 +61,8 @@ public static String createAuditsTopicIfNotExists(Properties
props, String propP
Map<String, Object> kafkaProp = new HashMap<>();
kafkaProp.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
bootstrapServers);
- kafkaProp.put("sasl.mechanism", saslMechanism);
- kafkaProp.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG,
securityProtocol);
- if (securityProtocol != null &&
securityProtocol.toUpperCase().contains("SASL")) {
- kafkaProp.put(AuditServerConstants.PROP_SASL_JAAS_CONFIG,
getJAASConfig(props, propPrefix));
-
kafkaProp.put(AuditServerConstants.PROP_SASL_KERBEROS_SERVICE_NAME,
AuditServerConstants.DEFAULT_SERVICE_NAME);
- }
+ RangerKafkaClientSecurityConfig.apply(props, propPrefix, kafkaProp);
kafkaProp.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG,
reqTimeoutMS);
kafkaProp.put(AdminClientConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG,
connMaxIdleTimeoutMS);
@@ -143,78 +135,6 @@ public static String
createAuditsTopicIfNotExists(Properties props, String propP
return ret;
}
- public static String getJAASConfig(Properties props, String propPrefix) {
- // Use ranger service principal and keytab for Kafka authentication
- // This ensures consistent identity across all Ranger services and
destination writes
- String hostName = props.getProperty(propPrefix + "." + "host");
- String principal = props.getProperty(propPrefix + "." +
AuditServerConstants.PROP_AUDIT_SERVICE_PRINCIPAL);
- String keytab = props.getProperty(propPrefix + "." +
AuditServerConstants.PROP_AUDIT_SERVICE_KEYTAB);
-
- AuditServerLogFormatter.builder("Kerberos Configuration")
- .add("Principal (raw)", principal)
- .add("Hostname", hostName)
- .add("Keytab path", keytab)
- .logInfo(LOG);
-
- // Validate keytab file exists and is readable
- if (keytab != null) {
- File keytabFile = new File(keytab);
-
- if (!keytabFile.exists()) {
- LOG.error("ERROR: Keytab file does not exist: {}", keytab);
-
- throw new IllegalStateException("Keytab file not found: " +
keytab);
- } else if (!keytabFile.canRead()) {
- LOG.error("ERROR: Keytab file is not readable: {}", keytab);
-
- throw new IllegalStateException("Keytab file not readable: " +
keytab);
- }
-
- AuditServerLogFormatter.builder("Keytab File Validation")
- .add("Exists", keytabFile.exists())
- .add("Readable", keytabFile.canRead())
- .add("Size (bytes)", keytabFile.length())
- .logInfo(LOG);
- }
-
- try {
- principal = SecureClientLogin.getPrincipal(principal, hostName);
-
- LOG.info("Principal (resolved): {}", principal);
- } catch (Exception e) {
- principal = null;
-
- LOG.error("ERROR: Failed to resolve principal from _HOST
pattern!", e);
- }
-
- if (keytab == null || principal == null) {
- AuditServerLogFormatter.builder("Please configure the following
properties in ranger-audit-ingestor-site.xml:")
- .add(propPrefix + "." +
AuditServerConstants.PROP_AUDIT_SERVICE_PRINCIPAL, "ranger/_HOST@YOUR-REALM")
- .add(propPrefix + "." +
AuditServerConstants.PROP_AUDIT_SERVICE_KEYTAB, "/path/to/ranger.keytab")
- .logError(LOG);
-
- throw new IllegalStateException("Ranger service principal and
keytab must be configured for Kafka authentication. ");
- }
-
- String jaasConfig = new StringBuilder()
- .append(AuditServerConstants.JAAS_KRB5_MODULE).append(" ")
- .append(AuditServerConstants.JAAS_USE_KEYTAB).append(" ")
-
.append(AuditServerConstants.JAAS_KEYTAB).append(keytab).append("\"").append("
")
- .append(AuditServerConstants.JAAS_STOKE_KEY).append(" ")
- .append(AuditServerConstants.JAAS_USER_TICKET_CACHE).append("
")
- .append(AuditServerConstants.JAAS_SERVICE_NAME).append(" ")
-
.append(AuditServerConstants.JAAS_PRINCIPAL).append(principal).append("\";")
- .toString();
-
- AuditServerLogFormatter.builder("JAAS Configuration Generated")
- .add("Principal", principal)
- .add("Keytab", keytab)
- .add("Full JAAS Config", jaasConfig)
- .logInfo(LOG);
-
- return jaasConfig;
- }
-
private static String updateExistingTopicPartitions(AdminClient admin,
String topicName, int partitions) {
LOG.info("==> AuditMessageQueueUtils:updateExistingTopicPartitions()
topic: {}, desired partitions: {}", topicName, partitions);
diff --git
a/audit-server/audit-dispatcher/dispatcher-common/src/main/java/org/apache/ranger/audit/dispatcher/kafka/AuditDispatcherBase.java
b/audit-server/audit-dispatcher/dispatcher-common/src/main/java/org/apache/ranger/audit/dispatcher/kafka/AuditDispatcherBase.java
index ae9327129..29112b612 100644
---
a/audit-server/audit-dispatcher/dispatcher-common/src/main/java/org/apache/ranger/audit/dispatcher/kafka/AuditDispatcherBase.java
+++
b/audit-server/audit-dispatcher/dispatcher-common/src/main/java/org/apache/ranger/audit/dispatcher/kafka/AuditDispatcherBase.java
@@ -18,7 +18,6 @@
*/
package org.apache.ranger.audit.dispatcher.kafka;
-import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -26,7 +25,7 @@
import org.apache.kafka.common.TopicPartition;
import org.apache.ranger.audit.provider.MiscUtil;
import org.apache.ranger.audit.server.AuditServerConstants;
-import org.apache.ranger.audit.utils.AuditMessageQueueUtils;
+import org.apache.ranger.kafka.auth.RangerKafkaClientSecurityConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -67,15 +66,7 @@ public AuditDispatcherBase(Properties props, String
propPrefix, String dispatche
dispatcherProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
Class.forName("org.apache.kafka.common.serialization.StringDeserializer"));
dispatcherProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
Class.forName("org.apache.kafka.common.serialization.StringDeserializer"));
- String securityProtocol = MiscUtil.getStringProperty(props, propPrefix
+ "." + AuditServerConstants.PROP_SECURITY_PROTOCOL,
AuditServerConstants.DEFAULT_SECURITY_PROTOCOL);
- dispatcherProps.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG,
securityProtocol);
-
- dispatcherProps.put("sasl.mechanism",
MiscUtil.getStringProperty(props, propPrefix + "." +
AuditServerConstants.PROP_SASL_MECHANISM,
AuditServerConstants.DEFAULT_SASL_MECHANISM));
-
dispatcherProps.put(AuditServerConstants.PROP_SASL_KERBEROS_SERVICE_NAME,
AuditServerConstants.DEFAULT_SERVICE_NAME);
-
- if
(securityProtocol.toUpperCase().contains(AuditServerConstants.PROP_SECURITY_PROTOCOL_VALUE))
{
- dispatcherProps.put(AuditServerConstants.PROP_SASL_JAAS_CONFIG,
AuditMessageQueueUtils.getJAASConfig(props, propPrefix));
- }
+ RangerKafkaClientSecurityConfig.apply(props, propPrefix,
dispatcherProps);
dispatcherProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,
MiscUtil.getIntProperty(props, propPrefix + "." +
AuditServerConstants.PROP_DISPATCHER_MAX_POLL_RECORDS,
AuditServerConstants.DEFAULT_MAX_POLL_RECORDS));
dispatcherProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"earliest");
diff --git
a/audit-server/audit-dispatcher/dispatcher-hdfs/src/main/resources/conf/ranger-audit-dispatcher-hdfs-site.xml
b/audit-server/audit-dispatcher/dispatcher-hdfs/src/main/resources/conf/ranger-audit-dispatcher-hdfs-site.xml
index 7f1ffc335..45a3cbcca 100644
---
a/audit-server/audit-dispatcher/dispatcher-hdfs/src/main/resources/conf/ranger-audit-dispatcher-hdfs-site.xml
+++
b/audit-server/audit-dispatcher/dispatcher-hdfs/src/main/resources/conf/ranger-audit-dispatcher-hdfs-site.xml
@@ -61,7 +61,30 @@
<property>
<name>ranger.audit.dispatcher.kafka.sasl.mechanism</name>
- <value>GSSAPI</value>
+ <value></value>
+ <description>SASL mechanism for SASL_PLAINTEXT/SASL_SSL: empty =
GSSAPI (default; service.kerberos.principal/keytab with the Krb5LoginModule
options below) or OAUTHBEARER</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kafka.sasl.jaas.option.useKeyTab</name>
+ <value>true</value>
+ <description>GSSAPI Krb5LoginModule option; any option can be set as
ranger.audit.dispatcher.kafka.sasl.jaas.option.<name></description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kafka.sasl.jaas.option.storeKey</name>
+ <value>true</value>
+ </property>
+
+ <property>
+
<name>ranger.audit.dispatcher.kafka.sasl.jaas.option.useTicketCache</name>
+ <value>false</value>
+ </property>
+
+ <property>
+
<name>ranger.audit.dispatcher.kafka.sasl.oauthbearer.token.endpoint.url</name>
+ <value>file:///var/run/secrets/kafka/token</value>
+ <description>OAUTHBEARER only: a file:// JWT (e.g. Kubernetes
projected service-account token) re-read on every login, or an http(s):// token
endpoint for Kafka's client-credentials handler</description>
</property>
<!-- HDFS Dispatcher Threading Configuration -->
diff --git
a/audit-server/audit-dispatcher/dispatcher-opensearch/src/main/resources/conf/ranger-audit-dispatcher-opensearch-site.xml
b/audit-server/audit-dispatcher/dispatcher-opensearch/src/main/resources/conf/ranger-audit-dispatcher-opensearch-site.xml
index b5f01f3d9..41b814348 100644
---
a/audit-server/audit-dispatcher/dispatcher-opensearch/src/main/resources/conf/ranger-audit-dispatcher-opensearch-site.xml
+++
b/audit-server/audit-dispatcher/dispatcher-opensearch/src/main/resources/conf/ranger-audit-dispatcher-opensearch-site.xml
@@ -62,7 +62,30 @@
<property>
<name>ranger.audit.dispatcher.kafka.sasl.mechanism</name>
- <value>GSSAPI</value>
+ <value></value>
+ <description>SASL mechanism for SASL_PLAINTEXT/SASL_SSL: empty =
GSSAPI (default; service.kerberos.principal/keytab with the Krb5LoginModule
options below) or OAUTHBEARER</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kafka.sasl.jaas.option.useKeyTab</name>
+ <value>true</value>
+ <description>GSSAPI Krb5LoginModule option; any option can be set as
ranger.audit.dispatcher.kafka.sasl.jaas.option.<name></description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kafka.sasl.jaas.option.storeKey</name>
+ <value>true</value>
+ </property>
+
+ <property>
+
<name>ranger.audit.dispatcher.kafka.sasl.jaas.option.useTicketCache</name>
+ <value>false</value>
+ </property>
+
+ <property>
+
<name>ranger.audit.dispatcher.kafka.sasl.oauthbearer.token.endpoint.url</name>
+ <value>file:///var/run/secrets/kafka/token</value>
+ <description>OAUTHBEARER token: a file:// JWT (e.g. Kubernetes
projected service-account token) re-read on every login, or an http(s):// token
endpoint for Kafka's client-credentials handler</description>
</property>
<!-- OpenSearch Dispatcher Threading Configuration -->
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 405e1d7ed..16d0c6d46 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
@@ -62,7 +62,30 @@
<property>
<name>ranger.audit.dispatcher.kafka.sasl.mechanism</name>
- <value>GSSAPI</value>
+ <value></value>
+ <description>SASL mechanism for SASL_PLAINTEXT/SASL_SSL: empty =
GSSAPI (default; service.kerberos.principal/keytab with the Krb5LoginModule
options below) or OAUTHBEARER</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kafka.sasl.jaas.option.useKeyTab</name>
+ <value>true</value>
+ <description>GSSAPI Krb5LoginModule option; any option can be set as
ranger.audit.dispatcher.kafka.sasl.jaas.option.<name></description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kafka.sasl.jaas.option.storeKey</name>
+ <value>true</value>
+ </property>
+
+ <property>
+
<name>ranger.audit.dispatcher.kafka.sasl.jaas.option.useTicketCache</name>
+ <value>false</value>
+ </property>
+
+ <property>
+
<name>ranger.audit.dispatcher.kafka.sasl.oauthbearer.token.endpoint.url</name>
+ <value>file:///var/run/secrets/kafka/token</value>
+ <description>OAUTHBEARER token: a file:// JWT (e.g. Kubernetes
projected service-account token) re-read on every login, or an http(s):// token
endpoint for Kafka's client-credentials handler</description>
</property>
<!-- Solr Dispatcher Threading Configuration -->
diff --git
a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/AuditProducer.java
b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/AuditProducer.java
index 497848012..856a26c93 100644
---
a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/AuditProducer.java
+++
b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/AuditProducer.java
@@ -19,7 +19,6 @@
package org.apache.ranger.audit.producer.kafka;
-import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
@@ -27,8 +26,8 @@
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.ranger.audit.provider.MiscUtil;
import org.apache.ranger.audit.server.AuditServerConstants;
-import org.apache.ranger.audit.utils.AuditMessageQueueUtils;
import org.apache.ranger.audit.utils.AuditServerLogFormatter;
+import org.apache.ranger.kafka.auth.RangerKafkaClientSecurityConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -167,15 +166,7 @@ static Properties createProducerConfig(Properties props,
String propPrefix) thro
producerPrefix +
AuditServerConstants.PROP_PRODUCER_BATCH_SEND_TIMEOUT_MS,
AuditServerConstants.DEFAULT_PRODUCER_BATCH_SEND_TIMEOUT_MS);
- String securityProtocol = MiscUtil.getStringProperty(props, propPrefix
+ "." + AuditServerConstants.PROP_SECURITY_PROTOCOL,
AuditServerConstants.DEFAULT_SECURITY_PROTOCOL);
- producerProps.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG,
securityProtocol);
-
- producerProps.put(AuditServerConstants.PROP_SASL_MECHANISM,
MiscUtil.getStringProperty(props, propPrefix + "." +
AuditServerConstants.PROP_SASL_MECHANISM,
AuditServerConstants.DEFAULT_SASL_MECHANISM));
-
producerProps.put(AuditServerConstants.PROP_SASL_KERBEROS_SERVICE_NAME,
AuditServerConstants.DEFAULT_SERVICE_NAME);
-
- if
(securityProtocol.toUpperCase().contains(AuditServerConstants.PROP_SECURITY_PROTOCOL_VALUE))
{
- producerProps.put(AuditServerConstants.PROP_SASL_JAAS_CONFIG,
AuditMessageQueueUtils.getJAASConfig(props, propPrefix));
- }
+ RangerKafkaClientSecurityConfig.apply(props, propPrefix,
producerProps);
String configuredPlugins = MiscUtil.getStringProperty(props,
propPrefix + "." + AuditServerConstants.PROP_CONFIGURED_PLUGINS, "");
if (configuredPlugins != null && !configuredPlugins.trim().isEmpty()) {
diff --git
a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/server/AuditServerConfig.java
b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/server/AuditServerConfig.java
index 185ca1000..12d38843e 100644
---
a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/server/AuditServerConfig.java
+++
b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/server/AuditServerConfig.java
@@ -31,6 +31,10 @@ public class AuditServerConfig extends AuditConfig {
private static final String CONFIG_FILE_PATH =
"conf/ranger-audit-ingestor-site.xml";
+ /* optional explicit site XML (e.g.
-Daudit.config=/opt/ranger/audit-ingestor/conf/ranger-audit-ingestor-site.xml);
+ * loaded after the packaged default so its values win */
+ public static final String PROP_AUDIT_CONFIG_FILE = "audit.config";
+
private static volatile AuditServerConfig sInstance;
private AuditServerConfig() {
@@ -66,6 +70,13 @@ private boolean addAuditServerResources() {
ret = false;
}
+ String configFile = System.getProperty(PROP_AUDIT_CONFIG_FILE);
+
+ if (configFile != null && !configFile.trim().isEmpty() &&
!addAuditResource(configFile.trim(), true)) {
+ LOG.error("Could not load configuration from -D{}={}",
PROP_AUDIT_CONFIG_FILE, configFile);
+ ret = false;
+ }
+
LOG.debug("<== AuditServerConfig.addAuditServerResources(),
result={}", ret);
return ret;
diff --git
a/audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml
b/audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml
index 94e7164a3..d3bf921cf 100644
---
a/audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml
+++
b/audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml
@@ -435,7 +435,30 @@
<property>
<name>ranger.audit.ingestor.kafka.sasl.mechanism</name>
- <value>GSSAPI</value>
+ <value></value>
+ <description>SASL mechanism for SASL_PLAINTEXT/SASL_SSL: empty =
GSSAPI (default; service.kerberos.principal/keytab with the Krb5LoginModule
options below) or OAUTHBEARER</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.ingestor.kafka.sasl.jaas.option.useKeyTab</name>
+ <value>true</value>
+ <description>GSSAPI Krb5LoginModule option; any option can be set as
ranger.audit.ingestor.kafka.sasl.jaas.option.<name></description>
+ </property>
+
+ <property>
+ <name>ranger.audit.ingestor.kafka.sasl.jaas.option.storeKey</name>
+ <value>true</value>
+ </property>
+
+ <property>
+
<name>ranger.audit.ingestor.kafka.sasl.jaas.option.useTicketCache</name>
+ <value>false</value>
+ </property>
+
+ <property>
+
<name>ranger.audit.ingestor.kafka.sasl.oauthbearer.token.endpoint.url</name>
+ <value>file:///var/run/secrets/kafka/token</value>
+ <description>OAUTHBEARER token: a file:// JWT (e.g. Kubernetes
projected service-account token) re-read on every login, or an http(s):// token
endpoint for Kafka's client-credentials handler</description>
</property>
<property>
diff --git
a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/AuditProducerTest.java
b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/AuditProducerTest.java
index 356cd6111..1a31f6b0a 100644
---
a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/AuditProducerTest.java
+++
b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/AuditProducerTest.java
@@ -17,19 +17,29 @@
package org.apache.ranger.audit.producer.kafka;
-import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.config.SaslConfigs;
import org.apache.ranger.audit.server.AuditServerConstants;
+import org.apache.ranger.kafka.auth.RangerKafkaClientSecurityConfig;
+import org.apache.ranger.kafka.auth.RangerKafkaOAuthBearerLoginCallbackHandler;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.util.Properties;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
public class AuditProducerTest {
private static final String PROP_PREFIX =
AuditServerConstants.PROP_PREFIX_AUDIT_SERVER + "kafka";
+ @TempDir
+ Path tempDir;
+
private static void assertProducerProp(Properties props, String key,
Object expected) {
assertEquals(String.valueOf(expected), String.valueOf(props.get(key)));
}
@@ -54,10 +64,61 @@ public void testCreateProducerConfigDefaults() throws
Exception {
assertProducerProp(producerProps,
ProducerConfig.MAX_REQUEST_SIZE_CONFIG,
AuditServerConstants.DEFAULT_PRODUCER_MAX_REQUEST_SIZE);
assertProducerProp(producerProps, ProducerConfig.MAX_BLOCK_MS_CONFIG,
AuditServerConstants.DEFAULT_PRODUCER_MAX_BLOCK_MS);
assertProducerProp(producerProps,
ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,
AuditServerConstants.DEFAULT_PRODUCER_REQUEST_TIMEOUT_MS);
- assertEquals("PLAINTEXT",
producerProps.get(AdminClientConfig.SECURITY_PROTOCOL_CONFIG));
+ assertEquals("PLAINTEXT",
producerProps.get(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG));
+ assertFalse(producerProps.containsKey(SaslConfigs.SASL_MECHANISM));
+ assertFalse(producerProps.containsKey(SaslConfigs.SASL_JAAS_CONFIG));
assertFalse(producerProps.containsKey(ProducerConfig.PARTITIONER_CLASS_CONFIG));
}
+ @Test
+ public void
testCreateProducerConfigSaslDefaultGssapiKeytabLoginRequiresKeytab() {
+ Properties props = new Properties();
+ props.setProperty(PROP_PREFIX + "." +
AuditServerConstants.PROP_BOOTSTRAP_SERVERS, "kafka:9092");
+ props.setProperty(PROP_PREFIX + "." +
AuditServerConstants.PROP_SECURITY_PROTOCOL, "SASL_PLAINTEXT");
+ props.setProperty(PROP_PREFIX + "." +
RangerKafkaClientSecurityConfig.PROP_JAAS_OPTION_PREFIX + "useKeyTab", "true");
+
+ assertThrows(IllegalStateException.class, () ->
AuditProducer.createProducerConfig(props, PROP_PREFIX));
+ }
+
+ @Test
+ public void testCreateProducerConfigSaslDefaultGssapiKeytab() throws
Exception {
+ Path keytab = Files.createFile(tempDir.resolve("ranger.keytab"));
+ Properties props = new Properties();
+ props.setProperty(PROP_PREFIX + "." +
AuditServerConstants.PROP_BOOTSTRAP_SERVERS, "kafka:9092");
+ props.setProperty(PROP_PREFIX + "." +
AuditServerConstants.PROP_SECURITY_PROTOCOL, "SASL_PLAINTEXT");
+ props.setProperty(PROP_PREFIX + "." +
AuditServerConstants.PROP_AUDIT_SERVICE_PRINCIPAL,
"[email protected]");
+ props.setProperty(PROP_PREFIX + "." +
AuditServerConstants.PROP_AUDIT_SERVICE_KEYTAB, keytab.toString());
+ props.setProperty(PROP_PREFIX + "." +
RangerKafkaClientSecurityConfig.PROP_JAAS_OPTION_PREFIX + "useKeyTab", "true");
+ props.setProperty(PROP_PREFIX + "." +
RangerKafkaClientSecurityConfig.PROP_JAAS_OPTION_PREFIX + "storeKey", "true");
+ props.setProperty(PROP_PREFIX + "." +
RangerKafkaClientSecurityConfig.PROP_JAAS_OPTION_PREFIX + "useTicketCache",
"false");
+
+ Properties producerProps = AuditProducer.createProducerConfig(props,
PROP_PREFIX);
+
+ assertEquals("GSSAPI", producerProps.get(SaslConfigs.SASL_MECHANISM));
+ assertEquals("kafka",
producerProps.get(SaslConfigs.SASL_KERBEROS_SERVICE_NAME));
+ assertEquals(RangerKafkaClientSecurityConfig.KRB5_LOGIN_MODULE + "
required keyTab=\"" + keytab + "\" serviceName=\"kafka\"
principal=\"[email protected]\" storeKey=\"true\"
useKeyTab=\"true\" useTicketCache=\"false\";",
+ producerProps.get(SaslConfigs.SASL_JAAS_CONFIG));
+ }
+
+ @Test
+ public void testCreateProducerConfigSaslOAuthBearerFileToken() throws
Exception {
+ Path token = Files.createFile(tempDir.resolve("kafka-token"));
+ Properties props = new Properties();
+ props.setProperty(PROP_PREFIX + "." +
AuditServerConstants.PROP_BOOTSTRAP_SERVERS, "kafka:9093");
+ props.setProperty(PROP_PREFIX + "." +
AuditServerConstants.PROP_SECURITY_PROTOCOL, "SASL_SSL");
+ props.setProperty(PROP_PREFIX + "." +
AuditServerConstants.PROP_SASL_MECHANISM, "OAUTHBEARER");
+ props.setProperty(PROP_PREFIX + "." +
RangerKafkaClientSecurityConfig.PROP_OAUTH_TOKEN_ENDPOINT_URL,
token.toUri().toString());
+
+ Properties producerProps = AuditProducer.createProducerConfig(props,
PROP_PREFIX);
+
+ assertEquals("SASL_SSL",
producerProps.get(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG));
+ assertEquals("OAUTHBEARER",
producerProps.get(SaslConfigs.SASL_MECHANISM));
+ assertEquals(RangerKafkaClientSecurityConfig.OAUTHBEARER_LOGIN_MODULE
+ " required;", producerProps.get(SaslConfigs.SASL_JAAS_CONFIG));
+ assertEquals(token.toUri().toString(),
producerProps.get(SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL));
+
assertEquals(RangerKafkaOAuthBearerLoginCallbackHandler.class.getName(),
producerProps.get(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS));
+
assertFalse(producerProps.containsKey(SaslConfigs.SASL_KERBEROS_SERVICE_NAME));
+ }
+
@Test
public void testCreateProducerConfigCustomTuning() throws Exception {
Properties props = new Properties();
diff --git a/dev-support/ranger-docker/Dockerfile.ranger-audit-ingestor
b/dev-support/ranger-docker/Dockerfile.ranger-audit-ingestor
index a9a9d1b31..f5223946d 100644
--- a/dev-support/ranger-docker/Dockerfile.ranger-audit-ingestor
+++ b/dev-support/ranger-docker/Dockerfile.ranger-audit-ingestor
@@ -27,6 +27,9 @@ RUN apt-get update && apt-get install -y \
net-tools \
&& rm -rf /var/lib/apt/lists/*
+# audit-ingestor-config.py renders conf/ranger-audit-ingestor-site.xml from
configs/*.yaml
+RUN pip install --no-cache-dir pyyaml
+
# Volume for keytabs
VOLUME /etc/keytabs
@@ -39,6 +42,7 @@ COPY ./dist/ranger-${RANGER_VERSION}-audit-ingestor.tar.gz
/tmp/
# Copy scripts and Kerberos configuration
COPY ./scripts/audit-ingestor/service-check-functions.sh ${RANGER_SCRIPTS}/
COPY ./scripts/audit-ingestor/ranger-audit-ingestor.sh
${RANGER_SCRIPTS}/
+COPY ./scripts/audit-ingestor/audit-ingestor-config.py
${RANGER_SCRIPTS}/
COPY ./scripts/kdc/krb5.conf /etc/krb5.conf
# Extract and setup
@@ -51,7 +55,7 @@ RUN tar xzf
/tmp/ranger-${RANGER_VERSION}-audit-ingestor.tar.gz -C /opt/ranger/a
ln -sf /var/log/ranger/audit-ingestor /opt/ranger/audit-ingestor/logs && \
chown -R ranger:ranger /opt/ranger/audit-ingestor/ /var/log/ranger/
${RANGER_SCRIPTS} && \
chmod -R 755 /opt/ranger/audit-ingestor && \
- chmod 755 ${RANGER_SCRIPTS}/ranger-audit-ingestor.sh
+ chmod 755 ${RANGER_SCRIPTS}/ranger-audit-ingestor.sh
${RANGER_SCRIPTS}/audit-ingestor-config.py
# Set Java home
ENV JAVA_HOME=/opt/java/openjdk
@@ -60,6 +64,7 @@ ENV PATH=$JAVA_HOME/bin:$PATH
# Environment variables for audit ingestor
ENV AUDIT_INGESTOR_HOME_DIR=/opt/ranger/audit-ingestor
ENV AUDIT_INGESTOR_CONF_DIR=/opt/ranger/audit-ingestor/conf
+ENV AUDIT_INGESTOR_CONFIGS=/opt/ranger/audit-ingestor/configs
ENV AUDIT_INGESTOR_LOG_DIR=/var/log/ranger/audit-ingestor
ENV RANGER_USER=ranger
diff --git a/dev-support/ranger-docker/Dockerfile.ranger-kafka
b/dev-support/ranger-docker/Dockerfile.ranger-kafka
index 69d58a59b..b95116a58 100644
--- a/dev-support/ranger-docker/Dockerfile.ranger-kafka
+++ b/dev-support/ranger-docker/Dockerfile.ranger-kafka
@@ -32,7 +32,7 @@ RUN tar xvfz
/home/ranger/dist/kafka_2.12-${KAFKA_VERSION}.tgz --directory=/opt/
ln -s /opt/ranger/ranger-${KAFKA_PLUGIN_VERSION}-kafka-plugin
/opt/ranger/ranger-kafka-plugin && \
rm -f /home/ranger/dist/ranger-${KAFKA_PLUGIN_VERSION}-kafka-plugin.tar.gz
&& \
rm -f /opt/ranger/ranger-kafka-plugin/install.properties && \
- chmod 744 ${RANGER_SCRIPTS}/ranger-kafka-setup.sh
${RANGER_SCRIPTS}/ranger-kafka.sh
+ chmod 744 ${RANGER_SCRIPTS}/ranger-kafka-setup.sh
${RANGER_SCRIPTS}/ranger-kafka.sh ${RANGER_SCRIPTS}/kafka-oauth-token-minter.sh
ENV KAFKA_HOME=/opt/kafka
ENV
PATH=/usr/java/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/opt/kafka/bin
diff --git a/dev-support/ranger-docker/README-KAFKA-CLIENT-AUTH.md
b/dev-support/ranger-docker/README-KAFKA-CLIENT-AUTH.md
new file mode 100644
index 000000000..1214f96da
--- /dev/null
+++ b/dev-support/ranger-docker/README-KAFKA-CLIENT-AUTH.md
@@ -0,0 +1,52 @@
+<!--
+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.
+-->
+
+# Kafka client authentication
+
+The audit ingestor, the audit dispatchers and tagsync authenticate to Kafka
through the shared module `ranger-kafka-client-auth`
+
+## Mechanisms
+
+| `kafka.security.protocol` | `kafka.sasl.mechanism` | Credential |
+|---|---|---|
+| `PLAINTEXT` (default), `SSL` | none | none |
+| `SASL_PLAINTEXT`, `SASL_SSL` | `GSSAPI` (default when empty) |
`service.kerberos.principal` and `service.kerberos.keytab` |
+| `SASL_PLAINTEXT`, `SASL_SSL` | `OAUTHBEARER` | JWT from
`kafka.sasl.oauthbearer.token.endpoint.url` |
+| `SASL_PLAINTEXT`, `SASL_SSL` | other (`PLAIN`, `SCRAM-*`) |
`kafka.sasl.jaas.loginModuleName` and `kafka.sasl.jaas.option.*`, or
`kafka.sasl.jaas.config` |
+
+- OAUTHBEARER with a `file://` URL uses
`RangerKafkaOAuthBearerLoginCallbackHandler`, which re-reads the
+ token on every login and refresh, so a token rotated in place (for example a
Kubernetes projected
+ service-account token) is picked up without a restart. An `http(s)://` URL
uses Kafka's own
+ client-credentials handler. The broker validates the token; its `sub` is the
Kafka principal.
+- JAAS is built from `kafka.sasl.jaas.loginModuleName`,
`kafka.sasl.jaas.loginModuleControlFlag` and any
+ `kafka.sasl.jaas.option.<name>` under the service prefix. For GSSAPI the
service principal and keytab are passed
+ as `principal` and `keyTab`; every other Krb5LoginModule option comes from
configuration, and the shipped site
+ XML sets `useKeyTab=true`, `storeKey=true`, `useTicketCache=false`.
+- Any `kafka.sasl.*` or `kafka.ssl.*` property under the service prefix is
passed to the Kafka client as is.
+
+## Docker Setup
+
+With `KERBEROS_ENABLED=true` the broker accepts both GSSAPI and OAUTHBEARER,
and the clients use OAUTHBEARER:
+`ranger-kafka` runs `scripts/kafka/kafka-oauth-token-minter.sh`, which writes
unsigned dev JWTs for
+`rangerauditserver` and `rangertagsync` to the `ranger-kafka-oauth-tokens`
volume (`/etc/kafka-oauth`) and
+rewrites them every two minutes with a ten minute lifetime; the broker accepts
them with Kafka's unsecured
+validator. Dev only, never use this outside docker.
+
+To switch a client to GSSAPI set its mechanism to `GSSAPI` in
`scripts/audit-ingestor/configs/ranger-audit-ingestor-site.yaml`,
+`scripts/audit-dispatcher/ranger-audit-dispatcher-<type>-site.xml` or
`scripts/tagsync/ranger-tagsync-install.properties`;
+the service keytabs are already configured.
diff --git a/dev-support/ranger-docker/README.md
b/dev-support/ranger-docker/README.md
index 99e1af1c9..1cdd91efb 100644
--- a/dev-support/ranger-docker/README.md
+++ b/dev-support/ranger-docker/README.md
@@ -98,6 +98,10 @@ docker compose --profile ${AUDIT_DESTINATIONS} -f
docker-compose.ranger.yml -f d
# Ranger Admin can be accessed at http://localhost:6080 (admin/rangerR0cks!)
~~~
+
+#### Kafka client authentication for the audit pipeline and tagsync
+See [README-KAFKA-CLIENT-AUTH.md](README-KAFKA-CLIENT-AUTH.md) for the
supported mechanisms (GSSAPI by default, OAUTHBEARER used here) and how this
setup exercises them.
+
#### Bring up hive container
~~~
docker compose --profile ${AUDIT_DESTINATIONS} -f docker-compose.ranger.yml -f
docker-compose.ranger-audit-service.yml -f docker-compose.ranger-hadoop.yml -f
docker-compose.ranger-hive.yml up -d
diff --git a/dev-support/ranger-docker/docker-compose.ranger-audit-service.yml
b/dev-support/ranger-docker/docker-compose.ranger-audit-service.yml
index 9806b9dcc..88a3e4617 100644
--- a/dev-support/ranger-docker/docker-compose.ranger-audit-service.yml
+++ b/dev-support/ranger-docker/docker-compose.ranger-audit-service.yml
@@ -200,6 +200,9 @@ services:
KAFKA_BOOTSTRAP_SERVERS: ranger-kafka.rangernw:9092
volumes:
- ./dist/keytabs/ranger-audit-ingestor:/etc/keytabs
+ - ranger-kafka-oauth-tokens:/etc/kafka-oauth:ro
+ # audit-ingestor-config.py renders conf/ranger-audit-ingestor-site.xml
from configs/*.yaml
+ - ./scripts/audit-ingestor/configs:/opt/ranger/audit-ingestor/configs:ro
- ranger-audit-ingestor-logs:/var/log/ranger/audit-ingestor
- ranger-audit-ingestor-spool:/var/log/ranger/audit-ingestor/audit
healthcheck:
@@ -241,6 +244,7 @@ services:
SOLR_URL: http://ranger-solr.rangernw:8983/solr
volumes:
- ./dist/keytabs/ranger-audit-dispatcher-solr:/etc/keytabs
+ - ranger-kafka-oauth-tokens:/etc/kafka-oauth:ro
- ranger-audit-dispatcher-solr-logs:/var/log/ranger/audit-dispatcher/solr
-
./scripts/audit-dispatcher/ranger-audit-dispatcher-solr-site.xml:/opt/ranger/audit-dispatcher/conf/ranger-audit-dispatcher-solr-site.xml
@@ -276,6 +280,7 @@ services:
OPENSEARCH_URL: http://ranger-opensearch.rangernw:9200
volumes:
- ./dist/keytabs/ranger-audit-dispatcher-opensearch:/etc/keytabs
+ - ranger-kafka-oauth-tokens:/etc/kafka-oauth:ro
-
ranger-audit-dispatcher-opensearch-logs:/var/log/ranger/audit-dispatcher/opensearch
-
./scripts/audit-dispatcher/ranger-audit-dispatcher-opensearch-site.xml:/opt/ranger/audit-dispatcher/conf/ranger-audit-dispatcher-opensearch-site.xml
@@ -309,10 +314,12 @@ services:
HDFS_NAMENODE: ranger-hadoop.rangernw:9000
volumes:
- ./dist/keytabs/ranger-audit-dispatcher-hdfs:/etc/keytabs
+ - ranger-kafka-oauth-tokens:/etc/kafka-oauth:ro
- ranger-audit-dispatcher-hdfs-logs:/var/log/ranger/audit-dispatcher/hdfs
-
./scripts/audit-dispatcher/ranger-audit-dispatcher-hdfs-site.xml:/opt/ranger/audit-dispatcher/conf/ranger-audit-dispatcher-hdfs-site.xml
volumes:
+ ranger-kafka-oauth-tokens:
ranger-audit-ingestor-logs:
ranger-audit-ingestor-spool:
ranger-audit-dispatcher-solr-logs:
diff --git a/dev-support/ranger-docker/docker-compose.ranger-kafka.yml
b/dev-support/ranger-docker/docker-compose.ranger-kafka.yml
index 7e5a9f40c..db3286622 100644
--- a/dev-support/ranger-docker/docker-compose.ranger-kafka.yml
+++ b/dev-support/ranger-docker/docker-compose.ranger-kafka.yml
@@ -21,6 +21,8 @@ services:
-
./scripts/kafka/kafka-client.properties:/opt/kafka/config/kafka-client.properties:ro
- ./dist/version:/home/ranger/dist/version:ro
-
./scripts/kafka/ranger-kafka-plugin-install.properties:/opt/ranger/ranger-kafka-plugin/install.properties
+ # dev SASL/OAUTHBEARER tokens minted by kafka-oauth-token-minter.sh,
read by the Ranger Kafka clients
+ - ranger-kafka-oauth-tokens:/etc/kafka-oauth
stdin_open: true
tty: true
networks:
@@ -60,6 +62,9 @@ services:
KAFKA_PLUGIN_VERSION: ${KAFKA_PLUGIN_VERSION}
KERBEROS_ENABLED: ${KERBEROS_ENABLED}
+volumes:
+ ranger-kafka-oauth-tokens:
+
networks:
ranger:
name: rangernw
diff --git a/dev-support/ranger-docker/docker-compose.ranger-tagsync.yml
b/dev-support/ranger-docker/docker-compose.ranger-tagsync.yml
index 01c2b4802..ac7eb1a34 100644
--- a/dev-support/ranger-docker/docker-compose.ranger-tagsync.yml
+++ b/dev-support/ranger-docker/docker-compose.ranger-tagsync.yml
@@ -18,6 +18,7 @@ services:
- ./dist/version:/home/ranger/dist/version:ro
-
./scripts/tagsync/ranger-tagsync-tags.json:/opt/ranger/tagsync/data/tags.json:ro
-
./scripts/tagsync/ranger-tagsync-install.properties:/opt/ranger/tagsync/install.properties
+ - ranger-kafka-oauth-tokens:/etc/kafka-oauth:ro
stdin_open: true
tty: true
networks:
@@ -37,6 +38,9 @@ services:
- RANGER_JVM_METASPACE
- RANGER_JVM_MAX_METASPACE
+volumes:
+ ranger-kafka-oauth-tokens:
+
networks:
ranger:
name: rangernw
diff --git
a/dev-support/ranger-docker/scripts/audit-dispatcher/ranger-audit-dispatcher-hdfs-site.xml
b/dev-support/ranger-docker/scripts/audit-dispatcher/ranger-audit-dispatcher-hdfs-site.xml
index 03cea2fba..42071f4af 100644
---
a/dev-support/ranger-docker/scripts/audit-dispatcher/ranger-audit-dispatcher-hdfs-site.xml
+++
b/dev-support/ranger-docker/scripts/audit-dispatcher/ranger-audit-dispatcher-hdfs-site.xml
@@ -68,7 +68,30 @@
<property>
<name>ranger.audit.dispatcher.kafka.sasl.mechanism</name>
- <value>GSSAPI</value>
+ <value>OAUTHBEARER</value>
+ <description>OAUTHBEARER; GSSAPI is also supported and uses
service.kerberos.principal/keytab</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kafka.sasl.jaas.option.useKeyTab</name>
+ <value>true</value>
+ <description>GSSAPI Krb5LoginModule option; any option can be set as
ranger.audit.dispatcher.kafka.sasl.jaas.option.<name></description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kafka.sasl.jaas.option.storeKey</name>
+ <value>true</value>
+ </property>
+
+ <property>
+
<name>ranger.audit.dispatcher.kafka.sasl.jaas.option.useTicketCache</name>
+ <value>false</value>
+ </property>
+
+ <property>
+
<name>ranger.audit.dispatcher.kafka.sasl.oauthbearer.token.endpoint.url</name>
+ <value>file:///etc/kafka-oauth/rangerauditserver.token</value>
+ <description>Dev token minted by the ranger-kafka container; re-read
on every login so rotation is picked up</description>
</property>
<!-- HDFS Dispatcher Threading Configuration -->
diff --git
a/dev-support/ranger-docker/scripts/audit-dispatcher/ranger-audit-dispatcher-opensearch-site.xml
b/dev-support/ranger-docker/scripts/audit-dispatcher/ranger-audit-dispatcher-opensearch-site.xml
index 2ef249569..cb985ad34 100644
---
a/dev-support/ranger-docker/scripts/audit-dispatcher/ranger-audit-dispatcher-opensearch-site.xml
+++
b/dev-support/ranger-docker/scripts/audit-dispatcher/ranger-audit-dispatcher-opensearch-site.xml
@@ -56,7 +56,30 @@
<property>
<name>ranger.audit.dispatcher.kafka.sasl.mechanism</name>
- <value>GSSAPI</value>
+ <value>OAUTHBEARER</value>
+ <description>OAUTHBEARER; GSSAPI is also supported and uses
service.kerberos.principal/keytab</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kafka.sasl.jaas.option.useKeyTab</name>
+ <value>true</value>
+ <description>GSSAPI Krb5LoginModule option; any option can be set as
ranger.audit.dispatcher.kafka.sasl.jaas.option.<name></description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kafka.sasl.jaas.option.storeKey</name>
+ <value>true</value>
+ </property>
+
+ <property>
+
<name>ranger.audit.dispatcher.kafka.sasl.jaas.option.useTicketCache</name>
+ <value>false</value>
+ </property>
+
+ <property>
+
<name>ranger.audit.dispatcher.kafka.sasl.oauthbearer.token.endpoint.url</name>
+ <value>file:///etc/kafka-oauth/rangerauditserver.token</value>
+ <description>Dev token minted by the ranger-kafka container; re-read
on every login so rotation is picked up</description>
</property>
<property>
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 1fc524485..9ea9d2b8b 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
@@ -68,7 +68,30 @@
<property>
<name>ranger.audit.dispatcher.kafka.sasl.mechanism</name>
- <value>GSSAPI</value>
+ <value>OAUTHBEARER</value>
+ <description>OAUTHBEARER; GSSAPI is also supported and uses
service.kerberos.principal/keytab</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kafka.sasl.jaas.option.useKeyTab</name>
+ <value>true</value>
+ <description>GSSAPI Krb5LoginModule option; any option can be set as
ranger.audit.dispatcher.kafka.sasl.jaas.option.<name></description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kafka.sasl.jaas.option.storeKey</name>
+ <value>true</value>
+ </property>
+
+ <property>
+
<name>ranger.audit.dispatcher.kafka.sasl.jaas.option.useTicketCache</name>
+ <value>false</value>
+ </property>
+
+ <property>
+
<name>ranger.audit.dispatcher.kafka.sasl.oauthbearer.token.endpoint.url</name>
+ <value>file:///etc/kafka-oauth/rangerauditserver.token</value>
+ <description>Dev token minted by the ranger-kafka container; re-read
on every login so rotation is picked up</description>
</property>
<!-- Solr Dispatcher Threading Configuration -->
diff --git
a/dev-support/ranger-docker/scripts/audit-ingestor/audit-ingestor-config.py
b/dev-support/ranger-docker/scripts/audit-ingestor/audit-ingestor-config.py
new file mode 100755
index 000000000..c981eb6f3
--- /dev/null
+++ b/dev-support/ranger-docker/scripts/audit-ingestor/audit-ingestor-config.py
@@ -0,0 +1,105 @@
+#!/usr/bin/env python3
+
+# 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.
+
+# Renders conf/ranger-audit-ingestor-site.xml from
configs/ranger-audit-ingestor-site.yaml (plus
+# KAFKA_BOOTSTRAP_SERVERS from env), the same way dba.py prepares Ranger
Admin. A mounted configs/ranger-audit-ingestor-site.xml
+# is copied as it is instead; nothing is rendered over it.
+
+import logging
+import os
+import shutil
+import sys
+import xml.etree.ElementTree as ET
+
+import yaml
+
+AUDIT_INGESTOR_DIR = os.environ.get('AUDIT_INGESTOR_HOME_DIR',
'/opt/ranger/audit-ingestor')
+CONF_DIR = os.environ.get('AUDIT_INGESTOR_CONF_DIR',
os.path.join(AUDIT_INGESTOR_DIR, 'conf'))
+CONFIGS_DIR = os.environ.get('AUDIT_INGESTOR_CONFIGS',
os.path.join(AUDIT_INGESTOR_DIR, 'configs'))
+SITE_XML_NAME = 'ranger-audit-ingestor-site.xml'
+SITE_YAML_NAME = 'ranger-audit-ingestor-site.yaml'
+PREFIX = 'ranger.audit.ingestor.'
+
+# env variable -> property
+ENV_OVERRIDES = {
+ 'KAFKA_BOOTSTRAP_SERVERS': PREFIX + 'kafka.bootstrap.servers',
+}
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s
audit-ingestor-config: %(message)s')
+LOG = logging.getLogger('audit-ingestor-config')
+
+
+def to_property_value(value):
+ if isinstance(value, bool):
+ return str(value).lower()
+
+ return '' if value is None else str(value)
+
+
+def render_site_xml():
+ site_xml = os.path.join(CONFIGS_DIR, SITE_XML_NAME)
+ site_yaml = os.path.join(CONFIGS_DIR, SITE_YAML_NAME)
+ target = os.path.join(CONF_DIR, SITE_XML_NAME)
+
+ if os.path.isfile(site_xml):
+ shutil.copyfile(site_xml, target)
+ LOG.info(f'Using {site_xml} as it is, conf/{SITE_XML_NAME} is not
rendered from YAML')
+
+ return
+
+ if not os.path.isfile(site_yaml):
+ raise FileNotFoundError(f'configuration not found: neither {site_xml}
nor {site_yaml} exists')
+
+ with open(site_yaml) as f:
+ config = yaml.safe_load(f) or {}
+
+ if not isinstance(config, dict) or any(isinstance(value, (dict, list)) for
value in config.values()):
+ raise ValueError(f'{site_yaml} must be a flat map of property names to
values')
+
+ props = {str(name): to_property_value(value) for name, value in
config.items()}
+
+ for env_name, prop_name in ENV_OVERRIDES.items():
+ value = os.environ.get(env_name, '').strip()
+
+ if value:
+ props[prop_name] = value
+ LOG.info(f"{prop_name}: '{value}' (from env {env_name})")
+
+ root = ET.Element('configuration')
+
+ root.append(ET.Comment(f' generated by audit-ingestor-config.py from
{SITE_YAML_NAME} '))
+
+ for name, value in props.items():
+ prop = ET.SubElement(root, 'property')
+
+ ET.SubElement(prop, 'name').text = name
+ ET.SubElement(prop, 'value').text = value
+
+ ET.indent(root)
+ os.makedirs(CONF_DIR, exist_ok=True)
+ ET.ElementTree(root).write(target, encoding='utf-8', xml_declaration=True)
+
+ LOG.info(f'Rendered {target} with {len(props)} properties from
{site_yaml}')
+
+
+if __name__ == '__main__':
+ try:
+ render_site_xml()
+ except (OSError, ValueError, ET.ParseError, yaml.YAMLError) as err:
+ LOG.error(f'{err}')
+ sys.exit(1)
diff --git
a/dev-support/ranger-docker/scripts/audit-ingestor/configs/ranger-audit-ingestor-site.yaml
b/dev-support/ranger-docker/scripts/audit-ingestor/configs/ranger-audit-ingestor-site.yaml
new file mode 100644
index 000000000..4a5c3d478
--- /dev/null
+++
b/dev-support/ranger-docker/scripts/audit-ingestor/configs/ranger-audit-ingestor-site.yaml
@@ -0,0 +1,130 @@
+# 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.
+
+# Complete Ranger Audit Ingestor configuration for ranger-docker:
audit-ingestor-config.py renders
+# conf/ranger-audit-ingestor-site.xml from this file on every start, without
taking any property from
+# the distribution. Mount a configs/ranger-audit-ingestor-site.xml instead to
bypass the YAML entirely.
+# The renderer takes ranger.audit.ingestor.kafka.bootstrap.servers from env
KAFKA_BOOTSTRAP_SERVERS when set.
+
+
+# RANGER AUDIT INGESTOR SERVICE CONFIGURATION
+
+# Server Configuration
+log.dir: '${audit.server.log.dir}'
+ranger.audit.ingestor.webapp.dir: webapp/audit-ingestor
+ranger.audit.ingestor.contextName: /
+ranger.audit.ingestor.host: ranger-audit-ingestor.rangernw
+ranger.audit.ingestor.http.port: '7081'
+ranger.audit.ingestor.https.port: '7182'
+ranger.audit.ingestor.https.attrib.ssl.enabled: 'false'
+
+# HTTPS/SSL Configuration
+ranger.audit.ingestor.https.attrib.keystore.keyalias: myKey
+ranger.audit.ingestor.https.attrib.keystore.pass: _
+ranger.audit.ingestor.https.attrib.keystore.file:
/etc/ranger/ranger-audit-ingestor/keys/server.jks
+ranger.audit.ingestor.https.attrib.keystore.credential.alias:
keyStoreCredentialAlias
+ranger.audit.ingestor.tomcat.ciphers: ''
+
+# KERBEROS CONFIGURATION FOR REST API (INCOMING REQUESTS) used by kerberos
filter
+hadoop.security.authentication: kerberos
+ranger.audit.ingestor.kerberos.enabled: 'true'
+ranger.audit.ingestor.kerberos.type: kerberos
+ranger.audit.ingestor.kerberos.principal: 'HTTP/[email protected]'
+ranger.audit.ingestor.kerberos.keytab: /etc/keytabs/HTTP.keytab
+ranger.audit.ingestor.kerberos.name.rules: DEFAULT
+ranger.audit.ingestor.bind.address: ranger-audit-ingestor.rangernw
+ranger.audit.ingestor.cookie.path: /
+
+# JWT AUTHENTICATION CONFIGURATION to support JWT authentication
+config.prefix: ranger.audit.jwt.auth
+ranger.audit.jwt.auth.enabled: 'false'
+ranger.audit.jwt.auth.provider-url: 'http://localhost:9180/rest/jwks'
+ranger.audit.jwt.auth.public-key: ''
+ranger.audit.jwt.auth.cookie.name: hadoop-jwt
+ranger.audit.jwt.auth.audiences: ''
+
+# Trusted Header Auth Configs
+ranger.audit.ingestor.authn.header.enabled: 'false'
+ranger.audit.ingestor.authn.header.username: ''
+ranger.audit.ingestor.authn.header.spiffe: ''
+ranger.audit.ingestor.service.dev_hdfs.allowed.users: hdfs
+ranger.audit.ingestor.service.dev_yarn.allowed.users: yarn
+ranger.audit.ingestor.service.dev_hive.allowed.users: hive
+ranger.audit.ingestor.service.dev_hbase.allowed.users: hbase
+ranger.audit.ingestor.service.dev_kafka.allowed.users: kafka
+ranger.audit.ingestor.service.dev_knox.allowed.users: knox
+ranger.audit.ingestor.service.dev_kms.allowed.users: rangerkms
+ranger.audit.ingestor.service.dev_trino.allowed.users: trino
+ranger.audit.ingestor.service.dev_ozone.allowed.users: om
+ranger.audit.ingestor.service.dev_solr.allowed.users: solr
+
+# AUTH_TO_LOCAL RULES FOR KERBEROS PRINCIPAL MAPPING
+ranger.audit.ingestor.auth.to.local: |
+ 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:[1:$1@$0](.*@.*)s/@.*//
+ DEFAULT
+ranger.audit.ingestor.service.kerberos.principal:
'rangerauditserver/[email protected]'
+ranger.audit.ingestor.service.kerberos.keytab:
/etc/keytabs/rangerauditserver.keytab
+
+# CONFIGURATION TO ENABLE AUDIT DESTINATION CONFIGURATION
+xasecure.audit.destination.kafka: 'true'
+
+# KAFKA PRODUCER CONFIGURATION IN THE INGESTOR
+ranger.audit.ingestor.kafka.bootstrap.servers: 'ranger-kafka.rangernw:9092'
+ranger.audit.ingestor.kafka.topic.name: ranger_audits
+ranger.audit.ingestor.kafka.partitioner.class:
org.apache.ranger.audit.producer.kafka.AuditPartitioner
+ranger.audit.ingestor.kafka.configured.plugins:
'hdfs,yarn,knox,hiveServer2,hiveMetastore,kafka,hbaseRegional,hbaseMaster,solr,trino,ozone,kudu,nifi'
+ranger.audit.ingestor.kafka.topic.partitions.per.configured.plugin: '3'
+ranger.audit.ingestor.kafka.topic.partitions.buffer: '9'
+
+# partition allocation overrides for plugins
+ranger.audit.ingestor.kafka.topic.partitions: '10'
+ranger.audit.ingestor.kafka.replication.factor: '1'
+ranger.audit.ingestor.kafka.security.protocol: SASL_PLAINTEXT
+# OAUTHBEARER; GSSAPI is also supported and uses
service.kerberos.principal/keytab
+ranger.audit.ingestor.kafka.sasl.mechanism: OAUTHBEARER
+# GSSAPI Krb5LoginModule options; any option can be set as
ranger.audit.ingestor.kafka.sasl.jaas.option.<name>
+ranger.audit.ingestor.kafka.sasl.jaas.option.useKeyTab: 'true'
+ranger.audit.ingestor.kafka.sasl.jaas.option.storeKey: 'true'
+ranger.audit.ingestor.kafka.sasl.jaas.option.useTicketCache: 'false'
+# dev token minted by the ranger-kafka container; re-read on every login so
rotation is picked up
+ranger.audit.ingestor.kafka.sasl.oauthbearer.token.endpoint.url:
'file:///etc/kafka-oauth/rangerauditserver.token'
+ranger.audit.ingestor.kafka.request.timeout.ms: '60000'
+
+# KAFKA PRODUCER PERFORMANCE (AuditProducer → ProducerConfig)
+ranger.audit.ingestor.kafka.producer.batch.size: '131072'
+ranger.audit.ingestor.kafka.producer.linger.ms: '20'
+ranger.audit.ingestor.kafka.producer.buffer.memory: '134217728'
+ranger.audit.ingestor.kafka.producer.compression.type: lz4
+ranger.audit.ingestor.kafka.producer.delivery.timeout.ms: '120000'
+ranger.audit.ingestor.kafka.producer.max.request.size: '1048576'
+ranger.audit.ingestor.kafka.producer.max.block.ms: '60000'
+ranger.audit.ingestor.kafka.producer.batch.send.timeout.ms: '30000'
+
+# Optional topic configs at create time (leave unset for broker defaults;
enable for production RF>=3)
+ranger.audit.ingestor.kafka.connections.max.idle.ms: '90000'
+
+# AUDIT RECOVERY CONFIGURATION (FOR KAFKA SEND FAILURES)
+ranger.audit.ingestor.recovery.enabled: 'true'
+ranger.audit.ingestor.recovery.spool.dir:
/var/log/ranger/ranger-audit-ingestor/audit/spool
+ranger.audit.ingestor.recovery.archive.dir:
/var/log/ranger/ranger-audit-ingestor/audit/archive
+ranger.audit.ingestor.recovery.file.rotation.interval.sec: '300'
+ranger.audit.ingestor.recovery.max.messages.per.file: '10000'
+ranger.audit.ingestor.recovery.retry.interval.sec: '60'
+ranger.audit.ingestor.recovery.archive.max.processed.files: '100'
+ranger.audit.ingestor.recovery.retry.max.attempts: '3'
diff --git
a/dev-support/ranger-docker/scripts/audit-ingestor/ranger-audit-ingestor.sh
b/dev-support/ranger-docker/scripts/audit-ingestor/ranger-audit-ingestor.sh
index 875f51b98..4557ca841 100755
--- a/dev-support/ranger-docker/scripts/audit-ingestor/ranger-audit-ingestor.sh
+++ b/dev-support/ranger-docker/scripts/audit-ingestor/ranger-audit-ingestor.sh
@@ -38,6 +38,13 @@ echo "=========================================="
# Source service check functions
source /home/ranger/scripts/service-check-functions.sh
+# Render conf/ranger-audit-ingestor-site.xml from
configs/ranger-audit-ingestor-site.yaml
+# (KAFKA_BOOTSTRAP_SERVERS from env); a mounted
configs/ranger-audit-ingestor-site.xml wins
+if ! python3 /home/ranger/scripts/audit-ingestor-config.py; then
+ echo "[ERROR] Audit ingestor configuration failed, see
audit-ingestor-config.py logs above"
+ exit 1
+fi
+
# Quick check for Kafka availability
# The audit server has a built-in recovery/spool mechanism for when Kafka is
unavailable
KAFKA_BOOTSTRAP_SERVERS="${KAFKA_BOOTSTRAP_SERVERS:-ranger-kafka:9092}"
diff --git
a/dev-support/ranger-docker/scripts/kafka/kafka-oauth-token-minter.sh
b/dev-support/ranger-docker/scripts/kafka/kafka-oauth-token-minter.sh
new file mode 100755
index 000000000..8ba0a8b42
--- /dev/null
+++ b/dev-support/ranger-docker/scripts/kafka/kafka-oauth-token-minter.sh
@@ -0,0 +1,65 @@
+#!/bin/bash
+
+# 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.
+
+# Dev-only stand-in for a platform token issuer (e.g. the kubelet writing
projected
+# service-account tokens). Mints unsigned (alg=none) JWTs for the Ranger Kafka
clients
+# into a volume shared with their containers and rewrites them in place on a
loop, so
+# the SASL/OAUTHBEARER login handler's re-read-on-refresh path is exercised.
The broker
+# accepts them through Kafka's OAuthBearerUnsecuredValidatorCallbackHandler;
never use
+# this outside the docker dev setup.
+
+set -u
+
+TOKEN_DIR="${KAFKA_OAUTH_TOKEN_DIR:-/etc/kafka-oauth}"
+TOKEN_TTL="${KAFKA_OAUTH_TOKEN_TTL_SECONDS:-600}"
+REFRESH="${KAFKA_OAUTH_TOKEN_REFRESH_SECONDS:-120}"
+SUBJECTS="${KAFKA_OAUTH_TOKEN_SUBJECTS:-rangerauditserver rangertagsync}"
+ISSUER="${KAFKA_OAUTH_TOKEN_ISSUER:-ranger-docker}"
+
+b64url() {
+ base64 -w0 | tr '+/' '-_' | tr -d '='
+}
+
+mint_token() {
+ local subject="$1"
+ local file="${TOKEN_DIR}/${subject}.token"
+ local now exp header payload
+
+ now=$(date +%s)
+ exp=$((now + TOKEN_TTL))
+ header=$(printf '{"alg":"none","typ":"JWT"}' | b64url)
+ payload=$(printf '{"iss":"%s","sub":"%s","aud":"kafka","iat":%d,"exp":%d}'
"${ISSUER}" "${subject}" "${now}" "${exp}" | b64url)
+
+ # same atomic replace kubelet does: write a temp file, then rename over the
old token
+ printf '%s.%s.' "${header}" "${payload}" > "${file}.tmp"
+ chmod 644 "${file}.tmp"
+ mv -f "${file}.tmp" "${file}"
+}
+
+mkdir -p "${TOKEN_DIR}"
+chmod 755 "${TOKEN_DIR}"
+
+echo "[INFO] kafka-oauth-token-minter: writing tokens for [${SUBJECTS}] to
${TOKEN_DIR} (ttl=${TOKEN_TTL}s, refresh every ${REFRESH}s)"
+
+while true; do
+ for subject in ${SUBJECTS}; do
+ mint_token "${subject}"
+ done
+
+ sleep "${REFRESH}"
+done
diff --git a/dev-support/ranger-docker/scripts/kafka/ranger-kafka-setup.sh
b/dev-support/ranger-docker/scripts/kafka/ranger-kafka-setup.sh
index 067042ada..f87bf508b 100755
--- a/dev-support/ranger-docker/scripts/kafka/ranger-kafka-setup.sh
+++ b/dev-support/ranger-docker/scripts/kafka/ranger-kafka-setup.sh
@@ -43,8 +43,12 @@ if [ "${KERBEROS_ENABLED}" == "true" ]; then
echo "Configuring Kafka with Kerberos (SASL_PLAINTEXT)"
cat <<EOF >> ${KAFKA_HOME}/config/server.properties
- # Enable SASL/GSSAPI mechanism
- sasl.enabled.mechanisms=GSSAPI
+ # SASL/GSSAPI for inter-broker, the kafka CLI/healthcheck and clients that
opt in; SASL/OAUTHBEARER
+ # for the Ranger clients (default). OAUTHBEARER tokens are dev tokens from
kafka-oauth-token-minter.sh,
+ # accepted unsigned by Kafka's unsecured validator; the token's "sub"
becomes the Kafka principal.
+ sasl.enabled.mechanisms=GSSAPI,OAUTHBEARER
+
listener.name.sasl_plaintext.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule
required unsecuredLoginStringClaim_sub="kafka";
+
listener.name.sasl_plaintext.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredValidatorCallbackHandler
sasl.mechanism.inter.broker.protocol=GSSAPI
security.inter.broker.protocol=SASL_PLAINTEXT
diff --git a/dev-support/ranger-docker/scripts/kafka/ranger-kafka.sh
b/dev-support/ranger-docker/scripts/kafka/ranger-kafka.sh
index 1e22101c3..0e1d862cf 100755
--- a/dev-support/ranger-docker/scripts/kafka/ranger-kafka.sh
+++ b/dev-support/ranger-docker/scripts/kafka/ranger-kafka.sh
@@ -45,6 +45,8 @@ fi
# Configure KAFKA_OPTS based on KERBEROS_ENABLED
if [ "${KERBEROS_ENABLED}" == "true" ]; then
echo "Starting Kafka with Kerberos authentication (SASL_PLAINTEXT)"
+ # dev token issuer for the SASL/OAUTHBEARER clients (audit ingestor,
dispatchers, tagsync)
+ "${RANGER_SCRIPTS}"/kafka-oauth-token-minter.sh &
KAFKA_OPTS='-Djava.security.krb5.conf=/etc/krb5.conf
-Djava.security.auth.login.config=/opt/kafka/config/kafka-server-jaas.conf'
else
echo "Starting Kafka with PLAINTEXT (no Kerberos authentication)"
diff --git
a/dev-support/ranger-docker/scripts/tagsync/ranger-tagsync-install.properties
b/dev-support/ranger-docker/scripts/tagsync/ranger-tagsync-install.properties
index 83d2eb249..f48c90b41 100755
---
a/dev-support/ranger-docker/scripts/tagsync/ranger-tagsync-install.properties
+++
b/dev-support/ranger-docker/scripts/tagsync/ranger-tagsync-install.properties
@@ -33,6 +33,9 @@ TAG_SOURCE_ATLAS_KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR = 1
TAG_SOURCE_ATLAS_KAFKA_BOOTSTRAP_SERVERS = ranger-kafka.rangernw:9092
TAG_SOURCE_ATLAS_KAFKA_ENTITIES_GROUP_ID = ranger_entities_consumer
TAG_SOURCE_ATLAS_KAFKA_SECURITY_PROTOCOL = SASL_PLAINTEXT
+# supported values: OAUTHBEARER (dev token minted by the ranger-kafka
container) and GSSAPI (the keytab below)
+TAG_SOURCE_ATLAS_KAFKA_SASL_MECHANISM = OAUTHBEARER
+TAG_SOURCE_ATLAS_KAFKA_OAUTH_TOKEN_URL =
file:///etc/kafka-oauth/rangertagsync.token
TAG_SOURCE_ATLAS_KAFKA_SERVICE_NAME = kafka
TAG_SOURCE_ATLAS_KERBEROS_PRINCIPAL =
rangertagsync/[email protected]
TAG_SOURCE_ATLAS_KERBEROS_KEYTAB = /etc/keytabs/rangertagsync.keytab
diff --git a/dev-support/ranger-docker/scripts/tagsync/ranger-tagsync.sh
b/dev-support/ranger-docker/scripts/tagsync/ranger-tagsync.sh
index b98848ce8..9c12086b7 100755
--- a/dev-support/ranger-docker/scripts/tagsync/ranger-tagsync.sh
+++ b/dev-support/ranger-docker/scripts/tagsync/ranger-tagsync.sh
@@ -45,6 +45,18 @@ then
fi
fi
+# Atlas Kafka consumer authenticates with the dev SASL/OAUTHBEARER token
minted by ranger-kafka unless
+# TAG_SOURCE_ATLAS_KAFKA_SASL_MECHANISM selects GSSAPI; give the minter a
moment on a fresh stack
+if [ "${KERBEROS_ENABLED}" == "true" ] && grep -qiE '^TAG_SOURCE_ATLAS_ENABLED
*= *true' ${RANGER_HOME}/tagsync/install.properties && ! grep -qiE
'^TAG_SOURCE_ATLAS_KAFKA_SASL_MECHANISM *= *GSSAPI'
${RANGER_HOME}/tagsync/install.properties
+then
+ for i in $(seq 1 30)
+ do
+ [ -s /etc/kafka-oauth/rangertagsync.token ] && break
+ echo "Waiting for Kafka OAUTHBEARER token
/etc/kafka-oauth/rangertagsync.token ($i/30)"
+ sleep 2
+ done
+fi
+
cd ${RANGER_HOME}/tagsync && ./ranger-tagsync-services.sh start
RANGER_TAGSYNC_PID=`ps -ef | grep -v grep | grep -i
"org.apache.ranger.tagsync.server.RangerTagSyncServer" | awk '{print $2}'`
diff --git a/distro/src/main/assembly/tagsync.xml
b/distro/src/main/assembly/tagsync.xml
index 555401f2f..a95138c29 100644
--- a/distro/src/main/assembly/tagsync.xml
+++ b/distro/src/main/assembly/tagsync.xml
@@ -88,6 +88,7 @@
<include>org.apache.ranger:ranger-authz-api</include>
<include>org.apache.ranger:ranger-common-ha:jar:${project.version}</include>
<include>org.apache.ranger:ranger-common-utils</include>
+
<include>org.apache.ranger:ranger-kafka-client-auth</include>
<include>org.apache.ranger:ranger-plugins-common</include>
<include>org.apache.ranger:ranger-util</include>
<include>org.apache.tomcat.embed:tomcat-embed*</include>
diff --git a/pom.xml b/pom.xml
index 70c307dee..6ecd0eae3 100644
--- a/pom.xml
+++ b/pom.xml
@@ -961,6 +961,7 @@
<module>ranger-hbase-plugin-shim</module>
<module>ranger-hdfs-plugin-shim</module>
<module>ranger-hive-plugin-shim</module>
+ <module>ranger-kafka-client-auth</module>
<module>ranger-kafka-plugin-shim</module>
<module>ranger-kms-plugin-shim</module>
<module>ranger-knox-plugin-shim</module>
@@ -1119,6 +1120,7 @@
<module>common-utils</module>
<module>credentialbuilder</module>
<module>plugin-kafka</module>
+ <module>ranger-kafka-client-auth</module>
<module>ranger-kafka-plugin-shim</module>
<module>ranger-plugin-classloader</module>
<module>ranger-util</module>
@@ -1312,6 +1314,7 @@
<module>ranger-hbase-plugin-shim</module>
<module>ranger-hdfs-plugin-shim</module>
<module>ranger-hive-plugin-shim</module>
+ <module>ranger-kafka-client-auth</module>
<module>ranger-kafka-plugin-shim</module>
<module>ranger-kms-plugin-shim</module>
<module>ranger-knox-plugin-shim</module>
@@ -1400,6 +1403,7 @@
<module>ranger-hbase-plugin-shim</module>
<module>ranger-hdfs-plugin-shim</module>
<module>ranger-hive-plugin-shim</module>
+ <module>ranger-kafka-client-auth</module>
<module>ranger-kafka-plugin-shim</module>
<module>ranger-kms-plugin-shim</module>
<module>ranger-knox-plugin-shim</module>
diff --git a/audit-server/audit-common/pom.xml
b/ranger-kafka-client-auth/pom.xml
similarity index 69%
copy from audit-server/audit-common/pom.xml
copy to ranger-kafka-client-auth/pom.xml
index 5266db1c2..a8dbc02e8 100644
--- a/audit-server/audit-common/pom.xml
+++ b/ranger-kafka-client-auth/pom.xml
@@ -17,35 +17,43 @@
-->
<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
<modelVersion>4.0.0</modelVersion>
-
<parent>
<groupId>org.apache.ranger</groupId>
<artifactId>ranger</artifactId>
<version>3.0.0-SNAPSHOT</version>
- <relativePath>../..</relativePath>
+ <relativePath>..</relativePath>
</parent>
-
- <artifactId>ranger-audit-server-common</artifactId>
+ <artifactId>ranger-kafka-client-auth</artifactId>
<packaging>jar</packaging>
- <name>Ranger Audit Server Common</name>
- <description>Shared classes between audit ingestor and
dispatcher</description>
-
+ <name>Ranger Kafka Client Auth</name>
+ <description>Shared Kafka client authentication (SASL/GSSAPI,
SASL/OAUTHBEARER) for Ranger services that produce to or consume from
Kafka</description>
+ <properties>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ </properties>
<dependencies>
<dependency>
- <groupId>org.apache.kafka</groupId>
- <artifactId>kafka-clients</artifactId>
- <version>${kafka.version}</version>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>${fasterxml.jackson.databind.version}</version>
</dependency>
<dependency>
- <groupId>org.apache.ranger</groupId>
- <artifactId>ranger-plugins-common</artifactId>
- <version>${project.version}</version>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
+ <version>${commons.lang3.version}</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>${slf4j.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>${kafka.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Test -->
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>
@@ -53,8 +61,4 @@
<scope>test</scope>
</dependency>
</dependencies>
-
- <build>
- <finalName>ranger-audit-server-common-${project.version}</finalName>
- </build>
-</project>
\ No newline at end of file
+</project>
diff --git
a/ranger-kafka-client-auth/src/main/java/org/apache/ranger/kafka/auth/RangerKafkaClientSecurityConfig.java
b/ranger-kafka-client-auth/src/main/java/org/apache/ranger/kafka/auth/RangerKafkaClientSecurityConfig.java
new file mode 100644
index 000000000..1e74fbecc
--- /dev/null
+++
b/ranger-kafka-client-auth/src/main/java/org/apache/ranger/kafka/auth/RangerKafkaClientSecurityConfig.java
@@ -0,0 +1,320 @@
+/*
+ * 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.kafka.auth;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.common.config.SaslConfigs;
+import
org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.URI;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Properties;
+import java.util.TreeMap;
+import java.util.stream.Collectors;
+
+/**
+ * Builds the security portion of a Kafka client configuration from
Ranger-style properties.
+ * <p>
+ * All keys are read relative to a caller-supplied prefix such as {@code
ranger.audit.ingestor}:
+ * <ul>
+ * <li>{@code <prefix>.kafka.security.protocol}: PLAINTEXT (default), SSL,
SASL_PLAINTEXT or SASL_SSL</li>
+ * <li>{@code <prefix>.kafka.sasl.mechanism}: GSSAPI (default), OAUTHBEARER
or any other Kafka mechanism</li>
+ * <li>{@code <prefix>.service.kerberos.principal}: GSSAPI principal, _HOST
is resolved</li>
+ * <li>{@code <prefix>.service.kerberos.keytab}: GSSAPI keytab path</li>
+ * <li>{@code <prefix>.kafka.sasl.oauthbearer.token.endpoint.url}:
OAUTHBEARER token, file:// or http(s)://</li>
+ * <li>{@code <prefix>.kafka.sasl.jaas.loginModuleName}: JAAS login module,
defaults per mechanism</li>
+ * <li>{@code <prefix>.kafka.sasl.jaas.loginModuleControlFlag}: JAAS control
flag, default required</li>
+ * <li>{@code <prefix>.kafka.sasl.jaas.option.<name>}: any JAAS login module
option (for GSSAPI the shipped configuration sets useKeyTab, storeKey,
useTicketCache)</li>
+ * <li>{@code <prefix>.kafka.sasl.*} and {@code <prefix>.kafka.ssl.*}:
passed through, overriding derived values</li>
+ * </ul>
+ * A file:// token URL selects {@link
RangerKafkaOAuthBearerLoginCallbackHandler}, an http(s):// URL Kafka's own
+ * client-credentials handler. Producers, consumers and admin clients all
accept the resulting keys.
+ */
+public final class RangerKafkaClientSecurityConfig {
+ private static final Logger LOG =
LoggerFactory.getLogger(RangerKafkaClientSecurityConfig.class);
+
+ public static final String PROP_SECURITY_PROTOCOL =
"kafka.security.protocol";
+ public static final String PROP_SASL_MECHANISM =
"kafka.sasl.mechanism";
+ public static final String PROP_SASL_KERBEROS_SERVICE_NAME =
"kafka.sasl.kerberos.service.name";
+ public static final String PROP_OAUTH_TOKEN_ENDPOINT_URL =
"kafka.sasl.oauthbearer.token.endpoint.url";
+ public static final String PROP_JAAS_LOGIN_MODULE_NAME =
"kafka.sasl.jaas.loginModuleName";
+ public static final String PROP_JAAS_CONTROL_FLAG =
"kafka.sasl.jaas.loginModuleControlFlag";
+ public static final String PROP_JAAS_OPTION_PREFIX =
"kafka.sasl.jaas.option.";
+ public static final String PROP_KERBEROS_PRINCIPAL =
"service.kerberos.principal";
+ public static final String PROP_KERBEROS_KEYTAB =
"service.kerberos.keytab";
+ public static final String PROP_HOST = "host";
+
+ public static final String SECURITY_PROTOCOL_PLAINTEXT = "PLAINTEXT";
+ public static final String SASL_MECHANISM_GSSAPI = "GSSAPI";
+ public static final String SASL_MECHANISM_OAUTHBEARER = "OAUTHBEARER";
+ public static final String DEFAULT_SASL_MECHANISM =
SASL_MECHANISM_GSSAPI;
+ public static final String DEFAULT_JAAS_CONTROL_FLAG = "required";
+ public static final String DEFAULT_KERBEROS_SERVICE_NAME = "kafka";
+ public static final String HOSTNAME_MARKER = "_HOST";
+
+ public static final String OAUTHBEARER_LOGIN_MODULE =
"org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule";
+ public static final String KRB5_LOGIN_MODULE =
"com.sun.security.auth.module.Krb5LoginModule";
+
+ public static final String JAAS_OPTION_USE_KEYTAB = "useKeyTab";
+ public static final String JAAS_OPTION_KEYTAB = "keyTab";
+ public static final String JAAS_OPTION_SERVICE_NAME = "serviceName";
+ public static final String JAAS_OPTION_PRINCIPAL = "principal";
+
+ private static final String SASL_PROTOCOL_MARKER = "SASL";
+ private static final String SASL_PASSTHROUGH = "kafka.sasl.";
+ private static final String SSL_PASSTHROUGH = "kafka.ssl.";
+ private static final String JAAS_RANGER_PREFIX =
"kafka.sasl.jaas.";
+ private static final String KAFKA_PREFIX = "kafka.";
+ private static final String SENSITIVE_KEY_PART = "password";
+ private static final String MASKED_VALUE = "***";
+ private static final String PROPERTY_NAME_FORMAT = "%s.%s";
+ private static final String KAFKA_PROPERTY_NAME_FORMAT = "%s.kafka.%s";
+ private static final String PRINCIPAL_FORMAT = "%s/%s@%s";
+
+ private RangerKafkaClientSecurityConfig() {
+ }
+
+ /** Returns a fresh map holding only the security keys derived from {@code
props}. */
+ public static Map<String, Object> build(Properties props, String
propPrefix) {
+ Map<String, Object> ret = new HashMap<>();
+
+ apply(props, propPrefix, ret);
+
+ return ret;
+ }
+
+ /** Adds the security keys derived from {@code props} to an existing
client configuration. */
+ public static void apply(Properties props, String propPrefix, Map<? super
String, ? super Object> target) {
+ Map<String, Object> security = new HashMap<>();
+ String protocol = getProperty(props, propPrefix,
PROP_SECURITY_PROTOCOL, SECURITY_PROTOCOL_PLAINTEXT);
+
+ security.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, protocol);
+
+ if (isSasl(protocol)) {
+ String mechanism = getProperty(props, propPrefix,
PROP_SASL_MECHANISM, DEFAULT_SASL_MECHANISM);
+
+ security.put(SaslConfigs.SASL_MECHANISM, mechanism);
+
+ if (SASL_MECHANISM_GSSAPI.equalsIgnoreCase(mechanism)) {
+ applyKerberos(props, propPrefix, security);
+ } else if (SASL_MECHANISM_OAUTHBEARER.equalsIgnoreCase(mechanism))
{
+ applyOAuthBearer(props, propPrefix, security);
+ } else {
+ applyConfiguredJaas(props, propPrefix, security, null);
+ }
+ }
+
+ applyPassThrough(props, propPrefix, security);
+
+ if (isSasl(protocol) &&
!security.containsKey(SaslConfigs.SASL_JAAS_CONFIG)) {
+ throw new IllegalStateException(String.format("Kafka security
protocol %s with mechanism %s requires %s or %s", protocol,
security.get(SaslConfigs.SASL_MECHANISM),
+ propertyName(propPrefix, PROP_JAAS_LOGIN_MODULE_NAME),
String.format(KAFKA_PROPERTY_NAME_FORMAT, propPrefix,
SaslConfigs.SASL_JAAS_CONFIG)));
+ }
+
+ target.putAll(security);
+
+ logSecurityConfig(propPrefix, security);
+ }
+
+ public static boolean isSasl(String securityProtocol) {
+ return StringUtils.startsWithIgnoreCase(securityProtocol,
SASL_PROTOCOL_MARKER);
+ }
+
+ /* Resolves _HOST in a Kerberos principal, falling back to the local
canonical host name. */
+ static String resolvePrincipal(String principal, String hostName) {
+ String[] components = principal.split("[/@]");
+
+ if (components.length != 3 || !HOSTNAME_MARKER.equals(components[1])) {
+ return principal;
+ }
+
+ String fqdn = hostName;
+
+ if (StringUtils.isBlank(fqdn) || "0.0.0.0".equals(fqdn)) {
+ try {
+ fqdn = InetAddress.getLocalHost().getCanonicalHostName();
+ } catch (IOException e) {
+ throw new IllegalStateException(String.format("Unable to
resolve %s in principal %s", HOSTNAME_MARKER, principal), e);
+ }
+ }
+
+ return String.format(PRINCIPAL_FORMAT, components[0],
fqdn.toLowerCase(Locale.ROOT), components[2]);
+ }
+
+ /* Krb5LoginModule whose options are all configuration driven; the shipped
site XML carries
+ * useKeyTab/storeKey/useTicketCache, so ticket-cache or other setups need
no code change. */
+ private static void applyKerberos(Properties props, String propPrefix,
Map<String, Object> security) {
+ String principal = getProperty(props, propPrefix,
PROP_KERBEROS_PRINCIPAL, null);
+ String keytab = getProperty(props, propPrefix,
PROP_KERBEROS_KEYTAB, null);
+ String hostName = getProperty(props, propPrefix,
PROP_HOST, null);
+ String serviceName = getProperty(props, propPrefix,
PROP_SASL_KERBEROS_SERVICE_NAME, DEFAULT_KERBEROS_SERVICE_NAME);
+ Map<String, String> options = new LinkedHashMap<>();
+
+ /* keyTab, serviceName and principal come from the service properties;
every other Krb5LoginModule option (useKeyTab, storeKey, useTicketCache, ...)
comes from <prefix>.kafka.sasl.jaas.option.* */
+ if (keytab != null) {
+ options.put(JAAS_OPTION_KEYTAB, keytab);
+ }
+
+ options.put(JAAS_OPTION_SERVICE_NAME, serviceName);
+
+ if (principal != null) {
+ options.put(JAAS_OPTION_PRINCIPAL, principal);
+ }
+
+ options.putAll(configuredJaasOptions(props, propPrefix));
+
+ if (options.containsKey(JAAS_OPTION_PRINCIPAL)) {
+ options.put(JAAS_OPTION_PRINCIPAL,
resolvePrincipal(options.get(JAAS_OPTION_PRINCIPAL), hostName));
+ }
+
+ if (Boolean.parseBoolean(options.get(JAAS_OPTION_USE_KEYTAB))) {
+ /* keytab login was asked for by configuration, so fail early on a
missing or unreadable keytab */
+ String keytabPath = options.get(JAAS_OPTION_KEYTAB);
+
+ if (keytabPath == null ||
!options.containsKey(JAAS_OPTION_PRINCIPAL)) {
+ throw new IllegalStateException(String.format("Kafka
SASL/GSSAPI with %s=true requires both %s and %s", JAAS_OPTION_USE_KEYTAB,
propertyName(propPrefix, PROP_KERBEROS_PRINCIPAL), propertyName(propPrefix,
PROP_KERBEROS_KEYTAB)));
+ }
+
+ File keytabFile = new File(keytabPath);
+
+ if (!keytabFile.isFile()) {
+ throw new IllegalStateException(String.format("Keytab file not
found: %s", keytabPath));
+ }
+
+ if (!keytabFile.canRead()) {
+ throw new IllegalStateException(String.format("Keytab file not
readable: %s", keytabPath));
+ }
+ }
+
+ security.put(SaslConfigs.SASL_KERBEROS_SERVICE_NAME, serviceName);
+ security.put(SaslConfigs.SASL_JAAS_CONFIG, buildJaasConfig(props,
propPrefix, KRB5_LOGIN_MODULE, options));
+ }
+
+ private static void applyOAuthBearer(Properties props, String propPrefix,
Map<String, Object> security) {
+ String url = getProperty(props, propPrefix,
PROP_OAUTH_TOKEN_ENDPOINT_URL, null);
+
+ if (url == null) {
+ throw new IllegalStateException(String.format("Kafka
SASL/OAUTHBEARER requires %s (file:// token or http(s):// token endpoint)",
propertyName(propPrefix, PROP_OAUTH_TOKEN_ENDPOINT_URL)));
+ }
+
+ /* file:// tokens rotate in place, so use the Ranger handler that
re-reads them; http(s):// endpoints go through Kafka's own client-credentials
handler. */
+ String handler = isFileUrl(url) ?
RangerKafkaOAuthBearerLoginCallbackHandler.class.getName() :
OAuthBearerLoginCallbackHandler.class.getName();
+
+ security.put(SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL, url);
+ security.put(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS, handler);
+
+ applyConfiguredJaas(props, propPrefix, security,
OAUTHBEARER_LOGIN_MODULE);
+ }
+
+ /* JAAS from <prefix>.kafka.sasl.jaas.* with an optional default login
module; nothing is set when neither is available, leaving the mechanism to a
passed-through sasl.jaas.config. */
+ private static void applyConfiguredJaas(Properties props, String
propPrefix, Map<String, Object> security, String defaultLoginModule) {
+ String loginModule = getProperty(props, propPrefix,
PROP_JAAS_LOGIN_MODULE_NAME, defaultLoginModule);
+
+ if (loginModule != null) {
+ security.put(SaslConfigs.SASL_JAAS_CONFIG, buildJaasConfig(props,
propPrefix, loginModule, configuredJaasOptions(props, propPrefix)));
+ }
+ }
+
+ private static String buildJaasConfig(Properties props, String propPrefix,
String defaultLoginModule, Map<String, String> options) {
+ String loginModule = getProperty(props, propPrefix,
PROP_JAAS_LOGIN_MODULE_NAME, defaultLoginModule);
+ String controlFlag = getProperty(props, propPrefix,
PROP_JAAS_CONTROL_FLAG, DEFAULT_JAAS_CONTROL_FLAG);
+ String optionText = options.entrySet().stream()
+ .map(option -> String.format(" %s=\"%s\"", option.getKey(),
option.getValue()))
+ .collect(Collectors.joining());
+
+ return String.format("%s %s%s;", loginModule, controlFlag, optionText);
+ }
+
+ /* sorted so the generated JAAS string is stable across runs */
+ private static Map<String, String> configuredJaasOptions(Properties props,
String propPrefix) {
+ Map<String, String> ret = new TreeMap<>();
+ String optionPrefix = propertyName(propPrefix,
PROP_JAAS_OPTION_PREFIX);
+
+ for (String name : props.stringPropertyNames()) {
+ if (name.startsWith(optionPrefix)) {
+ String value = props.getProperty(name);
+
+ if (StringUtils.isNotBlank(value)) {
+ ret.put(name.substring(optionPrefix.length()), value);
+ }
+ }
+ }
+
+ return ret;
+ }
+
+ private static void applyPassThrough(Properties props, String propPrefix,
Map<String, Object> security) {
+ String saslPrefix = propertyName(propPrefix, SASL_PASSTHROUGH);
+ String sslPrefix = propertyName(propPrefix, SSL_PASSTHROUGH);
+ String jaasPrefix = propertyName(propPrefix, JAAS_RANGER_PREFIX);
+ String jaasConfig = String.format(KAFKA_PROPERTY_NAME_FORMAT,
propPrefix, SaslConfigs.SASL_JAAS_CONFIG);
+
+ for (String name : props.stringPropertyNames()) {
+ boolean rangerOnlyJaasKey = name.startsWith(jaasPrefix) &&
!name.equals(jaasConfig);
+
+ if ((name.startsWith(saslPrefix) || name.startsWith(sslPrefix)) &&
!rangerOnlyJaasKey) {
+ String value = props.getProperty(name);
+
+ if (StringUtils.isNotBlank(value)) {
+ security.put(name.substring(propPrefix.length() + 1 +
KAFKA_PREFIX.length()), value);
+ }
+ }
+ }
+ }
+
+ private static boolean isFileUrl(String url) {
+ try {
+ return "file".equalsIgnoreCase(URI.create(url).getScheme());
+ } catch (IllegalArgumentException e) {
+ return false;
+ }
+ }
+
+ private static String getProperty(Properties props, String propPrefix,
String name, String defaultValue) {
+ String value = props.getProperty(propertyName(propPrefix, name));
+
+ return StringUtils.isBlank(value) ? defaultValue : value;
+ }
+
+ private static String propertyName(String propPrefix, String name) {
+ return String.format(PROPERTY_NAME_FORMAT, propPrefix, name);
+ }
+
+ private static void logSecurityConfig(String propPrefix, Map<String,
Object> security) {
+ String summary = security.entrySet().stream()
+ .map(entry -> String.format("%s=%s", entry.getKey(),
isSensitive(entry.getKey()) ? MASKED_VALUE : entry.getValue()))
+ .collect(Collectors.joining(" "));
+
+ LOG.info("Kafka client security for {}: {}", propPrefix, summary);
+ }
+
+ private static boolean isSensitive(String key) {
+ return SaslConfigs.SASL_JAAS_CONFIG.equals(key) ||
StringUtils.containsIgnoreCase(key, SENSITIVE_KEY_PART);
+ }
+}
diff --git
a/ranger-kafka-client-auth/src/main/java/org/apache/ranger/kafka/auth/RangerKafkaOAuthBearerLoginCallbackHandler.java
b/ranger-kafka-client-auth/src/main/java/org/apache/ranger/kafka/auth/RangerKafkaOAuthBearerLoginCallbackHandler.java
new file mode 100644
index 000000000..7bdce8cff
--- /dev/null
+++
b/ranger-kafka-client-auth/src/main/java/org/apache/ranger/kafka/auth/RangerKafkaOAuthBearerLoginCallbackHandler.java
@@ -0,0 +1,225 @@
+/*
+ * 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.kafka.auth;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler;
+import org.apache.kafka.common.security.auth.SaslExtensions;
+import org.apache.kafka.common.security.auth.SaslExtensionsCallback;
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerTokenCallback;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+import javax.security.auth.login.AppConfigurationEntry;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * SASL/OAUTHBEARER login callback handler that presents a JWT read from a
local file, such as a
+ * Kubernetes projected service-account token, to the Kafka broker.
+ * <p>
+ * Kafka's bundled {@code OAuthBearerLoginCallbackHandler} also accepts a
{@code file://} value for
+ * {@code sasl.oauthbearer.token.endpoint.url}, but it reads the file once and
caches the token, so a
+ * client fails to re-authenticate after the platform rotates the token in
place. This handler never
+ * caches: every login and every refresh re-reads the file and re-derives
{@code sub}, {@code exp}
+ * and {@code iat}, so Kafka's refresh scheduling always follows the current
token. Signature and
+ * audience validation are the broker's responsibility.
+ */
+public class RangerKafkaOAuthBearerLoginCallbackHandler implements
AuthenticateCallbackHandler, AutoCloseable {
+ private static final Logger LOG =
LoggerFactory.getLogger(RangerKafkaOAuthBearerLoginCallbackHandler.class);
+
+ /* same system property Kafka's bundled handler consults; when set, the
token URL must be listed */
+ public static final String ALLOWED_URLS_SYSTEM_PROPERTY =
"org.apache.kafka.sasl.oauthbearer.allowed.urls";
+ public static final int TOKEN_READ_MAX_ATTEMPTS = 5;
+ public static final long TOKEN_READ_RETRY_SLEEP_MS = 50L;
+
+ private static final String CLAIM_SUBJECT = "sub";
+ private static final String CLAIM_EXPIRATION = "exp";
+ private static final String CLAIM_ISSUED_AT = "iat";
+ private static final ObjectMapper MAPPER = new
ObjectMapper();
+
+ private volatile Path tokenFile;
+
+ @Override
+ public void configure(Map<String, ?> configs, String saslMechanism,
List<AppConfigurationEntry> jaasConfigEntries) {
+ Object urlValue =
configs.get(SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL);
+ String url = urlValue != null ? urlValue.toString() : null;
+
+ if (StringUtils.isBlank(url)) {
+ throw new ConfigException(String.format("%s must be set to a
file:// URL for %s", SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL,
getClass().getSimpleName()));
+ }
+
+ throwIfUrlNotAllowed(url);
+
+ tokenFile = resolveTokenFile(url);
+
+ LOG.info("{}: reading SASL/OAUTHBEARER token from {}",
getClass().getSimpleName(), tokenFile);
+ }
+
+ @Override
+ public void handle(Callback[] callbacks) throws IOException,
UnsupportedCallbackException {
+ Path file = tokenFile;
+
+ if (file == null) {
+ throw new IllegalStateException(String.format("%s is not
configured; call configure() first", getClass().getSimpleName()));
+ }
+
+ for (Callback callback : callbacks) {
+ if (callback instanceof OAuthBearerTokenCallback) {
+ RangerOAuthBearerToken token = readToken(file);
+
+ ((OAuthBearerTokenCallback) callback).token(token);
+
+ LOG.debug("{}: token loaded from {} for principal={},
expiresAtMs={}", getClass().getSimpleName(), file, token.principalName(),
token.lifetimeMs());
+ } else if (callback instanceof SaslExtensionsCallback) {
+ ((SaslExtensionsCallback)
callback).extensions(SaslExtensions.empty());
+ } else {
+ throw new UnsupportedCallbackException(callback);
+ }
+ }
+ }
+
+ @Override
+ public void close() {
+ tokenFile = null;
+ }
+
+ private void throwIfUrlNotAllowed(String url) {
+ String allowed = System.getProperty(ALLOWED_URLS_SYSTEM_PROPERTY);
+
+ if (allowed == null) {
+ return;
+ }
+
+ if
(Arrays.stream(allowed.split(",")).map(String::trim).noneMatch(url::equalsIgnoreCase))
{
+ throw new ConfigException(String.format("%s is not allowed. Update
system property '%s' to allow it", url, ALLOWED_URLS_SYSTEM_PROPERTY));
+ }
+ }
+
+ private Path resolveTokenFile(String url) {
+ URI uri;
+
+ try {
+ uri = URI.create(url);
+ } catch (IllegalArgumentException e) {
+ throw new ConfigException(String.format("%s is not a valid URL:
%s", SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL, url), e);
+ }
+
+ if (!"file".equalsIgnoreCase(uri.getScheme())) {
+ throw new ConfigException(String.format("%s supports only file://
token URLs, got: %s", getClass().getSimpleName(), url));
+ }
+
+ Path path = Paths.get(uri);
+
+ if (!Files.isRegularFile(path)) {
+ throw new ConfigException(String.format("SASL/OAUTHBEARER token
file does not exist: %s", path));
+ }
+
+ return path;
+ }
+
+ /* Re-reads and reparses on every call. Retries briefly to ride out the
moment where kubelet atomically replaces the projected token file. */
+ private RangerOAuthBearerToken readToken(Path file) throws IOException {
+ IOException lastFailure = null;
+
+ for (int attempt = 1; attempt <= TOKEN_READ_MAX_ATTEMPTS; attempt++) {
+ try {
+ String jwt = new String(Files.readAllBytes(file),
StandardCharsets.UTF_8).trim();
+
+ if (jwt.isEmpty()) {
+ throw new IOException(String.format("SASL/OAUTHBEARER
token file is empty: %s", file));
+ }
+
+ return parseToken(jwt);
+ } catch (IOException e) {
+ lastFailure = e;
+
+ LOG.debug("{}: token read/parse failed (attempt {}/{}): {}",
getClass().getSimpleName(), attempt, TOKEN_READ_MAX_ATTEMPTS, e.getMessage());
+
+ if (attempt < TOKEN_READ_MAX_ATTEMPTS) {
+ sleepBeforeRetry();
+ }
+ }
+ }
+
+ throw new IOException(String.format("Failed to read SASL/OAUTHBEARER
token from %s after %d attempts", file, TOKEN_READ_MAX_ATTEMPTS), lastFailure);
+ }
+
+ private RangerOAuthBearerToken parseToken(String jwt) throws IOException {
+ String[] parts = jwt.split("\\.");
+
+ if (parts.length < 2) {
+ throw new IOException("Malformed JWT: expected
header.payload[.signature]");
+ }
+
+ JsonNode claims;
+
+ try {
+ claims = MAPPER.readTree(Base64.getUrlDecoder().decode(parts[1]));
+ } catch (IllegalArgumentException e) {
+ throw new IOException("Malformed JWT: payload is not base64url",
e);
+ }
+
+ JsonNode sub = claims.get(CLAIM_SUBJECT);
+ JsonNode exp = claims.get(CLAIM_EXPIRATION);
+ JsonNode iat = claims.get(CLAIM_ISSUED_AT);
+
+ if (sub == null || !sub.isTextual() ||
StringUtils.isBlank(sub.asText())) {
+ throw new IOException(String.format("JWT is missing required '%s'
claim", CLAIM_SUBJECT));
+ }
+
+ if (exp == null || !exp.isNumber()) {
+ throw new IOException(String.format("JWT is missing required
numeric '%s' claim", CLAIM_EXPIRATION));
+ }
+
+ Long startTimeMs = iat != null && iat.isNumber() ? toEpochMillis(iat)
: null;
+
+ return new RangerOAuthBearerToken(jwt, sub.asText(),
toEpochMillis(exp), startTimeMs);
+ }
+
+ private long toEpochMillis(JsonNode epochSeconds) {
+ return Math.round(epochSeconds.asDouble() * 1000);
+ }
+
+ private void sleepBeforeRetry() throws IOException {
+ try {
+ Thread.sleep(TOKEN_READ_RETRY_SLEEP_MS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+
+ throw new IOException("Interrupted while waiting to re-read
SASL/OAUTHBEARER token", e);
+ }
+ }
+}
diff --git
a/ranger-kafka-client-auth/src/main/java/org/apache/ranger/kafka/auth/RangerOAuthBearerToken.java
b/ranger-kafka-client-auth/src/main/java/org/apache/ranger/kafka/auth/RangerOAuthBearerToken.java
new file mode 100644
index 000000000..b9f923c64
--- /dev/null
+++
b/ranger-kafka-client-auth/src/main/java/org/apache/ranger/kafka/auth/RangerOAuthBearerToken.java
@@ -0,0 +1,69 @@
+/*
+ * 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.kafka.auth;
+
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerToken;
+
+import java.util.Collections;
+import java.util.Set;
+
+/**
+ * {@link OAuthBearerToken} backed by a JWT whose claims were already
extracted by the caller.
+ * Carries no scopes; {@code expiresOnTimeMs} is the {@code exp} claim and
{@code startTimeMs} the optional {@code iat},
+ * both in epoch milliseconds, which is what Kafka's credential refresh
scheduling expects.
+ */
+public final class RangerOAuthBearerToken implements OAuthBearerToken {
+ private final String value;
+ private final String principalName;
+ private final long expiresOnTimeMs;
+ private final Long startTimeMs;
+
+ public RangerOAuthBearerToken(String value, String principalName, long
expiresOnTimeMs, Long startTimeMs) {
+ this.value = value;
+ this.principalName = principalName;
+ this.expiresOnTimeMs = expiresOnTimeMs;
+ this.startTimeMs = startTimeMs;
+ }
+
+ @Override
+ public String value() {
+ return value;
+ }
+
+ @Override
+ public Set<String> scope() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public long lifetimeMs() {
+ return expiresOnTimeMs;
+ }
+
+ @Override
+ public String principalName() {
+ return principalName;
+ }
+
+ @Override
+ public Long startTimeMs() {
+ return startTimeMs;
+ }
+}
diff --git
a/ranger-kafka-client-auth/src/test/java/org/apache/ranger/kafka/auth/RangerKafkaClientSecurityConfigTest.java
b/ranger-kafka-client-auth/src/test/java/org/apache/ranger/kafka/auth/RangerKafkaClientSecurityConfigTest.java
new file mode 100644
index 000000000..feffe91d4
--- /dev/null
+++
b/ranger-kafka-client-auth/src/test/java/org/apache/ranger/kafka/auth/RangerKafkaClientSecurityConfigTest.java
@@ -0,0 +1,301 @@
+/*
+ * 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.kafka.auth;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.common.config.SaslConfigs;
+import
org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class RangerKafkaClientSecurityConfigTest {
+ private static final String PREFIX = "ranger.audit.ingestor";
+
+ @TempDir
+ Path tempDir;
+
+ @Test
+ public void testPlaintextByDefaultSetsNoSaslKeys() {
+ Map<String, Object> config = RangerKafkaClientSecurityConfig.build(new
Properties(), PREFIX);
+
+ assertEquals("PLAINTEXT",
config.get(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG));
+ assertFalse(config.containsKey(SaslConfigs.SASL_MECHANISM));
+ assertFalse(config.containsKey(SaslConfigs.SASL_JAAS_CONFIG));
+
assertFalse(config.containsKey(SaslConfigs.SASL_KERBEROS_SERVICE_NAME));
+ }
+
+ @Test
+ public void testSaslDefaultsToGssapiWithConfiguredKeytabOptions() throws
IOException {
+ Path keytab = Files.createFile(tempDir.resolve("ranger.keytab"));
+ Properties props = keytabLoginProps(keytab);
+
+ props.setProperty(PREFIX + ".service.kerberos.principal",
"rangeraudit/[email protected]");
+ props.setProperty(PREFIX + ".host", "Audit-Host.example.com");
+
+ Map<String, Object> config =
RangerKafkaClientSecurityConfig.build(props, PREFIX);
+
+ assertEquals("SASL_PLAINTEXT",
config.get(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG));
+ assertEquals("GSSAPI", config.get(SaslConfigs.SASL_MECHANISM));
+ assertEquals("kafka",
config.get(SaslConfigs.SASL_KERBEROS_SERVICE_NAME));
+ assertEquals(RangerKafkaClientSecurityConfig.KRB5_LOGIN_MODULE + "
required keyTab=\"" + keytab + "\" serviceName=\"kafka\"
principal=\"rangeraudit/[email protected]\" storeKey=\"true\"
useKeyTab=\"true\" useTicketCache=\"false\";",
+ config.get(SaslConfigs.SASL_JAAS_CONFIG));
+
assertFalse(config.containsKey(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS));
+ }
+
+ @Test
+ public void testGssapiWithoutOptionsOnlyPassesServiceCredentials() {
+ Properties props = new Properties();
+
+ props.setProperty(PREFIX + ".kafka.security.protocol",
"SASL_PLAINTEXT");
+ props.setProperty(PREFIX + ".service.kerberos.principal",
"[email protected]");
+ props.setProperty(PREFIX + ".service.kerberos.keytab",
"/etc/keytabs/missing.keytab");
+
+ Map<String, Object> config =
RangerKafkaClientSecurityConfig.build(props, PREFIX);
+
+ assertEquals(RangerKafkaClientSecurityConfig.KRB5_LOGIN_MODULE + "
required keyTab=\"/etc/keytabs/missing.keytab\" serviceName=\"kafka\"
principal=\"[email protected]\";",
+ config.get(SaslConfigs.SASL_JAAS_CONFIG));
+ }
+
+ @Test
+ public void testGssapiHonoursConfiguredServiceName() throws IOException {
+ Path keytab = Files.createFile(tempDir.resolve("ranger.keytab"));
+ Properties props = keytabLoginProps(keytab);
+
+ props.setProperty(PREFIX + ".kafka.security.protocol", "SASL_SSL");
+ props.setProperty(PREFIX + ".kafka.sasl.mechanism", "GSSAPI");
+ props.setProperty(PREFIX + ".kafka.sasl.kerberos.service.name",
"kafka-broker");
+ props.setProperty(PREFIX + ".service.kerberos.principal",
"[email protected]");
+ props.setProperty(PREFIX + ".service.kerberos.keytab",
keytab.toString());
+
+ Map<String, Object> config =
RangerKafkaClientSecurityConfig.build(props, PREFIX);
+
+ assertEquals("kafka-broker",
config.get(SaslConfigs.SASL_KERBEROS_SERVICE_NAME));
+
assertTrue(config.get(SaslConfigs.SASL_JAAS_CONFIG).toString().contains("serviceName=\"kafka-broker\"
principal=\"[email protected]\" storeKey=\"true\" useKeyTab=\"true\""));
+ }
+
+ @Test
+ public void testGssapiKeytabLoginRequiresPrincipalAndKeytab() {
+ Properties props = new Properties();
+
+ props.setProperty(PREFIX + ".kafka.security.protocol",
"SASL_PLAINTEXT");
+ props.setProperty(PREFIX + ".kafka.sasl.jaas.option.useKeyTab",
"true");
+ props.setProperty(PREFIX + ".service.kerberos.principal",
"[email protected]");
+
+ IllegalStateException e = assertThrows(IllegalStateException.class, ()
-> RangerKafkaClientSecurityConfig.build(props, PREFIX));
+
+ assertTrue(e.getMessage().contains(PREFIX +
".service.kerberos.keytab"), e.getMessage());
+ }
+
+ @Test
+ public void testGssapiKeytabLoginRequiresReadableKeytab() {
+ Properties props = new Properties();
+
+ props.setProperty(PREFIX + ".kafka.security.protocol",
"SASL_PLAINTEXT");
+ props.setProperty(PREFIX + ".kafka.sasl.jaas.option.useKeyTab",
"true");
+ props.setProperty(PREFIX + ".service.kerberos.principal",
"[email protected]");
+ props.setProperty(PREFIX + ".service.kerberos.keytab",
tempDir.resolve("missing.keytab").toString());
+
+ IllegalStateException e = assertThrows(IllegalStateException.class, ()
-> RangerKafkaClientSecurityConfig.build(props, PREFIX));
+
+ assertTrue(e.getMessage().contains("not found"), e.getMessage());
+ }
+
+ @Test
+ public void testGssapiTicketCacheNeedsNoKeytab() {
+ Properties props = new Properties();
+
+ props.setProperty(PREFIX + ".kafka.security.protocol",
"SASL_PLAINTEXT");
+ props.setProperty(PREFIX + ".kafka.sasl.jaas.option.useTicketCache",
"true");
+ props.setProperty(PREFIX + ".kafka.sasl.jaas.option.renewTGT", "true");
+
+ Map<String, Object> config =
RangerKafkaClientSecurityConfig.build(props, PREFIX);
+
+ assertEquals(RangerKafkaClientSecurityConfig.KRB5_LOGIN_MODULE + "
required serviceName=\"kafka\" renewTGT=\"true\" useTicketCache=\"true\";",
+ config.get(SaslConfigs.SASL_JAAS_CONFIG));
+ }
+
+ @Test
+ public void testJaasOptionsOverrideDefaultsAndResolveHost() throws
IOException {
+ Path keytab = Files.createFile(tempDir.resolve("other.keytab"));
+ Properties props = new Properties();
+
+ props.setProperty(PREFIX + ".kafka.security.protocol",
"SASL_PLAINTEXT");
+ props.setProperty(PREFIX + ".kafka.sasl.jaas.loginModuleControlFlag",
"requisite");
+ props.setProperty(PREFIX + ".kafka.sasl.jaas.option.useKeyTab",
"true");
+ props.setProperty(PREFIX + ".kafka.sasl.jaas.option.keyTab",
keytab.toString());
+ props.setProperty(PREFIX + ".kafka.sasl.jaas.option.principal",
"svc/[email protected]");
+ props.setProperty(PREFIX + ".kafka.sasl.jaas.option.storeKey",
"false");
+ props.setProperty(PREFIX + ".host", "node1.example.com");
+
+ Map<String, Object> config =
RangerKafkaClientSecurityConfig.build(props, PREFIX);
+ String jaas =
config.get(SaslConfigs.SASL_JAAS_CONFIG).toString();
+
+
assertTrue(jaas.startsWith(RangerKafkaClientSecurityConfig.KRB5_LOGIN_MODULE +
" requisite "), jaas);
+ assertTrue(jaas.contains("keyTab=\"" + keytab + "\""), jaas);
+ assertTrue(jaas.contains("storeKey=\"false\""), jaas);
+
assertTrue(jaas.contains("principal=\"svc/[email protected]\""),
jaas);
+ assertFalse(config.containsKey("sasl.jaas.option.keyTab"));
+ assertFalse(config.containsKey("sasl.jaas.loginModuleControlFlag"));
+ }
+
+ @Test
+ public void testOAuthBearerFileTokenUsesRangerHandler() throws IOException
{
+ Path token = Files.createFile(tempDir.resolve("token"));
+ Properties props = new Properties();
+
+ props.setProperty(PREFIX + ".kafka.security.protocol", "SASL_SSL");
+ props.setProperty(PREFIX + ".kafka.sasl.mechanism", "OAUTHBEARER");
+ props.setProperty(PREFIX +
".kafka.sasl.oauthbearer.token.endpoint.url", token.toUri().toString());
+
+ Map<String, Object> config =
RangerKafkaClientSecurityConfig.build(props, PREFIX);
+
+ assertEquals("OAUTHBEARER", config.get(SaslConfigs.SASL_MECHANISM));
+ assertEquals(RangerKafkaClientSecurityConfig.OAUTHBEARER_LOGIN_MODULE
+ " required;", config.get(SaslConfigs.SASL_JAAS_CONFIG));
+ assertEquals(token.toUri().toString(),
config.get(SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL));
+
assertEquals(RangerKafkaOAuthBearerLoginCallbackHandler.class.getName(),
config.get(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS));
+
assertFalse(config.containsKey(SaslConfigs.SASL_KERBEROS_SERVICE_NAME));
+ }
+
+ @Test
+ public void testOAuthBearerHttpEndpointUsesKafkaHandlerWithJaasOptions() {
+ Properties props = new Properties();
+
+ props.setProperty(PREFIX + ".kafka.security.protocol", "SASL_SSL");
+ props.setProperty(PREFIX + ".kafka.sasl.mechanism", "OAUTHBEARER");
+ props.setProperty(PREFIX +
".kafka.sasl.oauthbearer.token.endpoint.url",
"https://idp.example.com/oauth/token";);
+ props.setProperty(PREFIX + ".kafka.sasl.jaas.option.clientId",
"ranger");
+ props.setProperty(PREFIX + ".kafka.sasl.jaas.option.clientSecret",
"secret");
+
+ Map<String, Object> config =
RangerKafkaClientSecurityConfig.build(props, PREFIX);
+
+ assertEquals(OAuthBearerLoginCallbackHandler.class.getName(),
config.get(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS));
+ assertEquals(RangerKafkaClientSecurityConfig.OAUTHBEARER_LOGIN_MODULE
+ " required clientId=\"ranger\" clientSecret=\"secret\";",
config.get(SaslConfigs.SASL_JAAS_CONFIG));
+ }
+
+ @Test
+ public void testOAuthBearerRequiresTokenUrl() {
+ Properties props = new Properties();
+
+ props.setProperty(PREFIX + ".kafka.security.protocol", "SASL_SSL");
+ props.setProperty(PREFIX + ".kafka.sasl.mechanism", "OAUTHBEARER");
+
+ IllegalStateException e = assertThrows(IllegalStateException.class, ()
-> RangerKafkaClientSecurityConfig.build(props, PREFIX));
+
+ assertTrue(e.getMessage().contains(PREFIX +
".kafka.sasl.oauthbearer.token.endpoint.url"), e.getMessage());
+ }
+
+ @Test
+ public void testPassThroughOverridesDerivedValues() throws IOException {
+ Path token = Files.createFile(tempDir.resolve("token"));
+ Properties props = new Properties();
+
+ props.setProperty(PREFIX + ".kafka.security.protocol", "SASL_SSL");
+ props.setProperty(PREFIX + ".kafka.sasl.mechanism", "OAUTHBEARER");
+ props.setProperty(PREFIX +
".kafka.sasl.oauthbearer.token.endpoint.url", token.toUri().toString());
+ props.setProperty(PREFIX + ".kafka.sasl.login.callback.handler.class",
"com.example.CustomHandler");
+ props.setProperty(PREFIX + ".kafka.sasl.login.refresh.window.factor",
"0.7");
+ props.setProperty(PREFIX + ".kafka.ssl.truststore.location",
"/etc/ssl/truststore.jks");
+ props.setProperty(PREFIX + ".kafka.ssl.truststore.password", " ");
+ props.setProperty(PREFIX + ".kafka.bootstrap.servers", "kafka:9093");
+
+ Map<String, Object> config =
RangerKafkaClientSecurityConfig.build(props, PREFIX);
+
+ assertEquals("com.example.CustomHandler",
config.get(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS));
+ assertEquals("0.7",
config.get(SaslConfigs.SASL_LOGIN_REFRESH_WINDOW_FACTOR));
+ assertEquals("/etc/ssl/truststore.jks",
config.get("ssl.truststore.location"));
+ assertFalse(config.containsKey("ssl.truststore.password"));
+ assertFalse(config.containsKey("bootstrap.servers"));
+ }
+
+ @Test
+ public void testOtherMechanismsUseConfiguredLoginModuleOrExplicitJaas() {
+ Properties props = new Properties();
+
+ props.setProperty(PREFIX + ".kafka.security.protocol", "SASL_SSL");
+ props.setProperty(PREFIX + ".kafka.sasl.mechanism", "SCRAM-SHA-512");
+
+ assertThrows(IllegalStateException.class, () ->
RangerKafkaClientSecurityConfig.build(props, PREFIX));
+
+ props.setProperty(PREFIX + ".kafka.sasl.jaas.loginModuleName",
"org.apache.kafka.common.security.scram.ScramLoginModule");
+ props.setProperty(PREFIX + ".kafka.sasl.jaas.option.username",
"ranger");
+ props.setProperty(PREFIX + ".kafka.sasl.jaas.option.password",
"secret");
+
+ Map<String, Object> config =
RangerKafkaClientSecurityConfig.build(props, PREFIX);
+
+ assertEquals("SCRAM-SHA-512", config.get(SaslConfigs.SASL_MECHANISM));
+ assertEquals("org.apache.kafka.common.security.scram.ScramLoginModule
required password=\"secret\" username=\"ranger\";",
config.get(SaslConfigs.SASL_JAAS_CONFIG));
+
+ props.setProperty(PREFIX + ".kafka.sasl.jaas.config",
"org.apache.kafka.common.security.scram.ScramLoginModule required
username=\"explicit\" password=\"x\";");
+
+ config = RangerKafkaClientSecurityConfig.build(props, PREFIX);
+
+
assertTrue(config.get(SaslConfigs.SASL_JAAS_CONFIG).toString().contains("username=\"explicit\""));
+ }
+
+ @Test
+ public void testApplyAddsToExistingProperties() throws IOException {
+ Path keytab = Files.createFile(tempDir.resolve("ranger.keytab"));
+ Properties props = new Properties();
+ Properties target = new Properties();
+
+ props.setProperty(PREFIX + ".kafka.security.protocol",
"SASL_PLAINTEXT");
+ props.setProperty(PREFIX + ".service.kerberos.principal",
"[email protected]");
+ props.setProperty(PREFIX + ".service.kerberos.keytab",
keytab.toString());
+ target.put("bootstrap.servers", "kafka:9092");
+
+ RangerKafkaClientSecurityConfig.apply(props, PREFIX, target);
+
+ assertEquals("kafka:9092", target.get("bootstrap.servers"));
+ assertEquals("SASL_PLAINTEXT",
target.get(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG));
+ assertEquals("GSSAPI", target.get(SaslConfigs.SASL_MECHANISM));
+ }
+
+ /* SASL_PLAINTEXT with the keytab-login options the shipped site XML
carries */
+ private static Properties keytabLoginProps(Path keytab) {
+ Properties props = new Properties();
+
+ props.setProperty(PREFIX + ".kafka.security.protocol",
"SASL_PLAINTEXT");
+ props.setProperty(PREFIX + ".service.kerberos.keytab",
keytab.toString());
+ props.setProperty(PREFIX + ".kafka.sasl.jaas.option.useKeyTab",
"true");
+ props.setProperty(PREFIX + ".kafka.sasl.jaas.option.storeKey", "true");
+ props.setProperty(PREFIX + ".kafka.sasl.jaas.option.useTicketCache",
"false");
+
+ return props;
+ }
+
+ @Test
+ public void testResolvePrincipal() {
+ assertEquals("svc/host.example.com@REALM",
RangerKafkaClientSecurityConfig.resolvePrincipal("svc/_HOST@REALM",
"Host.Example.COM"));
+ assertEquals("svc/other@REALM",
RangerKafkaClientSecurityConfig.resolvePrincipal("svc/other@REALM",
"host.example.com"));
+ assertEquals("svc@REALM",
RangerKafkaClientSecurityConfig.resolvePrincipal("svc@REALM", null));
+
assertFalse(RangerKafkaClientSecurityConfig.resolvePrincipal("svc/_HOST@REALM",
null).contains("_HOST"));
+ }
+}
diff --git
a/ranger-kafka-client-auth/src/test/java/org/apache/ranger/kafka/auth/RangerKafkaOAuthBearerLoginCallbackHandlerTest.java
b/ranger-kafka-client-auth/src/test/java/org/apache/ranger/kafka/auth/RangerKafkaOAuthBearerLoginCallbackHandlerTest.java
new file mode 100644
index 000000000..144af71c3
--- /dev/null
+++
b/ranger-kafka-client-auth/src/test/java/org/apache/ranger/kafka/auth/RangerKafkaOAuthBearerLoginCallbackHandlerTest.java
@@ -0,0 +1,231 @@
+/*
+ * 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.kafka.auth;
+
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.security.auth.SaslExtensionsCallback;
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerToken;
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerTokenCallback;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.NameCallback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+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 RangerKafkaOAuthBearerLoginCallbackHandlerTest {
+ private static final String SUBJECT =
"system:serviceaccount:ranger:ranger-audit";
+
+ @TempDir
+ Path tempDir;
+
+ @Test
+ public void testTokenFromFileWithoutScopeClaim() throws Exception {
+ Path tokenFile = writeToken("token", buildJwt(SUBJECT,
1_700_000_000L, 1_700_003_600L));
+ String jwt = Files.readString(tokenFile);
+
+ try (RangerKafkaOAuthBearerLoginCallbackHandler handler =
configuredHandler(tokenFile)) {
+ OAuthBearerToken token = login(handler);
+
+ assertEquals(jwt, token.value());
+ assertEquals(SUBJECT, token.principalName());
+ assertEquals(1_700_003_600_000L, token.lifetimeMs());
+ assertEquals(Long.valueOf(1_700_000_000_000L),
token.startTimeMs());
+ assertTrue(token.scope().isEmpty());
+ }
+ }
+
+ @Test
+ public void testRotatedTokenIsReadOnNextLogin() throws Exception {
+ String jwtV1 = buildJwt(SUBJECT, 1_700_000_000L, 1_700_003_600L);
+ String jwtV2 = buildJwt(SUBJECT, 1_700_003_601L, 1_700_007_201L);
+ Path tokenFile = writeToken("token", jwtV1);
+
+ try (RangerKafkaOAuthBearerLoginCallbackHandler handler =
configuredHandler(tokenFile)) {
+ assertEquals(jwtV1, login(handler).value());
+
+ Files.write(tokenFile, jwtV2.getBytes(StandardCharsets.UTF_8));
+
+ OAuthBearerToken refreshed = login(handler);
+
+ assertEquals(jwtV2, refreshed.value());
+ assertEquals(1_700_007_201_000L, refreshed.lifetimeMs());
+ }
+ }
+
+ @Test
+ public void testTokenWithoutIssuedAtHasNoStartTime() throws Exception {
+ Path tokenFile = writeToken("token", buildJwt(SUBJECT, null,
1_700_003_600L));
+
+ try (RangerKafkaOAuthBearerLoginCallbackHandler handler =
configuredHandler(tokenFile)) {
+ assertNull(login(handler).startTimeMs());
+ }
+ }
+
+ @Test
+ public void testTokenWithoutExpirationIsRejected() throws Exception {
+ String payload = base64Url("{\"sub\":\"" + SUBJECT + "\"}");
+ Path tokenFile = writeToken("token",
base64Url("{\"alg\":\"RS256\"}") + "." + payload + ".sig");
+
+ try (RangerKafkaOAuthBearerLoginCallbackHandler handler =
configuredHandler(tokenFile)) {
+ IOException e = assertThrows(IOException.class, () ->
login(handler));
+
+ assertTrue(e.getCause().getMessage().contains("exp"),
e.getCause().getMessage());
+ }
+ }
+
+ @Test
+ public void testEmptyTokenFileIsRejected() throws Exception {
+ Path tokenFile = writeToken("token", "");
+
+ try (RangerKafkaOAuthBearerLoginCallbackHandler handler =
configuredHandler(tokenFile)) {
+ assertThrows(IOException.class, () -> login(handler));
+ }
+ }
+
+ @Test
+ public void testNonFileUrlIsRejected() {
+ Map<String, Object> configs = new HashMap<>();
+
+ configs.put(SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL,
"https://idp.example.com/token";);
+
+ assertThrows(ConfigException.class, () -> new
RangerKafkaOAuthBearerLoginCallbackHandler().configure(configs, "OAUTHBEARER",
Collections.emptyList()));
+ }
+
+ @Test
+ public void testMissingTokenFileIsRejectedAtConfigure() {
+ Map<String, Object> configs = new HashMap<>();
+
+ configs.put(SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL,
tempDir.resolve("missing").toUri().toString());
+
+ assertThrows(ConfigException.class, () -> new
RangerKafkaOAuthBearerLoginCallbackHandler().configure(configs, "OAUTHBEARER",
Collections.emptyList()));
+ }
+
+ @Test
+ public void testMissingUrlIsRejectedAtConfigure() {
+ assertThrows(ConfigException.class, () -> new
RangerKafkaOAuthBearerLoginCallbackHandler().configure(Collections.emptyMap(),
"OAUTHBEARER", Collections.emptyList()));
+ }
+
+ @Test
+ public void testAllowedUrlsSystemPropertyIsEnforced() throws Exception {
+ Path tokenFile = writeToken("token", buildJwt(SUBJECT,
1_700_000_000L, 1_700_003_600L));
+ String url = tokenFile.toUri().toString();
+
+
System.setProperty(RangerKafkaOAuthBearerLoginCallbackHandler.ALLOWED_URLS_SYSTEM_PROPERTY,
"file:///somewhere/else");
+
+ try {
+ assertThrows(ConfigException.class, () ->
configuredHandler(tokenFile));
+
+
System.setProperty(RangerKafkaOAuthBearerLoginCallbackHandler.ALLOWED_URLS_SYSTEM_PROPERTY,
"file:///somewhere/else, " + url);
+
+ try (RangerKafkaOAuthBearerLoginCallbackHandler handler =
configuredHandler(tokenFile)) {
+ assertEquals(SUBJECT, login(handler).principalName());
+ }
+ } finally {
+
System.clearProperty(RangerKafkaOAuthBearerLoginCallbackHandler.ALLOWED_URLS_SYSTEM_PROPERTY);
+ }
+ }
+
+ @Test
+ public void testExtensionsCallbackGetsEmptyExtensions() throws Exception {
+ Path tokenFile = writeToken("token", buildJwt(SUBJECT, 1_700_000_000L,
1_700_003_600L));
+
+ try (RangerKafkaOAuthBearerLoginCallbackHandler handler =
configuredHandler(tokenFile)) {
+ SaslExtensionsCallback callback = new SaslExtensionsCallback();
+
+ handler.handle(new Callback[] {callback});
+
+ assertTrue(callback.extensions().map().isEmpty());
+ }
+ }
+
+ @Test
+ public void testUnsupportedCallbackIsRejected() throws Exception {
+ Path tokenFile = writeToken("token", buildJwt(SUBJECT, 1_700_000_000L,
1_700_003_600L));
+
+ try (RangerKafkaOAuthBearerLoginCallbackHandler handler =
configuredHandler(tokenFile)) {
+ assertThrows(UnsupportedCallbackException.class, () ->
handler.handle(new Callback[] {new NameCallback("name")}));
+ }
+ }
+
+ @Test
+ public void testHandleBeforeConfigureFails() {
+ assertThrows(IllegalStateException.class, () -> new
RangerKafkaOAuthBearerLoginCallbackHandler().handle(new Callback[] {new
OAuthBearerTokenCallback()}));
+ }
+
+ private RangerKafkaOAuthBearerLoginCallbackHandler configuredHandler(Path
tokenFile) {
+ Map<String, Object> configs = new HashMap<>();
+
+ configs.put(SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL,
tokenFile.toUri().toString());
+
+ RangerKafkaOAuthBearerLoginCallbackHandler handler = new
RangerKafkaOAuthBearerLoginCallbackHandler();
+
+ handler.configure(configs, "OAUTHBEARER", Collections.emptyList());
+
+ return handler;
+ }
+
+ private static OAuthBearerToken
login(RangerKafkaOAuthBearerLoginCallbackHandler handler) throws Exception {
+ OAuthBearerTokenCallback callback = new OAuthBearerTokenCallback();
+
+ handler.handle(new Callback[] {callback});
+
+ return callback.token();
+ }
+
+ private Path writeToken(String name, String content) throws IOException {
+ Path file = tempDir.resolve(name);
+
+ Files.write(file, content.getBytes(StandardCharsets.UTF_8));
+
+ return file;
+ }
+
+ static String buildJwt(String subject, Long issuedAtSeconds, long
expirationSeconds) {
+ StringBuilder payload = new
StringBuilder("{\"sub\":\"").append(subject).append("\",\"exp\":").append(expirationSeconds);
+
+ if (issuedAtSeconds != null) {
+ payload.append(",\"iat\":").append(issuedAtSeconds);
+ }
+
+
payload.append(",\"iss\":\"https://kubernetes.default.svc.cluster.local\"}";);
+
+ return base64Url("{\"alg\":\"RS256\",\"typ\":\"JWT\"}") + "." +
base64Url(payload.toString()) + ".signature";
+ }
+
+ private static String base64Url(String value) {
+ return
Base64.getUrlEncoder().withoutPadding().encodeToString(value.getBytes(StandardCharsets.UTF_8));
+ }
+}
diff --git a/tagsync/conf/templates/installprop2xml.properties
b/tagsync/conf/templates/installprop2xml.properties
index 29fe1a459..68a10e9ce 100644
--- a/tagsync/conf/templates/installprop2xml.properties
+++ b/tagsync/conf/templates/installprop2xml.properties
@@ -22,6 +22,8 @@
TAG_SOURCE_ATLAS_KAFKA_BOOTSTRAP_SERVERS=atlas.kafka.bootstrap.servers
TAG_SOURCE_ATLAS_KAFKA_ENTITIES_GROUP_ID=atlas.kafka.entities.group.id
TAG_SOURCE_ATLAS_KAFKA_SERVICE_NAME=atlas.kafka.sasl.kerberos.service.name
TAG_SOURCE_ATLAS_KAFKA_SECURITY_PROTOCOL=atlas.kafka.security.protocol
+TAG_SOURCE_ATLAS_KAFKA_SASL_MECHANISM=atlas.kafka.sasl.mechanism
+TAG_SOURCE_ATLAS_KAFKA_OAUTH_TOKEN_URL=atlas.kafka.sasl.oauthbearer.token.endpoint.url
TAG_SOURCE_ATLAS_KERBEROS_PRINCIPAL=atlas.jaas.kafkaClient.option.principal
TAG_SOURCE_ATLAS_KERBEROS_KEYTAB=atlas.jaas.kafkaClient.option.keyTab
TAG_SOURCE_ATLASREST_ENABLED=ranger.tagsync.source.atlasrest
diff --git a/tagsync/pom.xml b/tagsync/pom.xml
index 7a45207ef..8bd74e724 100644
--- a/tagsync/pom.xml
+++ b/tagsync/pom.xml
@@ -322,6 +322,11 @@
<artifactId>ranger-common-ha</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.ranger</groupId>
+ <artifactId>ranger-kafka-client-auth</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.ranger</groupId>
<artifactId>ranger-metrics</artifactId>
diff --git a/tagsync/scripts/install.properties
b/tagsync/scripts/install.properties
index 517112417..fa80e7ac7 100644
--- a/tagsync/scripts/install.properties
+++ b/tagsync/scripts/install.properties
@@ -29,6 +29,10 @@ TAG_SOURCE_ATLAS_KAFKA_BOOTSTRAP_SERVERS=localhost:6667
TAG_SOURCE_ATLAS_KAFKA_ENTITIES_GROUP_ID=ranger_entities_consumer
TAG_SOURCE_ATLAS_KAFKA_SERVICE_NAME=kafka
TAG_SOURCE_ATLAS_KAFKA_SECURITY_PROTOCOL=PLAINTEXTSASL
+# SASL mechanism for the Atlas Kafka consumer; supported values: GSSAPI
(Kerberos keytab, default) and OAUTHBEARER (token)
+TAG_SOURCE_ATLAS_KAFKA_SASL_MECHANISM=GSSAPI
+# OAUTHBEARER only: file:// URL of a JWT that is re-read on every login (e.g.
Kubernetes projected service-account token) or http(s):// token endpoint for
Kafka's own client-credentials handler
+TAG_SOURCE_ATLAS_KAFKA_OAUTH_TOKEN_URL=
TAG_SOURCE_ATLAS_KERBEROS_PRINCIPAL=
TAG_SOURCE_ATLAS_KERBEROS_KEYTAB=
TAG_SOURCE_ATLASREST_ENABLED=false
diff --git a/tagsync/scripts/setup.py b/tagsync/scripts/setup.py
index 737cff54e..d9f1b06cf 100755
--- a/tagsync/scripts/setup.py
+++ b/tagsync/scripts/setup.py
@@ -73,6 +73,8 @@
TAG_SOURCE_ATLAS_KAKFA_SERVICE_NAME_KEY = 'TAG_SOURCE_ATLAS_KAFKA_SERVICE_NAME'
TAG_SOURCE_ATLAS_KAFKA_SECURITY_PROTOCOL_KEY =
'TAG_SOURCE_ATLAS_KAFKA_SECURITY_PROTOCOL'
+TAG_SOURCE_ATLAS_KAFKA_SASL_MECHANISM_KEY =
'TAG_SOURCE_ATLAS_KAFKA_SASL_MECHANISM'
+TAG_SOURCE_ATLAS_KAFKA_OAUTH_TOKEN_URL_KEY =
'TAG_SOURCE_ATLAS_KAFKA_OAUTH_TOKEN_URL'
TAG_SOURCE_ATLAS_KERBEROS_PRINCIPAL_KEY = 'TAG_SOURCE_ATLAS_KERBEROS_PRINCIPAL'
TAG_SOURCE_ATLAS_KERBEROS_KEYTAB_KEY = 'TAG_SOURCE_ATLAS_KERBEROS_KEYTAB'
TAGSYNC_ATLAS_TO_RANGER_SERVICE_MAPPING =
'ranger.tagsync.atlas.to.ranger.service.mapping'
@@ -89,6 +91,12 @@
TAG_SOURCE_FILE_ENABLED_KEY = 'TAG_SOURCE_FILE_ENABLED'
TAG_SOURCE_FILE_ENABLED = 'ranger.tagsync.source.file'
+KAFKA_SASL_MECHANISM_GSSAPI = 'GSSAPI'
+KAFKA_SASL_MECHANISM_OAUTHBEARER = 'OAUTHBEARER'
+KAFKA_OAUTHBEARER_LOGIN_MODULE =
'org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule'
+KAFKA_OAUTHBEARER_LOGIN_HANDLER =
'org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler'
+RANGER_OAUTHBEARER_LOGIN_HANDLER =
'org.apache.ranger.kafka.auth.RangerKafkaOAuthBearerLoginCallbackHandler'
+
hadoopConfFileName = 'core-site.xml'
ENV_HADOOP_CONF_FILE = "ranger-tagsync-env-hadoopconfdir.sh"
ENV_PID_FILE = 'ranger-tagsync-env-piddir.sh'
@@ -215,8 +223,8 @@ def updatePropertyInJCKSFile(jcksFileName,propName,value):
sys.exit(1)
return ret
-def atlas_kafka_uses_kerberos(props):
- """Kerberos JAAS for Atlas Kafka consumer only when SASL is configured
(not PLAINTEXT)."""
+def atlas_kafka_uses_sasl(props):
+ """SASL settings for the Atlas Kafka consumer only when a SASL protocol
is configured (not PLAINTEXT)."""
if not configure_security:
return False
protocol = props.get(TAG_SOURCE_ATLAS_KAFKA_SECURITY_PROTOCOL_KEY,
'PLAINTEXT')
@@ -224,6 +232,13 @@ def atlas_kafka_uses_kerberos(props):
return False
return protocol.strip().upper().startswith('SASL')
+def atlas_kafka_sasl_mechanism(props):
+ """GSSAPI unless TAG_SOURCE_ATLAS_KAFKA_SASL_MECHANISM says otherwise
(OAUTHBEARER for token files)."""
+ mechanism = props.get(TAG_SOURCE_ATLAS_KAFKA_SASL_MECHANISM_KEY, '')
+ if mechanism is None or mechanism.strip() == '':
+ return KAFKA_SASL_MECHANISM_GSSAPI
+ return mechanism.strip().upper()
+
def convertInstallPropsToXML(props):
directKeyMap =
getPropertiesConfigMap(join(installTemplateDirName,install2xmlMapFileName))
ret = {}
@@ -233,7 +248,11 @@ def convertInstallPropsToXML(props):
atlas_principal = ''
atlas_keytab = ''
- atlas_kafka_kerberos = atlas_kafka_uses_kerberos(props)
+ atlas_oauth_token_url = ''
+ atlas_kafka_sasl = atlas_kafka_uses_sasl(props)
+ atlas_kafka_mechanism = atlas_kafka_sasl_mechanism(props)
+ atlas_kafka_kerberos = atlas_kafka_sasl and atlas_kafka_mechanism ==
KAFKA_SASL_MECHANISM_GSSAPI
+ atlas_kafka_oauth = atlas_kafka_sasl and atlas_kafka_mechanism ==
KAFKA_SASL_MECHANISM_OAUTHBEARER
for k,v in props.items():
if (k in list(directKeyMap)):
@@ -252,6 +271,12 @@ def convertInstallPropsToXML(props):
atlas_principal = v
elif (atlas_kafka_kerberos and k ==
TAG_SOURCE_ATLAS_KERBEROS_KEYTAB_KEY):
atlas_keytab = v
+ elif (k == TAG_SOURCE_ATLAS_KAFKA_SASL_MECHANISM_KEY):
+ if atlas_kafka_sasl:
+ atlasOutFile.write(newKey + "=" +
atlas_kafka_mechanism + "\n")
+ elif (k == TAG_SOURCE_ATLAS_KAFKA_OAUTH_TOKEN_URL_KEY):
+ if atlas_kafka_oauth:
+ atlas_oauth_token_url = v.strip()
else:
ret[newKey] = v
else:
@@ -266,6 +291,19 @@ def convertInstallPropsToXML(props):
atlasOutFile.write("atlas.jaas.KafkaClient.option.keyTab = " +
atlas_keytab + "\n")
atlasOutFile.write("atlas.jaas.KafkaClient.option.principal = "
+ atlas_principal + "\n")
+ if atlas_kafka_oauth:
+ if atlas_oauth_token_url == '':
+ print("ERROR: %s must be set when %s is %s" %
(TAG_SOURCE_ATLAS_KAFKA_OAUTH_TOKEN_URL_KEY,
TAG_SOURCE_ATLAS_KAFKA_SASL_MECHANISM_KEY, KAFKA_SASL_MECHANISM_OAUTHBEARER))
+ sys.exit(1)
+ # file:// tokens rotate in place, so use the Ranger handler
that re-reads them on every login
+ if atlas_oauth_token_url.lower().startswith('file:'):
+ handler = RANGER_OAUTHBEARER_LOGIN_HANDLER
+ else:
+ handler = KAFKA_OAUTHBEARER_LOGIN_HANDLER
+
atlasOutFile.write("atlas.kafka.sasl.oauthbearer.token.endpoint.url=" +
atlas_oauth_token_url + "\n")
+
atlasOutFile.write("atlas.kafka.sasl.login.callback.handler.class=" + handler +
"\n")
+ atlasOutFile.write("atlas.kafka.sasl.jaas.config=" +
KAFKA_OAUTHBEARER_LOGIN_MODULE + " required;" + "\n")
+
atlasOutFile.close()
if (TAG_SOURCE_ATLAS_ENABLED_KEY in ret):
