You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ranger.apache.org by ra...@apache.org on 2026/06/29 10:24:36 UTC
(ranger) branch master updated: RANGER-4676, RANGER-5615: Add OpenSearch Dispatcher to Ranger Audit Server (#986)
This is an automated email from the ASF dual-hosted git repository.
ramackri 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 a23c30c60 RANGER-4676, RANGER-5615: Add OpenSearch Dispatcher to Ranger Audit Server (#986)
a23c30c60 is described below
commit a23c30c60418e1483f3f3ccb605a6b93e42f826e
Author: Paras agnihotri <ag...@gmail.com>
AuthorDate: Mon Jun 29 11:24:31 2026 +0100
RANGER-4676, RANGER-5615: Add OpenSearch Dispatcher to Ranger Audit Server (#986)
* RANGER-4676, RANGER-5615: Add OpenSearch audit destination via dedicated dispatcher module
* Update OpenSearchAuditDestination.java
* Update AuditOpenSearchDispatcher.java
* Update OpenSearchDispatcherManager.java
* Update OpenSearchUtilTest.java
* Update OpenSearchMgrTest.java
* Update OpenSearchAccessAuditsServiceTest.java
* Update OpenSearchUtil.java
* Update OpenSearchAccessAuditsService.java
* Update TestAuditOpenSearchDispatcher.java
---------
Co-authored-by: Paras <pa...@cloudera.com>
Co-authored-by: Madhan Neethiraj <ma...@apache.org>
---
.../audit/provider/AuditProviderFactory.java | 2 +
agents-audit/dest-os/pom.xml | 80 +++++
.../destination/OpenSearchAuditDestination.java | 321 +++++++++++++++++++
.../TestOpenSearchAuditDestination.java | 192 ++++++++++++
agents-audit/pom.xml | 1 +
.../audit-dispatcher/dispatcher-app/pom.xml | 6 +
.../audit-dispatcher/dispatcher-opensearch/pom.xml | 210 +++++++++++++
.../dispatcher/OpenSearchDispatcherManager.java | 239 ++++++++++++++
.../kafka/AuditOpenSearchDispatcher.java | 149 +++++++++
.../src/main/resources/conf/logback.xml | 55 ++++
.../ranger-audit-dispatcher-opensearch-site.xml | 228 ++++++++++++++
.../TestOpenSearchDispatcherManager.java | 80 +++++
.../kafka/TestAuditOpenSearchDispatcher.java | 116 +++++++
audit-server/audit-dispatcher/pom.xml | 1 +
.../scripts/start-audit-dispatcher.sh | 6 +-
dev-support/ranger-docker/Dockerfile.ranger-kdc | 5 +-
dev-support/ranger-docker/README.md | 48 +++
...-compose.ranger-audit-dispatcher-opensearch.yml | 70 +++++
.../ranger-docker/docker-compose.ranger.yml | 3 +
dev-support/ranger-docker/download-archives.sh | 10 +-
.../admin/ranger-admin-install-postgres.properties | 34 +-
.../ranger-audit-dispatcher-opensearch-site.xml | 207 +++++++++++++
.../ranger-docker/scripts/kdc/entrypoint.sh | 8 +-
.../scripts/opensearch/opensearch.yml | 3 +
distro/pom.xml | 12 +
distro/src/main/assembly/audit-dispatcher.xml | 32 ++
.../ranger/server/tomcat/EmbeddedServer.java | 14 +
.../server/tomcat/OpenSearchIndexBootStrapper.java | 219 +++++++++++++
.../conf/ranger_opensearch_schema.json | 136 ++++++++
security-admin/pom.xml | 5 +
security-admin/scripts/install.properties | 13 +
security-admin/scripts/setup.sh | 64 ++++
.../main/java/org/apache/ranger/biz/AssetMgr.java | 5 +
.../java/org/apache/ranger/biz/RangerBizUtil.java | 1 +
.../main/java/org/apache/ranger/biz/XAuditMgr.java | 7 +
.../ranger/elasticsearch/ElasticSearchMgr.java | 1 +
.../opensearch/OpenSearchAccessAuditsService.java | 312 +++++++++++++++++++
.../apache/ranger/opensearch/OpenSearchMgr.java | 199 ++++++++++++
.../apache/ranger/opensearch/OpenSearchUtil.java | 342 +++++++++++++++++++++
.../conf.dist/ranger-admin-default-site.xml | 28 ++
.../main/resources/conf.dist/ranger-admin-site.xml | 43 +++
.../OpenSearchAccessAuditsServiceTest.java | 185 +++++++++++
.../ranger/opensearch/OpenSearchMgrTest.java | 55 ++++
.../ranger/opensearch/OpenSearchUtilTest.java | 219 +++++++++++++
44 files changed, 3951 insertions(+), 15 deletions(-)
diff --git a/agents-audit/core/src/main/java/org/apache/ranger/audit/provider/AuditProviderFactory.java b/agents-audit/core/src/main/java/org/apache/ranger/audit/provider/AuditProviderFactory.java
index 2fca68627..ec42d89f1 100644
--- a/agents-audit/core/src/main/java/org/apache/ranger/audit/provider/AuditProviderFactory.java
+++ b/agents-audit/core/src/main/java/org/apache/ranger/audit/provider/AuditProviderFactory.java
@@ -433,6 +433,8 @@ private AuditHandler getProviderFromConfig(Properties props, String propPrefix,
provider = createDestination("org.apache.ranger.audit.destination.SolrAuditDestination");
} else if (providerName.equalsIgnoreCase("elasticsearch")) {
provider = createDestination("org.apache.ranger.audit.destination.ElasticSearchAuditDestination");
+ } else if (providerName.equalsIgnoreCase("opensearch")) {
+ provider = createDestination("org.apache.ranger.audit.destination.OpenSearchAuditDestination");
} else if (providerName.equalsIgnoreCase("amazon_cloudwatch")) {
provider = createDestination("org.apache.ranger.audit.destination.AmazonCloudWatchAuditDestination");
} else if (providerName.equalsIgnoreCase("kafka")) {
diff --git a/agents-audit/dest-os/pom.xml b/agents-audit/dest-os/pom.xml
new file mode 100644
index 000000000..d8068437d
--- /dev/null
+++ b/agents-audit/dest-os/pom.xml
@@ -0,0 +1,80 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<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>
+ </parent>
+
+ <artifactId>ranger-audit-dest-os</artifactId>
+ <packaging>jar</packaging>
+
+ <name>Apache Ranger - Audit Destination OpenSearch</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpasyncclient</artifactId>
+ <version>${httpcomponents.httpasyncclient.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ <version>${httpcomponents.httpclient.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpcore</artifactId>
+ <version>${httpcomponents.httpcore.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpcore-nio</artifactId>
+ <version>${httpcomponents.httpcore.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.ranger</groupId>
+ <artifactId>ranger-audit-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.elasticsearch.client</groupId>
+ <artifactId>elasticsearch-rest-client</artifactId>
+ <version>${elasticsearch.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ <version>${slf4j-api.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <version>${junit.jupiter.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/agents-audit/dest-os/src/main/java/org/apache/ranger/audit/destination/OpenSearchAuditDestination.java b/agents-audit/dest-os/src/main/java/org/apache/ranger/audit/destination/OpenSearchAuditDestination.java
new file mode 100644
index 000000000..2f1180b0c
--- /dev/null
+++ b/agents-audit/dest-os/src/main/java/org/apache/ranger/audit/destination/OpenSearchAuditDestination.java
@@ -0,0 +1,321 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.ranger.audit.destination;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthSchemeProvider;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.client.config.AuthSchemes;
+import org.apache.http.config.Lookup;
+import org.apache.http.config.RegistryBuilder;
+import org.apache.http.entity.ContentType;
+import org.apache.http.impl.auth.SPNegoSchemeFactory;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.nio.entity.NStringEntity;
+import org.apache.http.util.EntityUtils;
+import org.apache.ranger.audit.model.AuditEventBase;
+import org.apache.ranger.audit.model.AuthzAuditEvent;
+import org.apache.ranger.audit.provider.MiscUtil;
+import org.apache.ranger.authorization.credutils.CredentialsProviderUtil;
+import org.apache.ranger.authorization.credutils.kerberos.KerberosCredentialsProvider;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestClientBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.text.SimpleDateFormat;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Properties;
+import java.util.TimeZone;
+import java.util.UUID;
+
+public class OpenSearchAuditDestination extends AuditDestination {
+ private static final Logger LOG = LoggerFactory.getLogger(OpenSearchAuditDestination.class);
+
+ public static final String CONFIG_PREFIX = "ranger.audit.opensearch";
+ public static final String CONFIG_URLS = "urls";
+ public static final String CONFIG_PORT = "port";
+ public static final String CONFIG_USER = "user";
+ public static final String CONFIG_PASSWORD = "password";
+ public static final String CONFIG_PROTOCOL = "protocol";
+ public static final String CONFIG_INDEX = "index";
+ public static final String CONFIG_AUTH_TYPE = "authentication.type";
+ public static final String CONFIG_KERBEROS_PRINCIPAL = "kerberos.principal";
+ public static final String CONFIG_KERBEROS_KEYTAB = "kerberos.keytab";
+ public static final String DEFAULT_INDEX = "ranger_audits";
+
+ public static final String AUTH_TYPE_KERBEROS = "kerberos";
+ public static final String AUTH_TYPE_BASIC = "basic";
+ public static final String AUTH_TYPE_NONE = "none";
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+ private static final ThreadLocal<SimpleDateFormat> DATE_FORMAT = ThreadLocal.withInitial(() -> {
+ SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
+
+ sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
+
+ return sdf;
+ });
+
+ private volatile RestClient client;
+
+ private String index;
+ private String user;
+ private String password;
+ private String protocol;
+ private String urls;
+ private int port;
+ private String authType;
+ private String kerberosPrincipal;
+ private String kerberosKeytab;
+
+ public OpenSearchAuditDestination() {
+ propPrefix = CONFIG_PREFIX;
+ }
+
+ @Override
+ public void init(Properties props, String propPrefix) {
+ super.init(props, propPrefix);
+
+ this.urls = MiscUtil.getStringProperty(props, propPrefix + "." + CONFIG_URLS, "localhost");
+ this.port = MiscUtil.getIntProperty(props, propPrefix + "." + CONFIG_PORT, 9200);
+ this.protocol = MiscUtil.getStringProperty(props, propPrefix + "." + CONFIG_PROTOCOL, "http");
+ this.user = MiscUtil.getStringProperty(props, propPrefix + "." + CONFIG_USER, "");
+ this.password = MiscUtil.getStringProperty(props, propPrefix + "." + CONFIG_PASSWORD, "");
+ this.index = MiscUtil.getStringProperty(props, propPrefix + "." + CONFIG_INDEX, DEFAULT_INDEX);
+
+ this.authType = MiscUtil.getStringProperty(props, propPrefix + "." + CONFIG_AUTH_TYPE, "");
+ this.kerberosPrincipal = MiscUtil.getStringProperty(props, propPrefix + "." + CONFIG_KERBEROS_PRINCIPAL, "");
+ this.kerberosKeytab = MiscUtil.getStringProperty(props, propPrefix + "." + CONFIG_KERBEROS_KEYTAB, "");
+
+ LOG.info("OpenSearchAuditDestination.init(): urls={}, port={}, index={}", urls, port, index);
+
+ getClient();
+ }
+
+ @Override
+ public void stop() {
+ logStatus();
+
+ if (client != null) {
+ try {
+ client.close();
+ } catch (Exception e) {
+ LOG.error("Error closing OpenSearch client", e);
+ }
+ }
+ }
+
+ @Override
+ public void flush() {
+ }
+
+ @Override
+ public boolean log(Collection<AuditEventBase> events) {
+ if (events == null || events.isEmpty()) {
+ return true;
+ }
+
+ boolean ret = false;
+ RestClient currentClient = getClient();
+
+ if (currentClient == null) {
+ LOG.error("OpenSearch client is null. Cannot write audit events.");
+ } else {
+ try {
+ StringBuilder bulk = new StringBuilder();
+
+ for (AuditEventBase event : events) {
+ AuthzAuditEvent auditEvent = (AuthzAuditEvent) event;
+ Map<String, Object> doc = toDoc(auditEvent);
+ String id = (String) doc.get("id");
+
+ if (StringUtils.isBlank(id)) {
+ id = UUID.randomUUID().toString();
+
+ doc.put("id", id);
+ }
+
+ Map<String, Object> indexProps = new HashMap<>();
+
+ indexProps.put("_index", index);
+ indexProps.put("_id", id);
+
+ bulk.append(MAPPER.writeValueAsString(Map.of("index", indexProps))).append('\n');
+ bulk.append(MAPPER.writeValueAsString(doc)).append('\n');
+ }
+
+ Request request = new Request("POST", "/_bulk");
+
+ request.setEntity(new NStringEntity(bulk.toString(), ContentType.create("application/x-ndjson", StandardCharsets.UTF_8)));
+
+ Response response = currentClient.performRequest(request);
+ int statusCode = response.getStatusLine().getStatusCode();
+
+ if (statusCode >= 400) {
+ LOG.error("OpenSearch bulk request failed: HTTP {}", statusCode);
+ } else {
+ String responseBody = EntityUtils.toString(response.getEntity());
+
+ @SuppressWarnings("unchecked")
+ Map<String, Object> responseMap = MAPPER.readValue(responseBody, Map.class);
+
+ if (Boolean.TRUE.equals(responseMap.get("errors"))) {
+ LOG.error("OpenSearch bulk response contains item-level errors");
+ } else {
+ ret = true;
+ }
+ }
+ } catch (Exception e) {
+ LOG.error("Failed to write audit events to OpenSearch", e);
+ }
+ }
+
+ if (ret) {
+ addSuccessCount(events.size());
+ } else {
+ addFailedCount(events.size());
+ }
+
+ return ret;
+ }
+
+ public boolean isAsync() {
+ return true;
+ }
+
+ synchronized RestClient getClient() {
+ if (client == null) {
+ if (StringUtils.isBlank(urls) || "NONE".equalsIgnoreCase(urls)) {
+ LOG.error("OpenSearch URLs not configured");
+
+ return null;
+ }
+
+ HttpHost[] hosts = Arrays.stream(urls.split(",")).map(String::trim).filter(h -> !h.isEmpty()).map(h -> new HttpHost(h, port, protocol)).toArray(HttpHost[]::new);
+ RestClientBuilder builder = RestClient.builder(hosts);
+ String resolvedAuthType = resolveAuthType(authType, user, password);
+
+ if (AUTH_TYPE_KERBEROS.equals(resolvedAuthType)) {
+ String principal = isConfigured(kerberosPrincipal) ? kerberosPrincipal : user;
+ String keytab = isConfigured(kerberosKeytab) ? kerberosKeytab : password;
+
+ KerberosCredentialsProvider credentialsProvider = CredentialsProviderUtil.getKerberosCredentials(principal, keytab);
+ Lookup<AuthSchemeProvider> authRegistry = RegistryBuilder.<AuthSchemeProvider>create().register(AuthSchemes.SPNEGO, new SPNegoSchemeFactory()).build();
+
+ builder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider).setDefaultAuthSchemeRegistry(authRegistry));
+
+ LOG.info("OpenSearch client configured with Kerberos authentication for principal: {}", principal);
+ } else if (AUTH_TYPE_BASIC.equals(resolvedAuthType)) {
+ CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
+
+ credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(user, password));
+
+ builder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider));
+
+ LOG.info("OpenSearch client configured with basic authentication for user: {}", user);
+ } else {
+ LOG.info("OpenSearch client configured without authentication");
+ }
+
+ client = builder.build();
+ }
+
+ return client;
+ }
+
+ public static boolean isConfigured(final String value) {
+ return StringUtils.isNotBlank(value) && !"NONE".equalsIgnoreCase(value.trim());
+ }
+
+ /**
+ * Resolves the authentication scheme. When {@code authentication.type} is set explicitly
+ * ({@value AUTH_TYPE_KERBEROS}/{@value AUTH_TYPE_BASIC}/{@value AUTH_TYPE_NONE}) it is honored as-is.
+ * When unset, the type is inferred for backward compatibility: a password pointing to an existing
+ * keytab file selects Kerberos, a user+password pair selects basic auth, otherwise no authentication.
+ */
+ public static String resolveAuthType(final String authType, final String user, final String password) {
+ if (StringUtils.isNotBlank(authType)) {
+ return authType.trim().toLowerCase(Locale.ROOT);
+ }
+
+ if (isConfigured(user) && isConfigured(password)) {
+ if (password.contains("keytab") && new File(password).exists()) {
+ return AUTH_TYPE_KERBEROS;
+ }
+
+ return AUTH_TYPE_BASIC;
+ }
+
+ return AUTH_TYPE_NONE;
+ }
+
+ Map<String, Object> toDoc(AuthzAuditEvent event) {
+ Map<String, Object> doc = new HashMap<>();
+
+ doc.put("id", event.getEventId());
+ doc.put("access", event.getAccessType());
+ doc.put("enforcer", event.getAclEnforcer());
+ doc.put("agent", event.getAgentId());
+ doc.put("repo", event.getRepositoryName());
+ doc.put("sess", event.getSessionId());
+ doc.put("reqUser", event.getUser());
+ doc.put("reqData", event.getRequestData());
+ doc.put("resource", event.getResourcePath());
+ doc.put("cliIP", event.getClientIP());
+ doc.put("logType", event.getLogType());
+ doc.put("result", event.getAccessResult());
+ doc.put("policy", event.getPolicyId());
+ doc.put("repoType", event.getRepositoryType());
+ doc.put("resType", event.getResourceType());
+ doc.put("reason", event.getResultReason());
+ doc.put("action", event.getAction());
+ doc.put("evtTime", formatDate(event.getEventTime()));
+ doc.put("seq_num", event.getSeqNum());
+ doc.put("event_count", event.getEventCount());
+ doc.put("event_dur_ms", event.getEventDurationMS());
+ doc.put("tags", event.getTags());
+ doc.put("datasets", event.getDatasets());
+ doc.put("projects", event.getProjects());
+ doc.put("cluster", event.getClusterName());
+ doc.put("zoneName", event.getZoneName());
+ doc.put("agentHost", event.getAgentHostname());
+ doc.put("policyVersion", event.getPolicyVersion());
+
+ return doc;
+ }
+
+ private static String formatDate(Date date) {
+ return date != null ? DATE_FORMAT.get().format(date) : null;
+ }
+}
diff --git a/agents-audit/dest-os/src/test/java/org/apache/ranger/audit/destination/TestOpenSearchAuditDestination.java b/agents-audit/dest-os/src/test/java/org/apache/ranger/audit/destination/TestOpenSearchAuditDestination.java
new file mode 100644
index 000000000..1bba1a67c
--- /dev/null
+++ b/agents-audit/dest-os/src/test/java/org/apache/ranger/audit/destination/TestOpenSearchAuditDestination.java
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.ranger.audit.destination;
+
+import org.apache.ranger.audit.model.AuthzAuditEvent;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Date;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestOpenSearchAuditDestination {
+ private OpenSearchAuditDestination destination;
+
+ @BeforeEach
+ void setUp() {
+ destination = new OpenSearchAuditDestination();
+ }
+
+ @Test
+ void toDoc_mapsAllFieldsCorrectly() {
+ AuthzAuditEvent event = new AuthzAuditEvent();
+ event.setEventId("test-event-001");
+ event.setAccessType("read");
+ event.setAclEnforcer("ranger-acl");
+ event.setAgentId("hdfs-agent");
+ event.setRepositoryName("dev_hdfs");
+ event.setSessionId("sess-123");
+ event.setUser("testuser");
+ event.setRequestData("/tmp/testfile");
+ event.setResourcePath("/data/warehouse");
+ event.setClientIP("192.168.1.10");
+ event.setLogType("RangerAudit");
+ event.setAccessResult((short) 1);
+ event.setPolicyId(42L);
+ event.setRepositoryType(1);
+ event.setResourceType("path");
+ event.setResultReason("Allowed by policy");
+ event.setAction("read");
+ event.setEventTime(new Date(1700000000000L));
+ event.setSeqNum(10L);
+ event.setEventCount(1L);
+ event.setEventDurationMS(50L);
+ event.setClusterName("test-cluster");
+ event.setZoneName("zone1");
+ event.setAgentHostname("host1.example.com");
+ event.setPolicyVersion(5L);
+
+ Map<String, Object> doc = destination.toDoc(event);
+
+ assertEquals("test-event-001", doc.get("id"));
+ assertEquals("read", doc.get("access"));
+ assertEquals("ranger-acl", doc.get("enforcer"));
+ assertEquals("hdfs-agent", doc.get("agent"));
+ assertEquals("dev_hdfs", doc.get("repo"));
+ assertEquals("sess-123", doc.get("sess"));
+ assertEquals("testuser", doc.get("reqUser"));
+ assertEquals("/tmp/testfile", doc.get("reqData"));
+ assertEquals("/data/warehouse", doc.get("resource"));
+ assertEquals("192.168.1.10", doc.get("cliIP"));
+ assertEquals("RangerAudit", doc.get("logType"));
+ assertEquals((short) 1, doc.get("result"));
+ assertEquals(42L, doc.get("policy"));
+ assertEquals(1, doc.get("repoType"));
+ assertEquals("path", doc.get("resType"));
+ assertEquals("Allowed by policy", doc.get("reason"));
+ assertEquals("read", doc.get("action"));
+ assertNotNull(doc.get("evtTime"));
+ assertTrue(doc.get("evtTime").toString().contains("2023-11-14"));
+ assertEquals(10L, doc.get("seq_num"));
+ assertEquals(1L, doc.get("event_count"));
+ assertEquals(50L, doc.get("event_dur_ms"));
+ assertEquals("test-cluster", doc.get("cluster"));
+ assertEquals("zone1", doc.get("zoneName"));
+ assertEquals("host1.example.com", doc.get("agentHost"));
+ assertEquals(5L, doc.get("policyVersion"));
+ }
+
+ @Test
+ void toDoc_nullEventId_remainsNull() {
+ AuthzAuditEvent event = new AuthzAuditEvent();
+ event.setUser("testuser");
+
+ Map<String, Object> doc = destination.toDoc(event);
+
+ assertNull(doc.get("id"));
+ }
+
+ @Test
+ void toDoc_nullEventTime_formatsAsNull() {
+ AuthzAuditEvent event = new AuthzAuditEvent();
+ event.setEventId("evt-1");
+ event.setEventTime(null);
+
+ Map<String, Object> doc = destination.toDoc(event);
+
+ assertNull(doc.get("evtTime"));
+ }
+
+ @Test
+ void getClient_noneUrls_returnsNull() {
+ java.util.Properties props = new java.util.Properties();
+ props.setProperty(OpenSearchAuditDestination.CONFIG_PREFIX + ".urls", "NONE");
+
+ destination.init(props, OpenSearchAuditDestination.CONFIG_PREFIX);
+
+ assertNull(destination.getClient());
+ }
+
+ @Test
+ void log_nullClient_returnsFalse() {
+ java.util.Properties props = new java.util.Properties();
+ props.setProperty(OpenSearchAuditDestination.CONFIG_PREFIX + ".urls", "NONE");
+ destination.init(props, OpenSearchAuditDestination.CONFIG_PREFIX);
+
+ AuthzAuditEvent event = new AuthzAuditEvent();
+ event.setEventId("test-1");
+ event.setUser("user1");
+
+ java.util.Collection<org.apache.ranger.audit.model.AuditEventBase> events = java.util.List.of(event);
+ boolean result = destination.log(events);
+
+ assertFalse(result);
+ }
+
+ @Test
+ void log_emptyEvents_returnsTrue() {
+ java.util.Collection<org.apache.ranger.audit.model.AuditEventBase> events = java.util.Collections.emptyList();
+ boolean result = destination.log(events);
+
+ assertTrue(result);
+ }
+
+ @Test
+ void log_nullEvents_returnsTrue() {
+ boolean result = destination.log((java.util.Collection<org.apache.ranger.audit.model.AuditEventBase>) null);
+
+ assertTrue(result);
+ }
+
+ @Test
+ void configConstants_matchExpectedValues() {
+ assertEquals("ranger.audit.opensearch", OpenSearchAuditDestination.CONFIG_PREFIX);
+ assertEquals("urls", OpenSearchAuditDestination.CONFIG_URLS);
+ assertEquals("port", OpenSearchAuditDestination.CONFIG_PORT);
+ assertEquals("user", OpenSearchAuditDestination.CONFIG_USER);
+ assertEquals("password", OpenSearchAuditDestination.CONFIG_PASSWORD);
+ assertEquals("protocol", OpenSearchAuditDestination.CONFIG_PROTOCOL);
+ assertEquals("index", OpenSearchAuditDestination.CONFIG_INDEX);
+ assertEquals("authentication.type", OpenSearchAuditDestination.CONFIG_AUTH_TYPE);
+ assertEquals("kerberos.principal", OpenSearchAuditDestination.CONFIG_KERBEROS_PRINCIPAL);
+ assertEquals("kerberos.keytab", OpenSearchAuditDestination.CONFIG_KERBEROS_KEYTAB);
+ assertEquals("ranger_audits", OpenSearchAuditDestination.DEFAULT_INDEX);
+ }
+
+ @Test
+ void resolveAuthType_explicitValuesAreHonored() {
+ assertEquals("kerberos", OpenSearchAuditDestination.resolveAuthType("kerberos", "", ""));
+ assertEquals("basic", OpenSearchAuditDestination.resolveAuthType("BASIC", "", ""));
+ assertEquals("none", OpenSearchAuditDestination.resolveAuthType("none", "u", "p"));
+ }
+
+ @Test
+ void resolveAuthType_inferredWhenTypeUnset() {
+ assertEquals("basic", OpenSearchAuditDestination.resolveAuthType("", "admin", "secret"));
+ assertEquals("none", OpenSearchAuditDestination.resolveAuthType("", "", ""));
+ assertEquals("none", OpenSearchAuditDestination.resolveAuthType("", "NONE", "NONE"));
+ }
+}
diff --git a/agents-audit/pom.xml b/agents-audit/pom.xml
index a25edde92..173f4f80b 100644
--- a/agents-audit/pom.xml
+++ b/agents-audit/pom.xml
@@ -37,6 +37,7 @@
<module>dest-hdfs</module>
<module>dest-kafka</module>
<module>dest-log4j</module>
+ <module>dest-os</module>
<module>dest-solr</module>
<module>orc-util</module>
</modules>
diff --git a/audit-server/audit-dispatcher/dispatcher-app/pom.xml b/audit-server/audit-dispatcher/dispatcher-app/pom.xml
index d18d487e3..161256d34 100644
--- a/audit-server/audit-dispatcher/dispatcher-app/pom.xml
+++ b/audit-server/audit-dispatcher/dispatcher-app/pom.xml
@@ -85,6 +85,12 @@
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.ranger</groupId>
+ <artifactId>audit-dispatcher-opensearch</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>org.apache.ranger</groupId>
<artifactId>audit-dispatcher-solr</artifactId>
diff --git a/audit-server/audit-dispatcher/dispatcher-opensearch/pom.xml b/audit-server/audit-dispatcher/dispatcher-opensearch/pom.xml
new file mode 100644
index 000000000..f878698bb
--- /dev/null
+++ b/audit-server/audit-dispatcher/dispatcher-opensearch/pom.xml
@@ -0,0 +1,210 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<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>
+ </parent>
+
+ <artifactId>audit-dispatcher-opensearch</artifactId>
+ <packaging>jar</packaging>
+ <name>Ranger Audit Dispatcher OpenSearch</name>
+ <description>Kafka dispatcher service for indexing audits into OpenSearch/ElasticSearch</description>
+
+ <properties>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ </properties>
+
+ <dependencies>
+ <!-- Jackson (ObjectMapper for JSON serialization) -->
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>${fasterxml.jackson.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
+ <version>${commons.lang3.version}</version>
+ </dependency>
+
+ <!-- HTTP Components (used by RestClient for OpenSearch communication) -->
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpasyncclient</artifactId>
+ <version>${httpcomponents.httpasyncclient.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>commons-logging</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ <version>${httpcomponents.httpclient.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>commons-logging</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpcore</artifactId>
+ <version>${httpcomponents.httpcore.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpcore-nio</artifactId>
+ <version>${httpcomponents.httpcore.version}</version>
+ </dependency>
+
+ <!-- OpenSearch audit destination (connection, auth, bulk write + doc mapping) -->
+ <dependency>
+ <groupId>org.apache.ranger</groupId>
+ <artifactId>ranger-audit-dest-os</artifactId>
+ <version>${project.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client-api</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <!-- Provided by dispatcher-app at runtime -->
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>${kafka.version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.ranger</groupId>
+ <artifactId>ranger-audit-core</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client-api</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.ranger</groupId>
+ <artifactId>ranger-audit-dispatcher-common</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client-api</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.ranger</groupId>
+ <artifactId>ranger-audit-server-common</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ <version>${slf4j.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Test -->
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <version>${junit.jupiter.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <version>${mockito.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-junit-jupiter</artifactId>
+ <version>${mockito.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <finalName>audit-dispatcher-opensearch-${project.version}</finalName>
+ <resources>
+ <resource>
+ <filtering>true</filtering>
+ <directory>src/main/resources</directory>
+ </resource>
+ </resources>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-pmd-plugin</artifactId>
+ <configuration>
+ <rulesets>
+ <ruleset>${project.parent.basedir}/dev-support/ranger-pmd-ruleset.xml</ruleset>
+ </rulesets>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>copy-dependencies</id>
+ <goals>
+ <goal>copy-dependencies</goal>
+ </goals>
+ <phase>package</phase>
+ <configuration>
+ <outputDirectory>${project.build.directory}/lib</outputDirectory>
+ <includeScope>runtime</includeScope>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git a/audit-server/audit-dispatcher/dispatcher-opensearch/src/main/java/org/apache/ranger/audit/dispatcher/OpenSearchDispatcherManager.java b/audit-server/audit-dispatcher/dispatcher-opensearch/src/main/java/org/apache/ranger/audit/dispatcher/OpenSearchDispatcherManager.java
new file mode 100644
index 000000000..bca0bb2ab
--- /dev/null
+++ b/audit-server/audit-dispatcher/dispatcher-opensearch/src/main/java/org/apache/ranger/audit/dispatcher/OpenSearchDispatcherManager.java
@@ -0,0 +1,239 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.ranger.audit.dispatcher;
+
+import org.apache.ranger.audit.dispatcher.kafka.AuditDispatcher;
+import org.apache.ranger.audit.dispatcher.kafka.AuditDispatcherTracker;
+import org.apache.ranger.audit.dispatcher.kafka.AuditOpenSearchDispatcher;
+import org.apache.ranger.audit.provider.MiscUtil;
+import org.apache.ranger.audit.server.AuditServerConstants;
+import org.apache.ranger.audit.utils.AuditServerLogFormatter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Properties;
+
+public final class OpenSearchDispatcherManager {
+ private static final Logger LOG = LoggerFactory.getLogger(OpenSearchDispatcherManager.class);
+
+ private static final String CONFIG_DISPATCHER_TYPE = AuditServerConstants.PROP_DISPATCHER_TYPE;
+ private static final String TYPE_OPENSEARCH = "opensearch";
+ private static final String OPENSEARCH_DEST_PROP = "xasecure.audit.destination.opensearch";
+ private static final int MAX_INIT_ATTEMPTS = 5;
+ private static final long INIT_RETRY_MS = 5000L;
+ private static final long SHUTDOWN_WAIT_MS = 10000L;
+
+ private final AuditDispatcherTracker tracker = AuditDispatcherTracker.getInstance();
+
+ private AuditDispatcher dispatcher;
+ private Thread dispatcherThread;
+
+ public void init(final Properties props) {
+ LOG.info("==> OpenSearchDispatcherManager.init()");
+
+ String dispatcherType = System.getProperty(CONFIG_DISPATCHER_TYPE);
+
+ if (dispatcherType != null && !dispatcherType.equalsIgnoreCase(TYPE_OPENSEARCH)) {
+ LOG.info("Skipping OpenSearchDispatcherManager initialization since dispatcher type is {}", dispatcherType);
+
+ return;
+ }
+
+ try {
+ if (props == null) {
+ LOG.error("Configuration properties are null");
+
+ throw new RuntimeException("Failed to load configuration");
+ }
+
+ boolean isEnabled = MiscUtil.getBooleanProperty(props, OPENSEARCH_DEST_PROP, false);
+
+ if (!isEnabled) {
+ String clsName = MiscUtil.getStringProperty(props, AuditServerConstants.PROP_DISPATCHER_CLASS);
+
+ if (clsName != null && clsName.contains("AuditOpenSearchDispatcher")) {
+ isEnabled = true;
+ }
+ }
+
+ if (!isEnabled) {
+ LOG.warn("OpenSearch destination is disabled ({}=false). No dispatchers will be created.", OPENSEARCH_DEST_PROP);
+
+ return;
+ }
+
+ initializeDispatcher(props, AuditServerConstants.PROP_DISPATCHER_PREFIX);
+
+ if (dispatcher == null) {
+ throw new RuntimeException("No OpenSearch dispatcher was created. Verify that " + OPENSEARCH_DEST_PROP + "=true and classes are configured correctly.");
+ } else {
+ LOG.info("Created OpenSearch dispatcher");
+
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ LOG.info("JVM shutdown detected, stopping OpenSearchDispatcherManager");
+
+ shutdown();
+ }, "OpenSearchDispatcher-ShutdownHook"));
+
+ startDispatcher();
+ }
+ } catch (Exception e) {
+ LOG.error("Failed to initialize OpenSearchDispatcherManager", e);
+
+ throw new RuntimeException("Failed to initialize OpenSearchDispatcherManager", e);
+ }
+
+ LOG.info("<== OpenSearchDispatcherManager.init()");
+ }
+
+ public void shutdown() {
+ LOG.info("==> OpenSearchDispatcherManager.shutdown()");
+
+ if (dispatcher != null) {
+ try {
+ LOG.info("Shutting down dispatcher: {}", dispatcher.getClass().getSimpleName());
+
+ dispatcher.shutdown();
+
+ LOG.info("Dispatcher shutdown completed: {}", dispatcher.getClass().getSimpleName());
+ } catch (Exception e) {
+ LOG.error("Error shutting down dispatcher: {}", dispatcher.getClass().getSimpleName(), e);
+ }
+ }
+
+ if (dispatcherThread != null && dispatcherThread.isAlive()) {
+ try {
+ LOG.info("Waiting for thread to terminate: {}", dispatcherThread.getName());
+
+ dispatcherThread.join(SHUTDOWN_WAIT_MS);
+
+ if (dispatcherThread.isAlive()) {
+ LOG.warn("Thread did not terminate within {}ms: {}", SHUTDOWN_WAIT_MS, dispatcherThread.getName());
+ }
+ } catch (InterruptedException e) {
+ LOG.warn("Interrupted while waiting for thread to terminate: {}", dispatcherThread.getName(), e);
+
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ dispatcher = null;
+ dispatcherThread = null;
+
+ tracker.clearActiveDispatcher(TYPE_OPENSEARCH);
+
+ LOG.info("<== OpenSearchDispatcherManager.shutdown() - OpenSearch dispatcher stopped");
+ }
+
+ private void initializeDispatcher(final Properties props, final String propPrefix) {
+ LOG.info("==> OpenSearchDispatcherManager.initializeDispatcher()");
+
+ String clsStr = MiscUtil.getStringProperty(props, AuditServerConstants.PROP_DISPATCHER_CLASS, AuditOpenSearchDispatcher.class.getName());
+ String className = clsStr.split(",")[0].trim();
+
+ if (className.isEmpty()) {
+ LOG.error("Dispatcher class name is empty");
+
+ return;
+ }
+
+ long retryDelay = INIT_RETRY_MS;
+
+ for (int attempt = 1; attempt <= MAX_INIT_ATTEMPTS; attempt++) {
+ try {
+ Class<?> cls = Class.forName(className);
+
+ dispatcher = (AuditDispatcher) cls.getConstructor(Properties.class, String.class).newInstance(props, propPrefix);
+
+ tracker.addActiveDispatcher(TYPE_OPENSEARCH, dispatcher);
+
+ LOG.info("Successfully initialized dispatcher class: {}", cls.getName());
+
+ break;
+ } catch (ClassNotFoundException e) {
+ LOG.error("Dispatcher class not found: {}. Ensure the class is on the classpath.", className, e);
+
+ break;
+ } catch (Exception e) {
+ if (attempt < MAX_INIT_ATTEMPTS) {
+ LOG.warn("Dispatcher init attempt {}/{} failed, retrying in {}ms...", attempt, MAX_INIT_ATTEMPTS, retryDelay, e);
+
+ try {
+ Thread.sleep(retryDelay);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+
+ break;
+ }
+
+ retryDelay *= 2;
+ } else {
+ LOG.error("Error initializing dispatcher class after {} attempts: {}", MAX_INIT_ATTEMPTS, className, e);
+ }
+ }
+ }
+
+ LOG.info("<== OpenSearchDispatcherManager.initializeDispatcher()");
+ }
+
+ private void startDispatcher() {
+ LOG.info("==> OpenSearchDispatcherManager.startDispatcher()");
+
+ logStartupBanner();
+
+ if (dispatcher != null) {
+ try {
+ String name = dispatcher.getClass().getSimpleName();
+
+ dispatcherThread = new Thread(dispatcher, name);
+
+ dispatcherThread.setDaemon(true);
+ dispatcherThread.start();
+
+ LOG.info("Started {} thread [Thread-ID: {}, Thread-Name: '{}']", name, dispatcherThread.getId(), dispatcherThread.getName());
+ } catch (Exception e) {
+ LOG.error("Error starting dispatcher: {}", dispatcher.getClass().getSimpleName(), e);
+ }
+ }
+
+ LOG.info("<== OpenSearchDispatcherManager.startDispatcher()");
+ }
+
+ private void logStartupBanner() {
+ LOG.info("########## OPENSEARCH DISPATCHER SERVICE STARTUP ##########");
+
+ if (dispatcher == null) {
+ LOG.warn("WARNING: No OpenSearch dispatchers are enabled!");
+ LOG.warn("Verify: {}=true in configuration", OPENSEARCH_DEST_PROP);
+ } else {
+ AuditServerLogFormatter.LogBuilder builder = AuditServerLogFormatter.builder("OpenSearch Dispatcher Status");
+
+ String type = dispatcher.getClass().getSimpleName();
+
+ builder.add(type, "ENABLED");
+ builder.add("Topic", dispatcher.getTopicName());
+ builder.logInfo(LOG);
+
+ LOG.info("Starting OpenSearch dispatcher thread...");
+ }
+
+ LOG.info("##########################################################");
+ }
+}
diff --git a/audit-server/audit-dispatcher/dispatcher-opensearch/src/main/java/org/apache/ranger/audit/dispatcher/kafka/AuditOpenSearchDispatcher.java b/audit-server/audit-dispatcher/dispatcher-opensearch/src/main/java/org/apache/ranger/audit/dispatcher/kafka/AuditOpenSearchDispatcher.java
new file mode 100644
index 000000000..f644e316b
--- /dev/null
+++ b/audit-server/audit-dispatcher/dispatcher-opensearch/src/main/java/org/apache/ranger/audit/dispatcher/kafka/AuditOpenSearchDispatcher.java
@@ -0,0 +1,149 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.ranger.audit.dispatcher.kafka;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.ranger.audit.destination.OpenSearchAuditDestination;
+import org.apache.ranger.audit.provider.MiscUtil;
+import org.apache.ranger.audit.server.AuditServerConstants;
+import org.apache.ranger.audit.utils.AuditServerLogFormatter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Properties;
+import java.util.stream.Collectors;
+
+public class AuditOpenSearchDispatcher extends AuditDispatcherBase {
+ private static final Logger LOG = LoggerFactory.getLogger(AuditOpenSearchDispatcher.class);
+
+ private static final String DEFAULT_GROUP = "ranger_audit_opensearch_dispatcher_group";
+ private static final long RETRY_SLEEP_MS = 5000L;
+
+ private final OpenSearchAuditDestination openSearchAuditDestination;
+
+ public AuditOpenSearchDispatcher(final Properties props, final String propPrefix) throws Exception {
+ super(props, propPrefix, DEFAULT_GROUP);
+
+ this.openSearchAuditDestination = new OpenSearchAuditDestination();
+
+ init(props, propPrefix);
+ }
+
+ @Override
+ protected final String getDispatcherName() {
+ return "OPENSEARCH";
+ }
+
+ @Override
+ protected final DispatcherWorker createDispatcherWorker(final String workerId, final List<Integer> assignedPartitions) {
+ return new OpenSearchDispatcherWorker(workerId, assignedPartitions);
+ }
+
+ @Override
+ protected final void shutdownDestination() {
+ if (openSearchAuditDestination != null) {
+ try {
+ openSearchAuditDestination.stop();
+ } catch (Exception e) {
+ LOG.error("Error shutting down OpenSearch destination handler", e);
+ }
+ }
+ }
+
+ private void init(final Properties props, final String propPrefix) {
+ LOG.info("==> AuditOpenSearchDispatcher.init()");
+
+ openSearchAuditDestination.init(props, propPrefix);
+
+ this.dispatcherThreadCount = MiscUtil.getIntProperty(props, propPrefix + "." + AuditServerConstants.PROP_DISPATCHER_THREAD_COUNT, 1);
+ this.offsetCommitStrategy = MiscUtil.getStringProperty(props, propPrefix + "." + AuditServerConstants.PROP_DISPATCHER_OFFSET_COMMIT_STRATEGY, AuditServerConstants.DEFAULT_OFFSET_COMMIT_STRATEGY);
+ this.offsetCommitInterval = MiscUtil.getLongProperty(props, propPrefix + "." + AuditServerConstants.PROP_DISPATCHER_OFFSET_COMMIT_INTERVAL, AuditServerConstants.DEFAULT_OFFSET_COMMIT_INTERVAL_MS);
+
+ AuditServerLogFormatter.builder("AuditOpenSearchDispatcher Offset Management Configuration")
+ .add("Thread Count", dispatcherThreadCount)
+ .add("Commit Strategy", offsetCommitStrategy)
+ .add("Commit Interval (ms)", offsetCommitInterval + " (manual mode only)")
+ .logInfo(LOG);
+
+ LOG.info("<== AuditOpenSearchDispatcher.init()");
+ }
+
+ private void processMessageBatch(final Collection<String> audits) throws Exception {
+ if (audits != null && !audits.isEmpty()) {
+ boolean processed = openSearchAuditDestination.logJSON(audits);
+
+ if (!processed) {
+ throw new Exception("Failure in sending audits into OpenSearch");
+ }
+ }
+ }
+
+ private class OpenSearchDispatcherWorker extends DispatcherWorker {
+ OpenSearchDispatcherWorker(final String workerId, final List<Integer> assignedPartitions) {
+ super(workerId, assignedPartitions);
+ }
+
+ @Override
+ protected void processRecordBatch(final ConsumerRecords<String, String> records) {
+ for (TopicPartition tp : records.partitions()) {
+ List<ConsumerRecord<String, String>> tpRecords = records.records(tp);
+
+ if (tpRecords.isEmpty()) {
+ continue;
+ }
+
+ try {
+ List<String> auditBatch = tpRecords.stream().map(ConsumerRecord::value).collect(Collectors.toList());
+
+ processMessageBatch(auditBatch);
+
+ ConsumerRecord<String, String> last = tpRecords.get(tpRecords.size() - 1);
+
+ pendingOffsets.put(tp, new OffsetAndMetadata(last.offset() + 1));
+
+ messagesProcessedSinceLastCommit.addAndGet(tpRecords.size());
+ } catch (Exception e) {
+ LOG.error("Error processing batch in worker '{}', partition={}, batch size: {}", workerId, tp, tpRecords.size(), e);
+
+ ConsumerRecord<String, String> first = tpRecords.get(0);
+
+ pendingOffsets.put(tp, new OffsetAndMetadata(first.offset()));
+
+ try {
+ workerDispatcher.seek(tp, first.offset());
+ } catch (Exception seekEx) {
+ LOG.error("Failed to seek to offset {} for partition {} after OpenSearch batch error", first.offset(), tp, seekEx);
+ }
+
+ try {
+ Thread.sleep(RETRY_SLEEP_MS);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+ }
+ }
+}
diff --git a/audit-server/audit-dispatcher/dispatcher-opensearch/src/main/resources/conf/logback.xml b/audit-server/audit-dispatcher/dispatcher-opensearch/src/main/resources/conf/logback.xml
new file mode 100644
index 000000000..29b33e86a
--- /dev/null
+++ b/audit-server/audit-dispatcher/dispatcher-opensearch/src/main/resources/conf/logback.xml
@@ -0,0 +1,55 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<configuration>
+
+ <property name="LOG_DIR" value="${audit.dispatcher.log.dir:-logs}" />
+ <property name="LOG_FILE" value="${audit.dispatcher.log.file:-ranger-audit-dispatcher-opensearch.log}" />
+
+ <appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
+ <file>${LOG_DIR}/${LOG_FILE}</file>
+ <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
+ <!-- rollover daily -->
+ <fileNamePattern>${LOG_DIR}/${LOG_FILE}.%d{yyyy-MM-dd}.%i.gz</fileNamePattern>
+ <!-- each file should be at most 100MB, keep 30 days worth of history, but at most 2GB -->
+ <maxFileSize>100MB</maxFileSize>
+ <maxHistory>30</maxHistory>
+ <totalSizeCap>2GB</totalSizeCap>
+ </rollingPolicy>
+ <encoder>
+ <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+ </encoder>
+ </appender>
+
+ <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+ <encoder>
+ <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+ </encoder>
+ </appender>
+
+ <!-- Suppress verbose Kafka logs -->
+ <logger name="org.apache.kafka" level="WARN" />
+ <logger name="org.apache.zookeeper" level="WARN" />
+ <logger name="org.apache.ranger.audit.dispatcher" level="INFO" />
+ <logger name="org.apache.ranger.audit.destination" level="INFO" />
+
+ <root level="INFO">
+ <appender-ref ref="FILE" />
+ <appender-ref ref="STDOUT" />
+ </root>
+
+</configuration>
\ No newline at end of file
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
new file mode 100644
index 000000000..b5f01f3d9
--- /dev/null
+++ b/audit-server/audit-dispatcher/dispatcher-opensearch/src/main/resources/conf/ranger-audit-dispatcher-opensearch-site.xml
@@ -0,0 +1,228 @@
+<!--
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License. See accompanying LICENSE file.
+-->
+<configuration>
+ <property>
+ <name>log.dir</name>
+ <value>${audit.dispatcher.opensearch.log.dir}</value>
+ <description>Log directory for OpenSearch dispatcher service</description>
+ </property>
+
+ <!-- Used for KERBEROS authentication into OpenSearch (if enabled) -->
+ <property>
+ <name>ranger.audit.dispatcher.host</name>
+ <value>ranger-audit-dispatcher-opensearch.rangernw</value>
+ <description>
+ - In Docker: Use full service name with domain (e.g., ranger-audit-server.rangernw)
+ - In VM: Use the actual FQDN (e.g., ranger.example.com)
+ </description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.service.kerberos.principal</name>
+ <value>rangerauditserver/_HOST@EXAMPLE.COM</value>
+ <description>
+ rangerauditserver user kerberos principal for authentication into kafka
+ </description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.service.kerberos.keytab</name>
+ <value>/etc/keytabs/rangerauditserver.keytab</value>
+ <description>
+ keytab of the rangerauditserver principal
+ </description>
+ </property>
+
+ <!-- KAFKA DISPATCHER CONFIGURATION -->
+ <property>
+ <name>ranger.audit.dispatcher.kafka.bootstrap.servers</name>
+ <value>ranger-kafka.rangernw:9092</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kafka.topic.name</name>
+ <value>ranger_audits</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kafka.security.protocol</name>
+ <value>SASL_PLAINTEXT</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kafka.sasl.mechanism</name>
+ <value>GSSAPI</value>
+ </property>
+
+ <!-- OpenSearch Dispatcher Threading Configuration -->
+ <property>
+ <name>ranger.audit.dispatcher.thread.count</name>
+ <value>5</value>
+ <description>Number of OpenSearch dispatcher worker threads (higher for indexing throughput)</description>
+ </property>
+
+ <!-- Offset Management -->
+ <property>
+ <name>ranger.audit.dispatcher.offset.commit.strategy</name>
+ <value>batch</value>
+ <description>batch or manual</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.offset.commit.interval.ms</name>
+ <value>30000</value>
+ <description>Used only if strategy is 'manual'</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.max.poll.records</name>
+ <value>500</value>
+ <description>Maximum records per poll for batch processing</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.partition.assignment.strategy</name>
+ <value>org.apache.kafka.clients.consumer.CooperativeStickyAssignor</value>
+ <description>Cooperative sticky assignor for K8s pod churn (Kafka 3.9+). Override with RangeAssignor if needed.</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.session.timeout.ms</name>
+ <value>60000</value>
+ <description>Consumer session timeout (failure detection). Must be > 3 × heartbeat.interval.ms.</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.heartbeat.interval.ms</name>
+ <value>10000</value>
+ <description>Consumer heartbeat interval while idle.</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.max.poll.interval.ms</name>
+ <value>300000</value>
+ <description>Max time between poll() calls while processing a batch (OpenSearch indexing). Increase if batches are slow.</description>
+ </property>
+
+ <!-- OpenSearch Dispatcher Class -->
+ <property>
+ <name>ranger.audit.dispatcher.class</name>
+ <value>org.apache.ranger.audit.dispatcher.kafka.AuditOpenSearchDispatcher</value>
+ </property>
+
+ <!-- OPENSEARCH DESTINATION CONFIGURATION -->
+ <property>
+ <name>xasecure.audit.destination.opensearch</name>
+ <value>true</value>
+ <description>Enable the OpenSearch audit destination</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.urls</name>
+ <value>ranger-opensearch</value>
+ <description>OpenSearch host(s), comma-separated</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.port</name>
+ <value>9200</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.protocol</name>
+ <value>http</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.index</name>
+ <value>ranger_audits</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.authentication.type</name>
+ <value></value>
+ <description>
+ OpenSearch authentication scheme: kerberos, basic, or none.
+ When left empty it is inferred for backward compatibility: kerberos if the
+ password/keytab points to an existing keytab file, basic if user and password
+ are set, otherwise none. Use NONE or empty for unauthenticated OpenSearch (dev only).
+ Production: set basic (with user/password) or kerberos (with kerberos.principal/keytab).
+ </description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.user</name>
+ <value></value>
+ <description>Username for OpenSearch Basic Auth (authentication.type=basic).</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.password</name>
+ <value></value>
+ <description>Password for OpenSearch Basic Auth (authentication.type=basic).</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kerberos.principal</name>
+ <value></value>
+ <description>Kerberos principal when authentication.type=kerberos. Falls back to ranger.audit.dispatcher.user when unset.</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kerberos.keytab</name>
+ <value></value>
+ <description>Path to the Kerberos keytab when authentication.type=kerberos. Falls back to ranger.audit.dispatcher.password when unset.</description>
+ </property>
+
+ <!-- DISPATCHER TYPE SELECTION -->
+ <property>
+ <name>ranger.audit.dispatcher.type</name>
+ <value>opensearch</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.war.file</name>
+ <value>ranger-audit-dispatcher.war</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.launcher.class</name>
+ <value>org.apache.ranger.audit.dispatcher.AuditDispatcherLauncher</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.main.class</name>
+ <value>org.apache.ranger.audit.dispatcher.AuditDispatcherApplication</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.http.port</name>
+ <value>7090</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.contextName</name>
+ <value>/</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kafka.group.id</name>
+ <value>ranger_audit_opensearch_dispatcher_group</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.opensearch.class</name>
+ <value>org.apache.ranger.audit.dispatcher.OpenSearchDispatcherManager</value>
+ </property>
+</configuration>
\ No newline at end of file
diff --git a/audit-server/audit-dispatcher/dispatcher-opensearch/src/test/java/org/apache/ranger/audit/dispatcher/TestOpenSearchDispatcherManager.java b/audit-server/audit-dispatcher/dispatcher-opensearch/src/test/java/org/apache/ranger/audit/dispatcher/TestOpenSearchDispatcherManager.java
new file mode 100644
index 000000000..39a0ebc7a
--- /dev/null
+++ b/audit-server/audit-dispatcher/dispatcher-opensearch/src/test/java/org/apache/ranger/audit/dispatcher/TestOpenSearchDispatcherManager.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.ranger.audit.dispatcher;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class TestOpenSearchDispatcherManager {
+ @AfterEach
+ void clearSystemProperty() {
+ System.clearProperty("ranger.audit.dispatcher.type");
+ }
+
+ @Test
+ void init_skipsWhenDispatcherTypeIsNotOpenSearch() {
+ System.setProperty("ranger.audit.dispatcher.type", "solr");
+
+ OpenSearchDispatcherManager manager = new OpenSearchDispatcherManager();
+ Properties props = new Properties();
+
+ assertDoesNotThrow(() -> manager.init(props));
+ }
+
+ @Test
+ void init_throwsWhenPropsAreNull() {
+ OpenSearchDispatcherManager manager = new OpenSearchDispatcherManager();
+
+ assertThrows(RuntimeException.class, () -> manager.init(null));
+ }
+
+ @Test
+ void init_skipsWhenOpenSearchDestinationDisabled() {
+ OpenSearchDispatcherManager manager = new OpenSearchDispatcherManager();
+ Properties props = new Properties();
+ props.setProperty("xasecure.audit.destination.opensearch", "false");
+
+ assertDoesNotThrow(() -> manager.init(props));
+ }
+
+ @Test
+ void shutdown_handlesNullDispatcherGracefully() {
+ OpenSearchDispatcherManager manager = new OpenSearchDispatcherManager();
+
+ assertDoesNotThrow(manager::shutdown);
+ }
+
+ @Test
+ void init_throwsWhenEnabledDispatcherClassCannotBeCreated() {
+ OpenSearchDispatcherManager manager = new OpenSearchDispatcherManager();
+ Properties props = new Properties();
+ props.setProperty("xasecure.audit.destination.opensearch", "true");
+ props.setProperty("ranger.audit.dispatcher.class", "com.nonexistent.FakeDispatcher");
+ props.setProperty("ranger.audit.dispatcher.kafka.bootstrap.servers", "localhost:9092");
+ props.setProperty("ranger.audit.dispatcher.kafka.topic", "ranger_audits");
+
+ assertThrows(RuntimeException.class, () -> manager.init(props));
+ }
+}
diff --git a/audit-server/audit-dispatcher/dispatcher-opensearch/src/test/java/org/apache/ranger/audit/dispatcher/kafka/TestAuditOpenSearchDispatcher.java b/audit-server/audit-dispatcher/dispatcher-opensearch/src/test/java/org/apache/ranger/audit/dispatcher/kafka/TestAuditOpenSearchDispatcher.java
new file mode 100644
index 000000000..1ac686fab
--- /dev/null
+++ b/audit-server/audit-dispatcher/dispatcher-opensearch/src/test/java/org/apache/ranger/audit/dispatcher/kafka/TestAuditOpenSearchDispatcher.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.ranger.audit.dispatcher.kafka;
+
+import org.apache.ranger.audit.destination.OpenSearchAuditDestination;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.anyCollection;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+public class TestAuditOpenSearchDispatcher {
+ @Mock
+ private OpenSearchAuditDestination openSearchAuditDestination;
+
+ private AuditOpenSearchDispatcher dispatcher;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ Field unsafeField = sun.misc.Unsafe.class.getDeclaredField("theUnsafe");
+ unsafeField.setAccessible(true);
+ sun.misc.Unsafe unsafe = (sun.misc.Unsafe) unsafeField.get(null);
+ dispatcher = (AuditOpenSearchDispatcher) unsafe.allocateInstance(AuditOpenSearchDispatcher.class);
+
+ setField(dispatcher, "openSearchAuditDestination", openSearchAuditDestination);
+ }
+
+ @Test
+ void processMessageBatch_delegatesToDestinationLogJSON() throws Exception {
+ List<String> audits = Collections.singletonList("{\"eventId\":\"id-1\"}");
+
+ when(openSearchAuditDestination.logJSON(anyCollection())).thenReturn(true);
+
+ invokeProcessMessageBatch(audits);
+
+ verify(openSearchAuditDestination).logJSON(audits);
+ }
+
+ @Test
+ void processMessageBatch_throwsWhenDestinationReportsFailure() {
+ List<String> audits = Collections.singletonList("{\"eventId\":\"id-1\"}");
+
+ when(openSearchAuditDestination.logJSON(anyCollection())).thenReturn(false);
+
+ assertThrows(Exception.class, () -> invokeProcessMessageBatch(audits));
+ }
+
+ @Test
+ void processMessageBatch_throwsOnNullInputWithoutDelegating() {
+ verifyNoInteractions(openSearchAuditDestination);
+ }
+
+ @Test
+ void processMessageBatch_throwsOnEmptyInputWithoutDelegating() {
+ verify(openSearchAuditDestination, never()).logJSON(anyCollection());
+ }
+
+ private void invokeProcessMessageBatch(Collection<String> audits) throws Exception {
+ Method method = AuditOpenSearchDispatcher.class.getDeclaredMethod("processMessageBatch", Collection.class);
+ method.setAccessible(true);
+ try {
+ method.invoke(dispatcher, audits);
+ } catch (java.lang.reflect.InvocationTargetException e) {
+ if (e.getCause() instanceof Exception) {
+ throw (Exception) e.getCause();
+ }
+ throw e;
+ }
+ }
+
+ private void setField(Object target, String fieldName, Object value) throws Exception {
+ Class<?> clazz = target.getClass();
+ while (clazz != null) {
+ try {
+ Field field = clazz.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ field.set(target, value);
+ return;
+ } catch (NoSuchFieldException e) {
+ clazz = clazz.getSuperclass();
+ }
+ }
+ throw new NoSuchFieldException(fieldName + " not found in class hierarchy");
+ }
+}
diff --git a/audit-server/audit-dispatcher/pom.xml b/audit-server/audit-dispatcher/pom.xml
index a306cc167..83a2f3d3b 100644
--- a/audit-server/audit-dispatcher/pom.xml
+++ b/audit-server/audit-dispatcher/pom.xml
@@ -34,6 +34,7 @@
<module>dispatcher-app</module>
<module>dispatcher-common</module>
<module>dispatcher-hdfs</module>
+ <module>dispatcher-opensearch</module>
<module>dispatcher-solr</module>
</modules>
diff --git a/audit-server/audit-dispatcher/scripts/start-audit-dispatcher.sh b/audit-server/audit-dispatcher/scripts/start-audit-dispatcher.sh
index ad72e71b7..ae7100d40 100755
--- a/audit-server/audit-dispatcher/scripts/start-audit-dispatcher.sh
+++ b/audit-server/audit-dispatcher/scripts/start-audit-dispatcher.sh
@@ -36,6 +36,10 @@ if [ -z "${DISPATCHER_TYPE}" ]; then
fi
CONF_FILE="${AUDIT_DISPATCHER_CONF_DIR}/ranger-audit-dispatcher-${DISPATCHER_TYPE}-site.xml"
+LOGBACK_CONFIG_FILE="${AUDIT_DISPATCHER_CONF_DIR}/logback.xml"
+if [ -n "${DISPATCHER_TYPE}" ] && [ -f "${AUDIT_DISPATCHER_CONF_DIR}/logback-${DISPATCHER_TYPE}.xml" ]; then
+ LOGBACK_CONFIG_FILE="${AUDIT_DISPATCHER_CONF_DIR}/logback-${DISPATCHER_TYPE}.xml"
+fi
# Set default heap size if not set
if [ -z "${AUDIT_DISPATCHER_HEAP}" ]; then
@@ -44,7 +48,7 @@ fi
# Set default Java options if not set
if [ -z "${AUDIT_DISPATCHER_OPTS}" ]; then
- AUDIT_DISPATCHER_OPTS="-Dlogback.configurationFile=${AUDIT_DISPATCHER_CONF_DIR}/logback.xml \
+ AUDIT_DISPATCHER_OPTS="-Dlogback.configurationFile=${LOGBACK_CONFIG_FILE} \
-Daudit.dispatcher.log.dir=${AUDIT_DISPATCHER_LOG_DIR} \
-Daudit.dispatcher.log.file=ranger-audit-dispatcher.log \
-Djava.net.preferIPv4Stack=true \
diff --git a/dev-support/ranger-docker/Dockerfile.ranger-kdc b/dev-support/ranger-docker/Dockerfile.ranger-kdc
index f3ce89d18..a81119bff 100644
--- a/dev-support/ranger-docker/Dockerfile.ranger-kdc
+++ b/dev-support/ranger-docker/Dockerfile.ranger-kdc
@@ -36,10 +36,13 @@ COPY ./scripts/kdc/kdc.conf /etc/krb5kdc/kdc.conf
COPY ./scripts/kdc/kadm5.acl /etc/krb5kdc/kadm5.acl
COPY ./scripts/kdc/entrypoint.sh /entrypoint.sh
-RUN chmod +x /entrypoint.sh
+RUN chmod +x /entrypoint.sh && apt-get update && apt-get install -y netcat-openbsd && rm -rf /var/lib/apt/lists/*
VOLUME /etc/keytabs
EXPOSE 88/tcp 88/udp 749/tcp
+HEALTHCHECK --interval=5s --timeout=3s --start-period=120s --retries=30 \
+ CMD test -f /etc/keytabs/.provisioned && echo | nc -w 1 localhost 88 || exit 1
+
ENTRYPOINT ["/entrypoint.sh"]
diff --git a/dev-support/ranger-docker/README.md b/dev-support/ranger-docker/README.md
index 3f8090d6e..1ddd9c3fd 100644
--- a/dev-support/ranger-docker/README.md
+++ b/dev-support/ranger-docker/README.md
@@ -106,6 +106,54 @@ docker compose -f docker-compose.ranger.yml -f docker-compose.ranger-trino.yml u
~~~
docker compose -f docker-compose.ranger.yml -f docker-compose.ranger-opensearch.yml up -d
~~~
+
+#### OpenSearch audit flow (replace Solr for access audits)
+
+OpenSearch can replace Solr for **audit storage and UI queries**. Ranger Admin reads audits via
+`audit_store=opensearch` using a native low-level REST client (compatible with any OpenSearch version).
+
+**Write path:** access audits flow through audit-server ingestor, Kafka, and the Java
+`ranger-audit-dispatcher-opensearch` service into the OpenSearch `ranger_audits` index.
+Ranger Admin policy/admin transaction audits remain DB-backed; this is the same boundary
+used by the Solr audit path.
+
+##### Setup
+
+~~~
+# Prerequisites: build the audit-dispatcher tarball and download archives
+mvn clean package -DskipTests -pl distro -am
+cp target/ranger-*-audit-dispatcher.tar.gz dev-support/ranger-docker/dist/
+cd dev-support/ranger-docker
+./download-archives.sh kafka opensearch hadoop
+
+export RANGER_DB_TYPE=postgres
+
+# 1. Start OpenSearch first (Ranger Admin's bootstrapper needs it on startup)
+docker compose -f docker-compose.ranger.yml -f docker-compose.ranger-opensearch.yml \
+ -f docker-compose.ranger-kafka.yml -f docker-compose.ranger-hadoop.yml \
+ -f docker-compose.ranger-audit-server.yml \
+ -f docker-compose.ranger-audit-dispatcher-opensearch.yml up -d ranger-opensearch
+
+# 2. Start core stack (Ranger Admin, Kafka, Hadoop)
+# Kafka auto-creates the ranger_audits topic on startup.
+# Ranger Admin auto-creates the OpenSearch index via OpenSearchIndexBootStrapper.
+docker compose -f docker-compose.ranger.yml -f docker-compose.ranger-opensearch.yml \
+ -f docker-compose.ranger-kafka.yml -f docker-compose.ranger-hadoop.yml \
+ -f docker-compose.ranger-audit-server.yml \
+ -f docker-compose.ranger-audit-dispatcher-opensearch.yml up -d ranger ranger-kafka ranger-hadoop
+
+# 3. Start audit ingestor and OpenSearch dispatcher
+docker compose -f docker-compose.ranger.yml -f docker-compose.ranger-opensearch.yml \
+ -f docker-compose.ranger-kafka.yml -f docker-compose.ranger-hadoop.yml \
+ -f docker-compose.ranger-audit-server.yml \
+ -f docker-compose.ranger-audit-dispatcher-opensearch.yml up -d ranger-audit-ingestor ranger-audit-dispatcher-opensearch
+~~~
+
+For **fresh Ranger installs** using OpenSearch for audits, set `audit_store=opensearch` in
+`scripts/admin/ranger-admin-install-postgres.properties` and configure the `audit_opensearch_*` properties.
+
+For **existing Solr-based installs**, switching stores requires updating `audit_store=opensearch`
+in install.properties, configuring the `audit_opensearch_*` properties, and restarting Ranger Admin.
Similarly, check the `depends` section of the `docker-compose.ranger-service.yaml` file and add docker-compose files for these services when trying to bring up the `service` container.
#### Bring up all containers
diff --git a/dev-support/ranger-docker/docker-compose.ranger-audit-dispatcher-opensearch.yml b/dev-support/ranger-docker/docker-compose.ranger-audit-dispatcher-opensearch.yml
new file mode 100644
index 000000000..b99a03c6d
--- /dev/null
+++ b/dev-support/ranger-docker/docker-compose.ranger-audit-dispatcher-opensearch.yml
@@ -0,0 +1,70 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file to You
+# under the Apache License, Version 2.0 (the "License"); you may not
+# use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+services:
+ ranger-audit-dispatcher-opensearch:
+ build:
+ context: .
+ dockerfile: Dockerfile.ranger-audit-dispatcher
+ args:
+ - RANGER_BASE_IMAGE=${RANGER_BASE_IMAGE}
+ - RANGER_BASE_VERSION=${RANGER_BASE_VERSION}
+ - RANGER_VERSION=${RANGER_VERSION}
+ - KERBEROS_ENABLED=${KERBEROS_ENABLED}
+ image: ranger-audit-dispatcher-opensearch:latest
+ container_name: ranger-audit-dispatcher-opensearch
+ hostname: ranger-audit-dispatcher-opensearch.rangernw
+ command: ["opensearch"]
+ stdin_open: true
+ tty: true
+ depends_on:
+ ranger-kdc:
+ condition: service_healthy
+ ranger-kafka:
+ condition: service_started
+ ranger-opensearch:
+ condition: service_started
+ ports:
+ - "7093:7090"
+ environment:
+ JAVA_HOME: /opt/java/openjdk
+ AUDIT_DISPATCHER_HOME_DIR: /opt/ranger/audit-dispatcher
+ AUDIT_DISPATCHER_CONF_DIR: /opt/ranger/audit-dispatcher/conf
+ AUDIT_DISPATCHER_LOG_DIR: /var/log/ranger/audit-dispatcher/opensearch
+ AUDIT_DISPATCHER_HEAP: "-Xms512m -Xmx2g"
+ RANGER_VERSION: ${RANGER_VERSION}
+ KERBEROS_ENABLED: ${KERBEROS_ENABLED}
+ KAFKA_BOOTSTRAP_SERVERS: ranger-kafka.rangernw:9092
+ OPENSEARCH_URL: http://ranger-opensearch.rangernw:9200
+ networks:
+ - ranger
+ volumes:
+ - ./dist/keytabs/ranger-audit-dispatcher-opensearch:/etc/keytabs
+ - 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
+ healthcheck:
+ test: ["CMD", "curl", "-f", "http://localhost:7090/api/health/ping"]
+ interval: 30s
+ timeout: 10s
+ retries: 3
+ start_period: 60s
+
+volumes:
+ ranger-audit-dispatcher-opensearch-logs:
+
+networks:
+ ranger:
+ name: rangernw
diff --git a/dev-support/ranger-docker/docker-compose.ranger.yml b/dev-support/ranger-docker/docker-compose.ranger.yml
index 3e263fecc..a75de1424 100644
--- a/dev-support/ranger-docker/docker-compose.ranger.yml
+++ b/dev-support/ranger-docker/docker-compose.ranger.yml
@@ -81,6 +81,9 @@ services:
image: ranger-zk
container_name: ranger-zk
hostname: ranger-zk.rangernw
+ depends_on:
+ ranger-kdc:
+ condition: service_healthy
volumes:
- ./dist/keytabs/ranger-zk:/etc/keytabs
- ./scripts/kdc/krb5.conf:/etc/krb5.conf:ro
diff --git a/dev-support/ranger-docker/download-archives.sh b/dev-support/ranger-docker/download-archives.sh
index 1b35be61b..b107cc50d 100755
--- a/dev-support/ranger-docker/download-archives.sh
+++ b/dev-support/ranger-docker/download-archives.sh
@@ -22,9 +22,15 @@
#
#
-# source .env file to get versions to download
+# Load .env file to get versions to download. Do not source it directly:
+# values like JAVA_OPTS contain spaces and are valid for docker compose .env
+# files, but not valid shell assignments unless quoted.
#
-source .env
+while IFS='=' read -r key value; do
+ if [[ "${key}" =~ ^[A-Za-z_][A-Za-z0-9_]*$ ]]; then
+ export "${key}=${value}"
+ fi
+done < .env
downloadIfNotPresent() {
diff --git a/dev-support/ranger-docker/scripts/admin/ranger-admin-install-postgres.properties b/dev-support/ranger-docker/scripts/admin/ranger-admin-install-postgres.properties
index e8d28825f..fc8e8682a 100644
--- a/dev-support/ranger-docker/scripts/admin/ranger-admin-install-postgres.properties
+++ b/dev-support/ranger-docker/scripts/admin/ranger-admin-install-postgres.properties
@@ -43,19 +43,33 @@ rangerUsersync_password=rangerR0cks!
keyadmin_password=rangerR0cks!
-audit_store=solr
+## --- Audit store: enable ONE of the following blocks ---
+
+## Option 1: Solr (default)
+# audit_store=solr
# FQDN required for SPNEGO to Solr (HTTP/ranger-solr.rangernw@REALM)
-audit_solr_urls=http://ranger-solr.rangernw:8983/solr/ranger_audits
-audit_solr_collection_name=ranger_audits
+# audit_solr_urls=http://ranger-solr.rangernw:8983/solr/ranger_audits
+# audit_solr_collection_name=ranger_audits
+## Option 2: OpenSearch / Elasticsearch (legacy - uses ES HighLevel REST Client)
# audit_store=elasticsearch
-audit_elasticsearch_urls=
-audit_elasticsearch_port=9200
-audit_elasticsearch_protocol=http
-audit_elasticsearch_user=elastic
-audit_elasticsearch_password=elasticsearch
-audit_elasticsearch_index=ranger_audits
-audit_elasticsearch_bootstrap_enabled=true
+# audit_elasticsearch_urls=ranger-opensearch
+# audit_elasticsearch_port=9200
+# audit_elasticsearch_protocol=http
+# audit_elasticsearch_user=NONE
+# audit_elasticsearch_password=NONE
+# audit_elasticsearch_index=ranger_audits
+# audit_elasticsearch_bootstrap_enabled=true
+
+## Option 3: OpenSearch (native - uses low-level REST client, works with any OpenSearch version)
+audit_store=opensearch
+audit_opensearch_urls=ranger-opensearch
+audit_opensearch_port=9200
+audit_opensearch_protocol=http
+audit_opensearch_user=admin
+audit_opensearch_password=admin
+audit_opensearch_index=ranger_audits
+audit_opensearch_bootstrap_enabled=true
policymgr_external_url=http://ranger-admin:6080
policymgr_http_enabled=true
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
new file mode 100644
index 000000000..2ef249569
--- /dev/null
+++ b/dev-support/ranger-docker/scripts/audit-dispatcher/ranger-audit-dispatcher-opensearch-site.xml
@@ -0,0 +1,207 @@
+<!--
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License. See accompanying LICENSE file.
+-->
+<configuration>
+ <property>
+ <name>ranger.audit.dispatcher.opensearch.class</name>
+ <value>org.apache.ranger.audit.dispatcher.OpenSearchDispatcherManager</value>
+ </property>
+
+ <property>
+ <name>log.dir</name>
+ <value>${audit.dispatcher.opensearch.log.dir}</value>
+ <description>Log directory for OpenSearch dispatcher service</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.host</name>
+ <value>ranger-audit-dispatcher-opensearch.rangernw</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.service.kerberos.principal</name>
+ <value>rangerauditserver/_HOST@EXAMPLE.COM</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.service.kerberos.keytab</name>
+ <value>/etc/keytabs/rangerauditserver.keytab</value>
+ </property>
+
+ <!-- KAFKA DISPATCHER CONFIGURATION -->
+ <property>
+ <name>ranger.audit.dispatcher.kafka.bootstrap.servers</name>
+ <value>ranger-kafka.rangernw:9092</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kafka.topic.name</name>
+ <value>ranger_audits</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kafka.security.protocol</name>
+ <value>SASL_PLAINTEXT</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kafka.sasl.mechanism</name>
+ <value>GSSAPI</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.thread.count</name>
+ <value>1</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.offset.commit.strategy</name>
+ <value>batch</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.offset.commit.interval.ms</name>
+ <value>30000</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.max.poll.records</name>
+ <value>500</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.partition.assignment.strategy</name>
+ <value>org.apache.kafka.clients.consumer.CooperativeStickyAssignor</value>
+ <description>Cooperative sticky assignor for K8s pod churn (Kafka 3.9+). Override with RangeAssignor if needed.</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.session.timeout.ms</name>
+ <value>60000</value>
+ <description>Consumer session timeout (failure detection). Must be > 3 × heartbeat.interval.ms.</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.heartbeat.interval.ms</name>
+ <value>10000</value>
+ <description>Consumer heartbeat interval while idle.</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.max.poll.interval.ms</name>
+ <value>300000</value>
+ <description>Max time between poll() calls while processing a batch (OpenSearch indexing). Increase if batches are slow.</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.class</name>
+ <value>org.apache.ranger.audit.dispatcher.kafka.AuditOpenSearchDispatcher</value>
+ </property>
+
+ <!-- OPENSEARCH DESTINATION CONFIGURATION -->
+ <property>
+ <name>xasecure.audit.destination.opensearch</name>
+ <value>true</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.urls</name>
+ <value>ranger-opensearch.rangernw</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.port</name>
+ <value>9200</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.protocol</name>
+ <value>http</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.index</name>
+ <value>ranger_audits</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.authentication.type</name>
+ <value></value>
+ <description>
+ OpenSearch authentication scheme: kerberos, basic, or none.
+ When left empty it is inferred for backward compatibility. Use NONE or empty
+ for unauthenticated OpenSearch (dev only). Production: set basic (with user/password)
+ or kerberos (with kerberos.principal/keytab).
+ </description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.user</name>
+ <value></value>
+ <description>Username for OpenSearch Basic Auth (authentication.type=basic).</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.password</name>
+ <value></value>
+ <description>Password for OpenSearch Basic Auth (authentication.type=basic).</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kerberos.principal</name>
+ <value></value>
+ <description>Kerberos principal when authentication.type=kerberos. Falls back to ranger.audit.dispatcher.user when unset.</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kerberos.keytab</name>
+ <value></value>
+ <description>Path to the Kerberos keytab when authentication.type=kerberos. Falls back to ranger.audit.dispatcher.password when unset.</description>
+ </property>
+
+ <!-- DISPATCHER TYPE SELECTION -->
+ <property>
+ <name>ranger.audit.dispatcher.type</name>
+ <value>opensearch</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.war.file</name>
+ <value>ranger-audit-dispatcher.war</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.launcher.class</name>
+ <value>org.apache.ranger.audit.dispatcher.AuditDispatcherLauncher</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.main.class</name>
+ <value>org.apache.ranger.audit.dispatcher.AuditDispatcherApplication</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.http.port</name>
+ <value>7090</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.contextName</name>
+ <value>/</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.kafka.group.id</name>
+ <value>ranger_audit_opensearch_dispatcher_group</value>
+ </property>
+</configuration>
diff --git a/dev-support/ranger-docker/scripts/kdc/entrypoint.sh b/dev-support/ranger-docker/scripts/kdc/entrypoint.sh
index ce2a7871e..db788684f 100644
--- a/dev-support/ranger-docker/scripts/kdc/entrypoint.sh
+++ b/dev-support/ranger-docker/scripts/kdc/entrypoint.sh
@@ -84,6 +84,9 @@ function create_keytabs() {
create_principal_and_keytab HTTP ranger-audit-dispatcher-hdfs
create_principal_and_keytab rangerauditserver ranger-audit-dispatcher-hdfs
+ create_principal_and_keytab HTTP ranger-audit-dispatcher-opensearch
+ create_principal_and_keytab rangerauditserver ranger-audit-dispatcher-opensearch
+
create_principal_and_keytab rangertagsync ranger-tagsync
create_principal_and_keytab rangerusersync ranger-usersync
@@ -146,7 +149,10 @@ if [ ! -f $DB_DIR/principal ]; then
echo "Database initialized"
create_keytabs
- create_testusers ranger ranger-usersync ranger-tagsync ranger-pdp ranger-audit-ingestor ranger-audit-dispatcher-solr ranger-audit-dispatcher-hdfs ranger-hadoop ranger-hive ranger-hbase ranger-kafka ranger-solr ranger-knox ranger-kms ranger-ozone ranger-trino ranger-opensearch
+ create_testusers ranger ranger-usersync ranger-tagsync ranger-pdp ranger-audit-ingestor ranger-audit-dispatcher-solr ranger-audit-dispatcher-hdfs ranger-audit-dispatcher-opensearch ranger-hadoop ranger-hive ranger-hbase ranger-kafka ranger-solr ranger-knox ranger-kms ranger-ozone ranger-trino ranger-opensearch
+
+ touch /etc/keytabs/.provisioned
+ echo "All keytabs provisioned"
else
echo "KDC DB already exists; skipping create"
fi
diff --git a/dev-support/ranger-docker/scripts/opensearch/opensearch.yml b/dev-support/ranger-docker/scripts/opensearch/opensearch.yml
index d63405043..42b0e8898 100644
--- a/dev-support/ranger-docker/scripts/opensearch/opensearch.yml
+++ b/dev-support/ranger-docker/scripts/opensearch/opensearch.yml
@@ -48,3 +48,6 @@ http.cors.allow-headers: "X-Requested-With, Content-Type, Content-Length, Author
# See opensearch-jaas.conf for Kerberos principal and keytab settings
# JVM is configured with: -Djava.security.auth.login.config and -Djava.security.krb5.conf
+# Wire-compatible with ES 7 REST client used by Ranger Admin
+compatibility.override_main_response_version: true
+
diff --git a/distro/pom.xml b/distro/pom.xml
index 3a1b38004..fa97c3fbd 100644
--- a/distro/pom.xml
+++ b/distro/pom.xml
@@ -35,6 +35,12 @@
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.ranger</groupId>
+ <artifactId>audit-dispatcher-opensearch</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>org.apache.ranger</groupId>
<artifactId>audit-dispatcher-solr</artifactId>
@@ -113,6 +119,12 @@
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.ranger</groupId>
+ <artifactId>ranger-audit-dest-os</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>org.apache.ranger</groupId>
<artifactId>ranger-audit-dest-solr</artifactId>
diff --git a/distro/src/main/assembly/audit-dispatcher.xml b/distro/src/main/assembly/audit-dispatcher.xml
index 34af9d79c..c8fe1bbbd 100644
--- a/distro/src/main/assembly/audit-dispatcher.xml
+++ b/distro/src/main/assembly/audit-dispatcher.xml
@@ -56,6 +56,15 @@
<fileMode>644</fileMode>
</fileSet>
+ <fileSet>
+ <outputDirectory>conf</outputDirectory>
+ <directory>${project.parent.basedir}/audit-server/audit-dispatcher/dispatcher-opensearch/src/main/resources/conf</directory>
+ <includes>
+ <include>ranger-audit-dispatcher-opensearch-site.xml</include>
+ </includes>
+ <fileMode>644</fileMode>
+ </fileSet>
+
<!-- Version file -->
<fileSet>
<outputDirectory></outputDirectory>
@@ -98,6 +107,22 @@
</includes>
</fileSet>
+ <!-- OpenSearch Dispatcher dependencies -->
+ <fileSet>
+ <outputDirectory>lib/dispatchers/opensearch</outputDirectory>
+ <directory>${project.parent.basedir}/audit-server/audit-dispatcher/dispatcher-opensearch/target/lib</directory>
+ <includes>
+ <include>*.jar</include>
+ </includes>
+ </fileSet>
+ <fileSet>
+ <outputDirectory>lib/dispatchers/opensearch</outputDirectory>
+ <directory>${project.parent.basedir}/audit-server/audit-dispatcher/dispatcher-opensearch/target</directory>
+ <includes>
+ <include>audit-dispatcher-opensearch-${project.version}.jar</include>
+ </includes>
+ </fileSet>
+
<!-- Solr Dispatcher dependencies -->
<fileSet>
<outputDirectory>lib/dispatchers/solr</outputDirectory>
@@ -116,6 +141,13 @@
</fileSets>
<files>
+ <file>
+ <outputDirectory>conf</outputDirectory>
+ <source>${project.parent.basedir}/audit-server/audit-dispatcher/dispatcher-opensearch/src/main/resources/conf/logback.xml</source>
+ <destName>logback-opensearch.xml</destName>
+ <fileMode>644</fileMode>
+ </file>
+
<!-- Unified WAR file -->
<file>
<outputDirectory>webapp</outputDirectory>
diff --git a/embeddedwebserver/src/main/java/org/apache/ranger/server/tomcat/EmbeddedServer.java b/embeddedwebserver/src/main/java/org/apache/ranger/server/tomcat/EmbeddedServer.java
index 93394cf6f..296da342a 100644
--- a/embeddedwebserver/src/main/java/org/apache/ranger/server/tomcat/EmbeddedServer.java
+++ b/embeddedwebserver/src/main/java/org/apache/ranger/server/tomcat/EmbeddedServer.java
@@ -72,8 +72,10 @@ public class EmbeddedServer {
private static final String AUDIT_SOURCE_TYPE = "ranger.audit.source.type";
private static final String AUDIT_SOURCE_SOLR = "solr";
private static final String AUDIT_SOURCE_ES = "elasticsearch";
+ private static final String AUDIT_SOURCE_OPENSEARCH = "opensearch";
private static final String SOLR_BOOTSTRAP_ENABLED = "ranger.audit.solr.bootstrap.enabled";
private static final String ES_BOOTSTRAP_ENABLED = "ranger.audit.elasticsearch.bootstrap.enabled";
+ private static final String OS_BOOTSTRAP_ENABLED = "ranger.audit.opensearch.bootstrap.enabled";
private static final String ADMIN_USER_KEYTAB = "ranger.admin.kerberos.keytab";
private static final String ADMIN_NAME_RULES = "hadoop.security.auth_to_local";
private static final String ADMIN_SERVER_NAME = "rangeradmin";
@@ -451,6 +453,18 @@ private void startServer(final Tomcat server) {
LOG.severe("Error while setting elasticsearch " + e);
}
}
+ } else if (AUDIT_SOURCE_OPENSEARCH.equalsIgnoreCase(auditSourceType)) {
+ boolean osBootstrapEnabled = Boolean.parseBoolean(EmbeddedServerUtil.getConfig(OS_BOOTSTRAP_ENABLED, "true"));
+
+ if (osBootstrapEnabled) {
+ try {
+ OpenSearchIndexBootStrapper osSchemaSetup = new OpenSearchIndexBootStrapper();
+
+ osSchemaSetup.start();
+ } catch (Exception e) {
+ LOG.severe("Error while setting opensearch " + e);
+ }
+ }
}
}
diff --git a/embeddedwebserver/src/main/java/org/apache/ranger/server/tomcat/OpenSearchIndexBootStrapper.java b/embeddedwebserver/src/main/java/org/apache/ranger/server/tomcat/OpenSearchIndexBootStrapper.java
new file mode 100644
index 000000000..9e8cd1001
--- /dev/null
+++ b/embeddedwebserver/src/main/java/org/apache/ranger/server/tomcat/OpenSearchIndexBootStrapper.java
@@ -0,0 +1,219 @@
+/*
+ * 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.server.tomcat;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.entity.ContentType;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.nio.entity.NStringEntity;
+import org.apache.http.util.EntityUtils;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestClientBuilder;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.logging.Logger;
+
+public class OpenSearchIndexBootStrapper extends Thread {
+ private static final Logger LOG = Logger.getLogger(OpenSearchIndexBootStrapper.class.getName());
+
+ private static final String CONFIG_PREFIX = "ranger.audit.opensearch";
+ private static final String CONFIG_URLS = CONFIG_PREFIX + ".urls";
+ private static final String CONFIG_PORT = CONFIG_PREFIX + ".port";
+ private static final String CONFIG_PROTOCOL = CONFIG_PREFIX + ".protocol";
+ private static final String CONFIG_USER = CONFIG_PREFIX + ".user";
+ private static final String CONFIG_PASSWORD = CONFIG_PREFIX + ".password";
+ private static final String CONFIG_INDEX = CONFIG_PREFIX + ".index";
+ private static final String CONFIG_INTERVAL = CONFIG_PREFIX + ".time.interval";
+ private static final String CONFIG_SHARDS = CONFIG_PREFIX + ".no.shards";
+ private static final String CONFIG_REPLICAS = CONFIG_PREFIX + ".no.replica";
+ private static final String CONFIG_MAX_RETRY = CONFIG_PREFIX + ".max.retry";
+ private static final String SCHEMA_FILE_NAME = "ranger_opensearch_schema.json";
+
+ private volatile RestClient client;
+ private String index;
+ private String urls;
+ private String protocol;
+ private String user;
+ private String password;
+ private int port;
+ private int noOfShards;
+ private int noOfReplicas;
+ private int maxRetry;
+ private long timeInterval;
+
+ @Override
+ public void run() {
+ LOG.info("OpenSearchIndexBootStrapper: starting...");
+
+ readConfig();
+
+ if (StringUtils.isBlank(urls) || "NONE".equalsIgnoreCase(urls)) {
+ LOG.severe("OpenSearch URL is not configured. Aborting bootstrap.");
+
+ return;
+ }
+
+ int retryCounter = 0;
+
+ while (maxRetry == -1 || retryCounter < maxRetry) {
+ retryCounter++;
+
+ try {
+ LOG.info("Trying to acquire OpenSearch connection");
+
+ connect();
+
+ if (client == null) {
+ logErrorAndWait(retryCounter, "Failed to create OpenSearch client");
+
+ continue;
+ }
+
+ LOG.info("Connection to OpenSearch established successfully");
+
+ if (indexExists()) {
+ LOG.info("Index '" + index + "' already exists. Bootstrap complete.");
+
+ return;
+ }
+
+ createIndex();
+
+ LOG.info("Index '" + index + "' created successfully.");
+
+ return;
+ } catch (Exception e) {
+ logErrorAndWait(retryCounter, e.getMessage());
+ }
+ }
+
+ LOG.severe("OpenSearch index bootstrap failed after " + retryCounter + " attempts.");
+ }
+
+ private void readConfig() {
+ urls = EmbeddedServerUtil.getConfig(CONFIG_URLS, "");
+ protocol = EmbeddedServerUtil.getConfig(CONFIG_PROTOCOL, "http");
+ user = EmbeddedServerUtil.getConfig(CONFIG_USER, "");
+ password = EmbeddedServerUtil.getConfig(CONFIG_PASSWORD, "");
+ port = Integer.parseInt(EmbeddedServerUtil.getConfig(CONFIG_PORT, "9200"));
+ index = EmbeddedServerUtil.getConfig(CONFIG_INDEX, "ranger_audits");
+ noOfShards = Integer.parseInt(EmbeddedServerUtil.getConfig(CONFIG_SHARDS, "1"));
+ noOfReplicas = Integer.parseInt(EmbeddedServerUtil.getConfig(CONFIG_REPLICAS, "1"));
+ maxRetry = Integer.parseInt(EmbeddedServerUtil.getConfig(CONFIG_MAX_RETRY, "30"));
+ timeInterval = Long.parseLong(EmbeddedServerUtil.getConfig(CONFIG_INTERVAL, "60000"));
+ }
+
+ private void connect() {
+ if (client != null) {
+ return;
+ }
+
+ HttpHost[] hosts = Arrays.stream(urls.split(",")).map(String::trim).filter(h -> !h.isEmpty()).map(h -> new HttpHost(h, port, protocol)).toArray(HttpHost[]::new);
+
+ RestClientBuilder builder = RestClient.builder(hosts);
+
+ if (StringUtils.isNotBlank(user) && StringUtils.isNotBlank(password) && !"NONE".equalsIgnoreCase(user) && !"NONE".equalsIgnoreCase(password)) {
+ CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
+ credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(user, password));
+ builder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider));
+ }
+
+ client = builder.build();
+ }
+
+ private boolean indexExists() throws IOException {
+ Request request = new Request("HEAD", "/" + index);
+ request.addParameter("ignore", "404");
+
+ Response response = client.performRequest(request);
+
+ return response.getStatusLine().getStatusCode() == 200;
+ }
+
+ private void createIndex() throws IOException {
+ String schemaJson = loadSchemaFile();
+ String body = buildCreateIndexBody(schemaJson);
+
+ Request request = new Request("PUT", "/" + index);
+
+ request.setEntity(new NStringEntity(body, ContentType.APPLICATION_JSON));
+
+ Response response = client.performRequest(request);
+ String result = EntityUtils.toString(response.getEntity());
+
+ if (response.getStatusLine().getStatusCode() >= 400) {
+ throw new IOException("Failed to create index '" + index + "': " + result);
+ }
+
+ LOG.info("Create index response: " + result);
+ }
+
+ private String buildCreateIndexBody(String mappingsJson) {
+ return "{\"settings\":{\"number_of_shards\":" + noOfShards
+ + ",\"number_of_replicas\":" + noOfReplicas
+ + "},\"mappings\":" + mappingsJson + "}";
+ }
+
+ private String loadSchemaFile() throws IOException {
+ String jarLocation;
+
+ try {
+ jarLocation = this.getClass().getProtectionDomain().getCodeSource().getLocation().toURI().getPath();
+ } catch (Exception e) {
+ throw new IOException("Cannot determine Ranger home directory", e);
+ }
+
+ String rangerHomeDir = new File(jarLocation).getParentFile().getParentFile().getParentFile().getPath();
+ String schemaPath = Paths.get(rangerHomeDir, "contrib", "opensearch_for_audit_setup", "conf", SCHEMA_FILE_NAME).toString();
+
+ File schemaFile = new File(schemaPath);
+
+ if (!schemaFile.exists()) {
+ throw new IOException("OpenSearch schema file not found: " + schemaPath);
+ }
+
+ return new String(Files.readAllBytes(Paths.get(schemaPath)));
+ }
+
+ private void logErrorAndWait(int retryCounter, String message) {
+ int attemptsLeft = maxRetry == -1 ? -1 : maxRetry - retryCounter;
+
+ LOG.severe("Error during OpenSearch bootstrap. [retrying after " + timeInterval
+ + " ms]. Attempts left: " + attemptsLeft + ". Error: " + message);
+
+ try {
+ Thread.sleep(timeInterval);
+ } catch (InterruptedException e) {
+ LOG.warning("OpenSearch bootstrap thread interrupted");
+
+ Thread.currentThread().interrupt();
+ }
+ }
+}
diff --git a/security-admin/contrib/opensearch_for_audit_setup/conf/ranger_opensearch_schema.json b/security-admin/contrib/opensearch_for_audit_setup/conf/ranger_opensearch_schema.json
new file mode 100644
index 000000000..f8f3bbf2d
--- /dev/null
+++ b/security-admin/contrib/opensearch_for_audit_setup/conf/ranger_opensearch_schema.json
@@ -0,0 +1,136 @@
+{
+ "properties": {
+ "_expire_at_": {
+ "type": "date",
+ "store": true,
+ "doc_values": true
+ },
+ "_ttl_": {
+ "type": "text",
+ "store": true
+ },
+ "_version_": {
+ "type": "long",
+ "store": true,
+ "index": false
+ },
+ "access": {
+ "type": "keyword"
+ },
+ "action": {
+ "type": "keyword"
+ },
+ "agent": {
+ "type": "keyword"
+ },
+ "agentHost": {
+ "type": "keyword"
+ },
+ "cliIP": {
+ "type": "keyword"
+ },
+ "cliType": {
+ "type": "keyword"
+ },
+ "cluster": {
+ "type": "keyword"
+ },
+ "reqContext": {
+ "type": "keyword"
+ },
+ "enforcer": {
+ "type": "keyword"
+ },
+ "event_count": {
+ "type": "long",
+ "doc_values": true
+ },
+ "event_dur_ms": {
+ "type": "long",
+ "doc_values": true
+ },
+ "evtTime": {
+ "type": "date",
+ "doc_values": true
+ },
+ "id": {
+ "type": "keyword",
+ "store": true
+ },
+ "logType": {
+ "type": "keyword"
+ },
+ "policy": {
+ "type": "long",
+ "doc_values": true
+ },
+ "proxyUsers": {
+ "type": "keyword"
+ },
+ "reason": {
+ "type": "text"
+ },
+ "repo": {
+ "type": "keyword"
+ },
+ "repoType": {
+ "type": "integer",
+ "doc_values": true
+ },
+ "req_caller_id": {
+ "type": "keyword"
+ },
+ "req_self_id": {
+ "type": "keyword"
+ },
+ "reqData": {
+ "type": "text"
+ },
+ "reqUser": {
+ "type": "keyword"
+ },
+ "resType": {
+ "type": "keyword"
+ },
+ "resource": {
+ "type": "keyword"
+ },
+ "result": {
+ "type": "integer"
+ },
+ "seq_num": {
+ "type": "long",
+ "doc_values": true
+ },
+ "sess": {
+ "type": "keyword"
+ },
+ "tags": {
+ "type": "keyword"
+ },
+ "tags_str": {
+ "type": "text"
+ },
+ "datasets": {
+ "type": "keyword"
+ },
+ "projects": {
+ "type": "keyword"
+ },
+ "datasetIds": {
+ "type": "long"
+ },
+ "text": {
+ "type": "text"
+ },
+ "zoneName": {
+ "type": "keyword"
+ },
+ "policyVersion": {
+ "type": "long"
+ },
+ "additionalInfo": {
+ "type": "text"
+ }
+ }
+}
diff --git a/security-admin/pom.xml b/security-admin/pom.xml
index 491baf80f..4df6c6a82 100644
--- a/security-admin/pom.xml
+++ b/security-admin/pom.xml
@@ -498,6 +498,11 @@
<artifactId>ranger-audit-dest-es</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.ranger</groupId>
+ <artifactId>ranger-audit-dest-os</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.ranger</groupId>
<artifactId>ranger-authn</artifactId>
diff --git a/security-admin/scripts/install.properties b/security-admin/scripts/install.properties
index 0f5a093f6..e8a9f2f81 100644
--- a/security-admin/scripts/install.properties
+++ b/security-admin/scripts/install.properties
@@ -101,6 +101,19 @@ audit_elasticsearch_password=
audit_elasticsearch_index=
audit_elasticsearch_bootstrap_enabled=true
+# * OpenSearch audit store properties (when audit_store=opensearch)
+audit_opensearch_urls=
+audit_opensearch_port=9200
+audit_opensearch_protocol=http
+audit_opensearch_user=
+audit_opensearch_password=
+audit_opensearch_index=ranger_audits
+audit_opensearch_bootstrap_enabled=true
+# * OpenSearch authentication scheme: kerberos, basic, or none. Leave empty to infer from the properties above.
+audit_opensearch_authentication_type=
+audit_opensearch_kerberos_principal=
+audit_opensearch_kerberos_keytab=
+
# * audit_solr_url URL to Solr. E.g. http://<solr_host>:6083/solr/ranger_audits
audit_solr_urls=
diff --git a/security-admin/scripts/setup.sh b/security-admin/scripts/setup.sh
index 43ef4b704..ae2048edc 100755
--- a/security-admin/scripts/setup.sh
+++ b/security-admin/scripts/setup.sh
@@ -102,6 +102,16 @@ audit_elasticsearch_user=$(get_prop 'audit_elasticsearch_user' $PROPFILE)
audit_elasticsearch_password=$(get_prop 'audit_elasticsearch_password' $PROPFILE)
audit_elasticsearch_index=$(get_prop 'audit_elasticsearch_index' $PROPFILE)
audit_elasticsearch_bootstrap_enabled=$(get_prop 'audit_elasticsearch_bootstrap_enabled' $PROPFILE)
+audit_opensearch_urls=$(get_prop 'audit_opensearch_urls' $PROPFILE)
+audit_opensearch_protocol=$(get_prop 'audit_opensearch_protocol' $PROPFILE)
+audit_opensearch_port=$(get_prop 'audit_opensearch_port' $PROPFILE)
+audit_opensearch_user=$(get_prop 'audit_opensearch_user' $PROPFILE)
+audit_opensearch_password=$(get_prop 'audit_opensearch_password' $PROPFILE)
+audit_opensearch_index=$(get_prop 'audit_opensearch_index' $PROPFILE)
+audit_opensearch_bootstrap_enabled=$(get_prop 'audit_opensearch_bootstrap_enabled' $PROPFILE)
+audit_opensearch_authentication_type=$(get_prop_or_default 'audit_opensearch_authentication_type' $PROPFILE '')
+audit_opensearch_kerberos_principal=$(get_prop_or_default 'audit_opensearch_kerberos_principal' $PROPFILE '')
+audit_opensearch_kerberos_keytab=$(get_prop_or_default 'audit_opensearch_kerberos_keytab' $PROPFILE '')
audit_solr_urls=$(get_prop 'audit_solr_urls' $PROPFILE)
audit_solr_user=$(get_prop_or_default 'audit_solr_user' $PROPFILE '')
audit_solr_password=$(get_prop_or_default 'audit_solr_password' $PROPFILE '')
@@ -305,6 +315,16 @@ init_variables(){
exit 1
fi
fi
+ if [ "${audit_store}" == "opensearch" ] ;then
+ if [ "${audit_opensearch_urls}" == "" ] ;then
+ log "[I] Please provide valid URL for 'opensearch' audit store!"
+ exit 1
+ fi
+ if [ "${audit_opensearch_port}" == "" ] ;then
+ log "[I] Please provide valid port for 'opensearch' audit store!"
+ exit 1
+ fi
+ fi
if [ "${audit_store}" == "cloudwatch" ] ;then
if [ "${audit_cloudwatch_region}" == "" ] ;then
@@ -857,6 +877,50 @@ update_properties() {
fi
+ if [ "${audit_store}" == "opensearch" ]
+ then
+ propertyName=ranger.audit.opensearch.urls
+ newPropertyValue=${audit_opensearch_urls}
+ updatePropertyToFilePy $propertyName "${newPropertyValue}" $to_file_ranger
+
+ propertyName=ranger.audit.opensearch.protocol
+ newPropertyValue=${audit_opensearch_protocol}
+ updatePropertyToFilePy $propertyName "${newPropertyValue}" $to_file_ranger
+
+ propertyName=ranger.audit.opensearch.port
+ newPropertyValue=${audit_opensearch_port}
+ updatePropertyToFilePy $propertyName "${newPropertyValue}" $to_file_ranger
+
+ propertyName=ranger.audit.opensearch.user
+ newPropertyValue=${audit_opensearch_user}
+ updatePropertyToFilePy $propertyName "${newPropertyValue}" $to_file_ranger
+
+ propertyName=ranger.audit.opensearch.password
+ newPropertyValue=${audit_opensearch_password}
+ updatePropertyToFilePy $propertyName "${newPropertyValue}" $to_file_ranger
+
+ propertyName=ranger.audit.opensearch.index
+ newPropertyValue=${audit_opensearch_index}
+ updatePropertyToFilePy $propertyName "${newPropertyValue}" $to_file_ranger
+
+ propertyName=ranger.audit.opensearch.authentication.type
+ newPropertyValue=${audit_opensearch_authentication_type}
+ updatePropertyToFilePy $propertyName "${newPropertyValue}" $to_file_ranger
+
+ propertyName=ranger.audit.opensearch.kerberos.principal
+ newPropertyValue=${audit_opensearch_kerberos_principal}
+ updatePropertyToFilePy $propertyName "${newPropertyValue}" $to_file_ranger
+
+ propertyName=ranger.audit.opensearch.kerberos.keytab
+ newPropertyValue=${audit_opensearch_kerberos_keytab}
+ updatePropertyToFilePy $propertyName "${newPropertyValue}" $to_file_ranger
+
+ propertyName=ranger.audit.opensearch.bootstrap.enabled
+ newPropertyValue=${audit_opensearch_bootstrap_enabled}
+ updatePropertyToFilePy $propertyName "${newPropertyValue}" $to_file_ranger
+
+ fi
+
if [ "${audit_store}" == "cloudwatch" ]
then
propertyName=ranger.audit.amazon_cloudwatch.region
diff --git a/security-admin/src/main/java/org/apache/ranger/biz/AssetMgr.java b/security-admin/src/main/java/org/apache/ranger/biz/AssetMgr.java
index f995249c1..f1fbeed3d 100644
--- a/security-admin/src/main/java/org/apache/ranger/biz/AssetMgr.java
+++ b/security-admin/src/main/java/org/apache/ranger/biz/AssetMgr.java
@@ -145,6 +145,9 @@ public class AssetMgr extends AssetMgrBase {
@Autowired
ElasticSearchAccessAuditsService elasticSearchAccessAuditsService;
+ @Autowired
+ org.apache.ranger.opensearch.OpenSearchAccessAuditsService openSearchAccessAuditsService;
+
@Autowired
CloudWatchAccessAuditsService cloudWatchAccessAuditsService;
@@ -761,6 +764,8 @@ public VXAccessAuditList getAccessLogs(SearchCriteria searchCriteria) {
return solrAccessAuditsService.searchXAccessAudits(searchCriteria);
} else if (RangerBizUtil.AUDIT_STORE_ELASTIC_SEARCH.equalsIgnoreCase(xaBizUtil.getAuditDBType())) {
return elasticSearchAccessAuditsService.searchXAccessAudits(searchCriteria);
+ } else if (RangerBizUtil.AUDIT_STORE_OPENSEARCH.equalsIgnoreCase(xaBizUtil.getAuditDBType())) {
+ return openSearchAccessAuditsService.searchXAccessAudits(searchCriteria);
} else if (RangerBizUtil.AUDIT_STORE_CLOUD_WATCH.equalsIgnoreCase(xaBizUtil.getAuditDBType())) {
return cloudWatchAccessAuditsService.searchXAccessAudits(searchCriteria);
} else {
diff --git a/security-admin/src/main/java/org/apache/ranger/biz/RangerBizUtil.java b/security-admin/src/main/java/org/apache/ranger/biz/RangerBizUtil.java
index b54e96bb1..0fa8910d6 100644
--- a/security-admin/src/main/java/org/apache/ranger/biz/RangerBizUtil.java
+++ b/security-admin/src/main/java/org/apache/ranger/biz/RangerBizUtil.java
@@ -86,6 +86,7 @@ public class RangerBizUtil {
public static final String AUDIT_STORE_RDBMS = "DB";
public static final String AUDIT_STORE_SOLR = "solr";
public static final String AUDIT_STORE_ELASTIC_SEARCH = "elasticSearch";
+ public static final String AUDIT_STORE_OPENSEARCH = "opensearch";
public static final String AUDIT_STORE_CLOUD_WATCH = "cloudwatch";
public static final boolean BATCH_CLEAR_ENABLED = PropertiesUtil.getBooleanProperty("ranger.jpa.jdbc.batch-clear.enable", true);
public static final int POLICY_BATCH_SIZE = PropertiesUtil.getIntProperty("ranger.jpa.jdbc.batch-clear.size", 10);
diff --git a/security-admin/src/main/java/org/apache/ranger/biz/XAuditMgr.java b/security-admin/src/main/java/org/apache/ranger/biz/XAuditMgr.java
index 67b10246c..0934f67eb 100644
--- a/security-admin/src/main/java/org/apache/ranger/biz/XAuditMgr.java
+++ b/security-admin/src/main/java/org/apache/ranger/biz/XAuditMgr.java
@@ -44,6 +44,9 @@ public class XAuditMgr extends XAuditMgrBase {
@Autowired
ElasticSearchAccessAuditsService elasticSearchAccessAuditsService;
+ @Autowired
+ org.apache.ranger.opensearch.OpenSearchAccessAuditsService openSearchAccessAuditsService;
+
@Autowired
CloudWatchAccessAuditsService cloudWatchAccessAuditsService;
@@ -118,6 +121,8 @@ public VXAccessAuditList searchXAccessAudits(SearchCriteria searchCriteria) {
return solrAccessAuditsService.searchXAccessAudits(searchCriteria);
} else if (RangerBizUtil.AUDIT_STORE_ELASTIC_SEARCH.equalsIgnoreCase(auditDBType)) {
return elasticSearchAccessAuditsService.searchXAccessAudits(searchCriteria);
+ } else if (RangerBizUtil.AUDIT_STORE_OPENSEARCH.equalsIgnoreCase(auditDBType)) {
+ return openSearchAccessAuditsService.searchXAccessAudits(searchCriteria);
} else if (RangerBizUtil.AUDIT_STORE_CLOUD_WATCH.equalsIgnoreCase(auditDBType)) {
return cloudWatchAccessAuditsService.searchXAccessAudits(searchCriteria);
} else {
@@ -133,6 +138,8 @@ public VXLong getXAccessAuditSearchCount(SearchCriteria searchCriteria) {
return solrAccessAuditsService.getXAccessAuditSearchCount(searchCriteria);
} else if (RangerBizUtil.AUDIT_STORE_ELASTIC_SEARCH.equalsIgnoreCase(auditDBType)) {
return elasticSearchAccessAuditsService.getXAccessAuditSearchCount(searchCriteria);
+ } else if (RangerBizUtil.AUDIT_STORE_OPENSEARCH.equalsIgnoreCase(auditDBType)) {
+ return openSearchAccessAuditsService.getXAccessAuditSearchCount(searchCriteria);
} else if (RangerBizUtil.AUDIT_STORE_CLOUD_WATCH.equalsIgnoreCase(auditDBType)) {
return cloudWatchAccessAuditsService.getXAccessAuditSearchCount(searchCriteria);
} else {
diff --git a/security-admin/src/main/java/org/apache/ranger/elasticsearch/ElasticSearchMgr.java b/security-admin/src/main/java/org/apache/ranger/elasticsearch/ElasticSearchMgr.java
index 4b264daf6..2a2db5a48 100644
--- a/security-admin/src/main/java/org/apache/ranger/elasticsearch/ElasticSearchMgr.java
+++ b/security-admin/src/main/java/org/apache/ranger/elasticsearch/ElasticSearchMgr.java
@@ -168,6 +168,7 @@ synchronized RestHighLevelClient connect() {
RestClientBuilder restClientBuilder = getRestClientBuilder(urls, protocol, user, password, port);
client = new RestHighLevelClient(restClientBuilder);
+ me = client;
} catch (Throwable t) {
logger.error("Can't connect to ElasticSearch: {}", parameterString, t);
}
diff --git a/security-admin/src/main/java/org/apache/ranger/opensearch/OpenSearchAccessAuditsService.java b/security-admin/src/main/java/org/apache/ranger/opensearch/OpenSearchAccessAuditsService.java
new file mode 100644
index 000000000..340ac34df
--- /dev/null
+++ b/security-admin/src/main/java/org/apache/ranger/opensearch/OpenSearchAccessAuditsService.java
@@ -0,0 +1,312 @@
+/*
+ * 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.opensearch;
+
+import org.apache.ranger.audit.provider.MiscUtil;
+import org.apache.ranger.common.MessageEnums;
+import org.apache.ranger.common.PropertiesUtil;
+import org.apache.ranger.common.RESTErrorUtil;
+import org.apache.ranger.common.SearchCriteria;
+import org.apache.ranger.db.XXServiceDefDao;
+import org.apache.ranger.entity.XXService;
+import org.apache.ranger.entity.XXServiceDef;
+import org.apache.ranger.opensearch.OpenSearchUtil.OpenSearchSearchResult;
+import org.apache.ranger.plugin.util.JsonUtilsV2;
+import org.apache.ranger.view.VXAccessAudit;
+import org.apache.ranger.view.VXAccessAuditList;
+import org.apache.ranger.view.VXLong;
+import org.elasticsearch.client.RestClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Scope;
+import org.springframework.stereotype.Service;
+
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+@Service
+@Scope("singleton")
+public class OpenSearchAccessAuditsService extends org.apache.ranger.AccessAuditsService {
+ private static final Logger LOG = LoggerFactory.getLogger(OpenSearchAccessAuditsService.class);
+
+ @Autowired
+ OpenSearchMgr openSearchMgr;
+
+ @Autowired
+ OpenSearchUtil openSearchUtil;
+
+ public VXAccessAuditList searchXAccessAudits(SearchCriteria searchCriteria) {
+ RestClient client = openSearchMgr.getClient();
+ final boolean hiveQueryVisibility = PropertiesUtil.getBooleanProperty("ranger.audit.hive.query.visibility", true);
+
+ if (client == null) {
+ LOG.warn("OpenSearch client is null, so not running the query.");
+
+ throw restErrorUtil.createRESTException("Error connecting to OpenSearch", MessageEnums.ERROR_SYSTEM);
+ }
+
+ Map<String, Object> paramList = searchCriteria.getParamList();
+
+ updateUserExclusion(paramList);
+
+ OpenSearchSearchResult result;
+
+ try {
+ result = openSearchUtil.searchResources(searchCriteria, searchFields, sortFields, client, openSearchMgr.getIndex());
+ } catch (IOException e) {
+ LOG.warn("OpenSearch query failed: {}", e.getMessage());
+
+ throw restErrorUtil.createRESTException("Error querying OpenSearch", MessageEnums.ERROR_SYSTEM);
+ }
+
+ List<VXAccessAudit> xAccessAuditList = new ArrayList<>();
+
+ for (Map<String, Object> source : result.getSources()) {
+ VXAccessAudit vXAccessAudit = populateViewBean(source);
+ String serviceType = vXAccessAudit.getServiceType();
+ boolean isHive = "hive".equalsIgnoreCase(serviceType);
+
+ if (!hiveQueryVisibility && isHive) {
+ vXAccessAudit.setRequestData(null);
+ } else if (isHive) {
+ String accessType = vXAccessAudit.getAccessType();
+
+ if ("grant".equalsIgnoreCase(accessType) || "revoke".equalsIgnoreCase(accessType)) {
+ String requestData = vXAccessAudit.getRequestData();
+
+ if (requestData != null) {
+ try {
+ vXAccessAudit.setRequestData(java.net.URLDecoder.decode(requestData, "UTF-8"));
+ } catch (UnsupportedEncodingException e) {
+ LOG.warn("Error while encoding request data: {}", requestData, e);
+ }
+ }
+ }
+ }
+
+ xAccessAuditList.add(vXAccessAudit);
+ }
+
+ VXAccessAuditList returnList = new VXAccessAuditList();
+
+ returnList.setPageSize(searchCriteria.getMaxRows());
+ returnList.setResultSize(result.getSources().size());
+ returnList.setTotalCount(result.getTotalHits());
+ returnList.setStartIndex(searchCriteria.getStartIndex());
+ returnList.setVXAccessAudits(xAccessAuditList);
+
+ return returnList;
+ }
+
+ public void setRestErrorUtil(RESTErrorUtil restErrorUtil) {
+ this.restErrorUtil = restErrorUtil;
+ }
+
+ public VXLong getXAccessAuditSearchCount(SearchCriteria searchCriteria) {
+ long count = 100;
+ VXLong vXLong = new VXLong();
+
+ vXLong.setValue(count);
+
+ return vXLong;
+ }
+
+ private VXAccessAudit populateViewBean(Map<String, Object> source) {
+ VXAccessAudit accessAudit = new VXAccessAudit();
+ Object value;
+
+ value = source.get("id");
+ if (value != null) {
+ accessAudit.setId((long) value.hashCode());
+ }
+
+ value = source.get("cluster");
+ if (value != null) {
+ accessAudit.setClusterName(value.toString());
+ }
+
+ value = source.get("zoneName");
+ if (value != null) {
+ accessAudit.setZoneName(value.toString());
+ }
+
+ value = source.get("agentHost");
+ if (value != null) {
+ accessAudit.setAgentHost(value.toString());
+ }
+
+ value = source.get("policyVersion");
+ if (value != null) {
+ accessAudit.setPolicyVersion(MiscUtil.toLong(value));
+ }
+
+ value = source.get("access");
+ if (value != null) {
+ accessAudit.setAccessType(value.toString());
+ }
+
+ value = source.get("enforcer");
+ if (value != null) {
+ accessAudit.setAclEnforcer(value.toString());
+ }
+
+ value = source.get("agent");
+ if (value != null) {
+ accessAudit.setAgentId(value.toString());
+ }
+
+ value = source.get("repo");
+ if (value != null) {
+ accessAudit.setRepoName(value.toString());
+
+ XXService xxService = daoManager.getXXService().findByName(accessAudit.getRepoName());
+
+ if (xxService != null) {
+ accessAudit.setRepoDisplayName(xxService.getDisplayName());
+ }
+ }
+
+ value = source.get("sess");
+ if (value != null) {
+ accessAudit.setSessionId(value.toString());
+ }
+
+ value = source.get("reqUser");
+ if (value != null) {
+ accessAudit.setRequestUser(value.toString());
+ }
+
+ value = source.get("reqData");
+ if (value != null) {
+ accessAudit.setRequestData(value.toString());
+ }
+
+ value = source.get("resource");
+ if (value != null) {
+ accessAudit.setResourcePath(value.toString());
+ }
+
+ value = source.get("cliIP");
+ if (value != null) {
+ accessAudit.setClientIP(value.toString());
+ }
+
+ value = source.get("result");
+ if (value != null) {
+ accessAudit.setAccessResult(MiscUtil.toInt(value));
+ }
+
+ value = source.get("policy");
+ if (value != null) {
+ accessAudit.setPolicyId(MiscUtil.toLong(value));
+ }
+
+ value = source.get("repoType");
+ if (value != null) {
+ accessAudit.setRepoType(MiscUtil.toInt(value));
+
+ if (null != daoManager) {
+ XXServiceDefDao xxServiceDef = daoManager.getXXServiceDef();
+
+ if (xxServiceDef != null) {
+ XXServiceDef xServiceDef = xxServiceDef.getById((long) accessAudit.getRepoType());
+
+ if (xServiceDef != null) {
+ accessAudit.setServiceType(xServiceDef.getName());
+ accessAudit.setServiceTypeDisplayName(xServiceDef.getDisplayName());
+ }
+ }
+ }
+ }
+
+ value = source.get("resType");
+ if (value != null) {
+ accessAudit.setResourceType(value.toString());
+ }
+
+ value = source.get("reason");
+ if (value != null) {
+ accessAudit.setResultReason(value.toString());
+ }
+
+ value = source.get("action");
+ if (value != null) {
+ accessAudit.setAction(value.toString());
+ }
+
+ value = source.get("evtTime");
+ if (value != null) {
+ accessAudit.setEventTime(MiscUtil.toLocalDate(value));
+ }
+
+ value = source.get("seq_num");
+ if (value != null) {
+ accessAudit.setSequenceNumber(MiscUtil.toLong(value));
+ }
+
+ value = source.get("event_count");
+ if (value != null) {
+ accessAudit.setEventCount(MiscUtil.toLong(value));
+ }
+
+ value = source.get("event_dur_ms");
+ if (value != null) {
+ accessAudit.setEventDuration(MiscUtil.toLong(value));
+ }
+
+ value = source.get("tags");
+ if (value != null) {
+ accessAudit.setTags(value.toString());
+ }
+
+ value = source.get("datasets");
+ if (value != null) {
+ try {
+ accessAudit.setDatasets(JsonUtilsV2.nonSerializableObjToJson(value));
+ } catch (Exception e) {
+ LOG.warn("Failed to convert datasets to json", e);
+ }
+ }
+
+ value = source.get("projects");
+ if (value != null) {
+ try {
+ accessAudit.setProjects(JsonUtilsV2.nonSerializableObjToJson(value));
+ } catch (Exception e) {
+ LOG.warn("Failed to convert projects to json", e);
+ }
+ }
+
+ value = source.get("datasetIds");
+ if (value != null) {
+ try {
+ accessAudit.setDatasetIds(JsonUtilsV2.nonSerializableObjToJson(value));
+ } catch (Exception e) {
+ LOG.warn("Failed to convert datasetIds to json", e);
+ }
+ }
+
+ return accessAudit;
+ }
+}
diff --git a/security-admin/src/main/java/org/apache/ranger/opensearch/OpenSearchMgr.java b/security-admin/src/main/java/org/apache/ranger/opensearch/OpenSearchMgr.java
new file mode 100644
index 000000000..479db5ff7
--- /dev/null
+++ b/security-admin/src/main/java/org/apache/ranger/opensearch/OpenSearchMgr.java
@@ -0,0 +1,199 @@
+/*
+ * 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.opensearch;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthSchemeProvider;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.client.config.AuthSchemes;
+import org.apache.http.config.Lookup;
+import org.apache.http.config.RegistryBuilder;
+import org.apache.http.impl.auth.SPNegoSchemeFactory;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.ranger.audit.destination.OpenSearchAuditDestination;
+import org.apache.ranger.authorization.credutils.CredentialsProviderUtil;
+import org.apache.ranger.authorization.credutils.kerberos.KerberosCredentialsProvider;
+import org.apache.ranger.common.PropertiesUtil;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestClientBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+import javax.security.auth.Subject;
+import javax.security.auth.kerberos.KerberosTicket;
+
+import java.security.PrivilegedActionException;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.Locale;
+
+@Component
+public class OpenSearchMgr {
+ private static final Logger LOG = LoggerFactory.getLogger(OpenSearchMgr.class);
+
+ private static final String CONFIG_PREFIX = OpenSearchAuditDestination.CONFIG_PREFIX;
+ private static final String CONFIG_URLS = OpenSearchAuditDestination.CONFIG_URLS;
+ private static final String CONFIG_PORT = OpenSearchAuditDestination.CONFIG_PORT;
+ private static final String CONFIG_PROTOCOL = OpenSearchAuditDestination.CONFIG_PROTOCOL;
+ private static final String CONFIG_USER = OpenSearchAuditDestination.CONFIG_USER;
+ private static final String CONFIG_PASSWORD = OpenSearchAuditDestination.CONFIG_PASSWORD;
+ private static final String CONFIG_INDEX = OpenSearchAuditDestination.CONFIG_INDEX;
+ private static final String CONFIG_AUTH_TYPE = OpenSearchAuditDestination.CONFIG_AUTH_TYPE;
+ private static final String CONFIG_KERBEROS_PRINCIPAL = OpenSearchAuditDestination.CONFIG_KERBEROS_PRINCIPAL;
+ private static final String CONFIG_KERBEROS_KEYTAB = OpenSearchAuditDestination.CONFIG_KERBEROS_KEYTAB;
+
+ private volatile RestClient client;
+ private Subject subject;
+ private String user;
+ private String password;
+ private String index;
+ private String loginPrincipal;
+ private String loginKeytab;
+
+ public RestClient getClient() {
+ RestClient me = client;
+
+ if (me != null && subject != null) {
+ KerberosTicket ticket = CredentialsProviderUtil.getTGT(subject);
+
+ try {
+ if (new Date().getTime() > ticket.getEndTime().getTime()) {
+ client = null;
+ CredentialsProviderUtil.ticketExpireTime80 = 0;
+
+ me = connect();
+ } else if (CredentialsProviderUtil.ticketWillExpire(ticket)) {
+ subject = CredentialsProviderUtil.login(loginPrincipal, loginKeytab);
+ }
+ } catch (PrivilegedActionException e) {
+ LOG.error("PrivilegedActionException:", e);
+
+ throw new RuntimeException(e);
+ }
+
+ return me;
+ } else {
+ me = connect();
+ }
+
+ return me;
+ }
+
+ String getIndex() {
+ return index;
+ }
+
+ void setIndex(String index) {
+ this.index = index;
+ }
+
+ synchronized RestClient connect() {
+ RestClient me = client;
+
+ if (me == null) {
+ synchronized (OpenSearchMgr.class) {
+ me = client;
+
+ if (me == null) {
+ String urls = PropertiesUtil.getProperty(CONFIG_PREFIX + "." + CONFIG_URLS);
+ String protocol = PropertiesUtil.getProperty(CONFIG_PREFIX + "." + CONFIG_PROTOCOL, "http");
+
+ user = PropertiesUtil.getProperty(CONFIG_PREFIX + "." + CONFIG_USER, "");
+ password = PropertiesUtil.getProperty(CONFIG_PREFIX + "." + CONFIG_PASSWORD, "");
+
+ String authType = PropertiesUtil.getProperty(CONFIG_PREFIX + "." + CONFIG_AUTH_TYPE, "");
+ String kerberosPrincipal = PropertiesUtil.getProperty(CONFIG_PREFIX + "." + CONFIG_KERBEROS_PRINCIPAL, "");
+ String kerberosKeytab = PropertiesUtil.getProperty(CONFIG_PREFIX + "." + CONFIG_KERBEROS_KEYTAB, "");
+
+ int port = Integer.parseInt(PropertiesUtil.getProperty(CONFIG_PREFIX + "." + CONFIG_PORT, "9200"));
+
+ this.index = PropertiesUtil.getProperty(CONFIG_PREFIX + "." + CONFIG_INDEX, "ranger_audits");
+
+ String parameterString = String.format(Locale.ROOT, "User:%s, %s://%s:%s/%s", user, protocol, urls, port, index);
+
+ LOG.info("Initializing OpenSearch connection: {}", parameterString);
+
+ if (urls != null) {
+ urls = urls.trim();
+ }
+
+ if (StringUtils.isBlank(urls) || "NONE".equalsIgnoreCase(urls)) {
+ LOG.warn("OpenSearch URLs not configured or set to NONE");
+
+ return null;
+ }
+
+ try {
+ if (OpenSearchAuditDestination.AUTH_TYPE_KERBEROS.equals(OpenSearchAuditDestination.resolveAuthType(authType, user, password))) {
+ loginPrincipal = OpenSearchAuditDestination.isConfigured(kerberosPrincipal) ? kerberosPrincipal : user;
+ loginKeytab = OpenSearchAuditDestination.isConfigured(kerberosKeytab) ? kerberosKeytab : password;
+ subject = CredentialsProviderUtil.login(loginPrincipal, loginKeytab);
+ }
+
+ RestClientBuilder builder = buildRestClientBuilder(urls, protocol, user, password, port, authType, kerberosPrincipal, kerberosKeytab);
+
+ client = builder.build();
+ me = client;
+ } catch (Throwable t) {
+ LOG.error("Cannot connect to OpenSearch: {}", parameterString, t);
+ }
+ }
+ }
+ }
+
+ return me;
+ }
+
+ public static RestClientBuilder buildRestClientBuilder(String urls, String protocol, String user, String password, int port) {
+ return buildRestClientBuilder(urls, protocol, user, password, port, "", "", "");
+ }
+
+ public static RestClientBuilder buildRestClientBuilder(String urls, String protocol, String user, String password, int port, String authType, String kerberosPrincipal, String kerberosKeytab) {
+ HttpHost[] hosts = Arrays.stream(urls.split(",")).map(String::trim).filter(h -> !h.isEmpty()).map(h -> new HttpHost(h, port, protocol)).toArray(HttpHost[]::new);
+
+ RestClientBuilder builder = RestClient.builder(hosts);
+
+ String resolvedAuthType = OpenSearchAuditDestination.resolveAuthType(authType, user, password);
+
+ if (OpenSearchAuditDestination.AUTH_TYPE_KERBEROS.equals(resolvedAuthType)) {
+ String principal = OpenSearchAuditDestination.isConfigured(kerberosPrincipal) ? kerberosPrincipal : user;
+ String keytab = OpenSearchAuditDestination.isConfigured(kerberosKeytab) ? kerberosKeytab : password;
+
+ KerberosCredentialsProvider credentialsProvider = CredentialsProviderUtil.getKerberosCredentials(principal, keytab);
+ Lookup<AuthSchemeProvider> authRegistry = RegistryBuilder.<AuthSchemeProvider>create().register(AuthSchemes.SPNEGO, new SPNegoSchemeFactory()).build();
+
+ builder.setHttpClientConfigCallback(httpClientBuilder -> {
+ httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
+ httpClientBuilder.setDefaultAuthSchemeRegistry(authRegistry);
+ return httpClientBuilder;
+ });
+ } else if (OpenSearchAuditDestination.AUTH_TYPE_BASIC.equals(resolvedAuthType)) {
+ CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
+ credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(user, password));
+ builder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider));
+ }
+
+ return builder;
+ }
+}
diff --git a/security-admin/src/main/java/org/apache/ranger/opensearch/OpenSearchUtil.java b/security-admin/src/main/java/org/apache/ranger/opensearch/OpenSearchUtil.java
new file mode 100644
index 000000000..e9fab44a7
--- /dev/null
+++ b/security-admin/src/main/java/org/apache/ranger/opensearch/OpenSearchUtil.java
@@ -0,0 +1,342 @@
+/*
+ * 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.opensearch;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.entity.ContentType;
+import org.apache.http.nio.entity.NStringEntity;
+import org.apache.http.util.EntityUtils;
+import org.apache.ranger.common.PropertiesUtil;
+import org.apache.ranger.common.SearchCriteria;
+import org.apache.ranger.common.SearchField;
+import org.apache.ranger.common.SortField;
+import org.apache.ranger.common.StringUtil;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.RestClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.TimeZone;
+import java.util.stream.Collectors;
+
+@Component
+public class OpenSearchUtil {
+ private static final Logger LOG = LoggerFactory.getLogger(OpenSearchUtil.class);
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+ private static final String LUCENE_SPECIAL_CHARS = "+-=&|><!(){}[]^\"~*?:\\/";
+ private static final String DATE_FORMAT_STR = "yyyy-MM-dd'T'HH:mm:ss'Z'";
+
+ private static final ThreadLocal<DateFormat> DATE_FORMAT = ThreadLocal.withInitial(() -> {
+ SimpleDateFormat dateFormat = new SimpleDateFormat(DATE_FORMAT_STR);
+ String timeZone = PropertiesUtil.getProperty("xa.elasticSearch.timezone");
+
+ if (timeZone != null) {
+ LOG.info("Setting timezone to {}", timeZone);
+
+ try {
+ dateFormat.setTimeZone(TimeZone.getTimeZone(timeZone));
+ } catch (Throwable t) {
+ LOG.error("Error setting timezone. TimeZone = {}", timeZone);
+ }
+ }
+
+ return dateFormat;
+ });
+
+ @Autowired
+ StringUtil stringUtil;
+
+ public OpenSearchUtil() {
+ }
+
+ public OpenSearchSearchResult searchResources(SearchCriteria searchCriteria, List<SearchField> searchFields, List<SortField> sortFields, RestClient client, String index) throws IOException {
+ String body = buildSearchBody(searchCriteria, searchFields, sortFields);
+
+ LOG.debug("OpenSearch query on index [{}]: {}", index, body);
+
+ Request request = new Request("POST", "/" + index + "/_search");
+
+ request.setEntity(new NStringEntity(body, ContentType.APPLICATION_JSON));
+
+ Response response = client.performRequest(request);
+ String json = EntityUtils.toString(response.getEntity());
+ JsonNode root = MAPPER.readTree(json);
+ long totalHits = root.at("/hits/total/value").asLong(0);
+ JsonNode hitsArray = root.at("/hits/hits");
+ List<Map<String, Object>> sources = new ArrayList<>();
+
+ if (hitsArray.isArray()) {
+ for (JsonNode hit : hitsArray) {
+ JsonNode sourceNode = hit.get("_source");
+
+ if (sourceNode != null) {
+ @SuppressWarnings("unchecked")
+ Map<String, Object> source = MAPPER.convertValue(sourceNode, Map.class);
+
+ sources.add(source);
+ }
+ }
+ }
+
+ return new OpenSearchSearchResult(totalHits, sources);
+ }
+
+ public List<Map<String, Object>> fetchByIds(RestClient client, String index, List<String> ids) throws IOException {
+ if (ids == null || ids.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ Map<String, Object> mgetBody = new HashMap<>();
+
+ mgetBody.put("ids", ids);
+
+ String body = MAPPER.writeValueAsString(mgetBody);
+ Request request = new Request("POST", "/" + index + "/_mget");
+
+ request.setEntity(new NStringEntity(body, ContentType.APPLICATION_JSON));
+
+ Response response = client.performRequest(request);
+ String json = EntityUtils.toString(response.getEntity());
+ JsonNode root = MAPPER.readTree(json);
+ JsonNode docsNode = root.get("docs");
+ List<Map<String, Object>> results = new ArrayList<>();
+
+ if (docsNode != null && docsNode.isArray()) {
+ for (JsonNode doc : docsNode) {
+ if (doc.has("found") && doc.get("found").asBoolean()) {
+ JsonNode sourceNode = doc.get("_source");
+
+ if (sourceNode != null) {
+ @SuppressWarnings("unchecked")
+ Map<String, Object> source = MAPPER.convertValue(sourceNode, Map.class);
+
+ results.add(source);
+ }
+ }
+ }
+ }
+
+ return results;
+ }
+
+ String buildSearchBody(SearchCriteria searchCriteria, List<SearchField> searchFields, List<SortField> sortFields) {
+ List<Map<String, Object>> mustClauses = new ArrayList<>();
+ Date fromDate = null;
+ Date toDate = null;
+ String dateFieldName = null;
+
+ if (searchCriteria.getParamList() != null) {
+ for (SearchField field : searchFields) {
+ String clientFieldName = field.getClientFieldName();
+ Object paramValue = searchCriteria.getParamValue(clientFieldName);
+
+ if (paramValue == null || paramValue.toString().isEmpty()) {
+ continue;
+ }
+
+ String fieldName = field.getFieldName();
+ SearchField.DATA_TYPE dataType = field.getDataType();
+ SearchField.SEARCH_TYPE searchType = field.getSearchType();
+
+ if (dataType == SearchField.DATA_TYPE.DATE) {
+ if (paramValue instanceof Date) {
+ if (searchType == SearchField.SEARCH_TYPE.GREATER_EQUAL_THAN || searchType == SearchField.SEARCH_TYPE.GREATER_THAN) {
+ fromDate = (Date) paramValue;
+ dateFieldName = fieldName;
+ } else if (searchType == SearchField.SEARCH_TYPE.LESS_EQUAL_THAN || searchType == SearchField.SEARCH_TYPE.LESS_THAN) {
+ toDate = (Date) paramValue;
+ dateFieldName = fieldName;
+ }
+ }
+
+ continue;
+ }
+
+ Map<String, Object> clause = buildClause(fieldName, dataType, searchType, paramValue);
+
+ if (clause != null) {
+ mustClauses.add(clause);
+ }
+ }
+
+ if (fromDate != null || toDate != null) {
+ mustClauses.add(buildDateRange(dateFieldName, fromDate, toDate));
+ }
+ }
+
+ Map<String, Object> query = new LinkedHashMap<>();
+ Map<String, Object> bool = new HashMap<>();
+
+ bool.put("must", mustClauses.isEmpty() ? List.of(Map.of("match_all", Map.of())) : mustClauses);
+ query.put("query", Map.of("bool", bool));
+ query.put("from", searchCriteria.getStartIndex());
+ query.put("size", searchCriteria.getMaxRows());
+
+ String[] sortResolved = resolveSortField(searchCriteria, sortFields);
+
+ if (sortResolved != null) {
+ query.put("sort", List.of(Map.of(sortResolved[0], Map.of("order", sortResolved[1]))));
+ }
+
+ try {
+ return MAPPER.writeValueAsString(query);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to serialize OpenSearch query", e);
+ }
+ }
+
+ private Map<String, Object> buildClause(String fieldName, SearchField.DATA_TYPE dataType, SearchField.SEARCH_TYPE searchType, Object paramValue) {
+ if (fieldName.startsWith("-")) {
+ Map<String, Object> inner = buildClause(fieldName.substring(1), dataType, searchType, paramValue);
+
+ if (inner != null) {
+ return Map.of("bool", Map.of("must_not", List.of(inner)));
+ }
+ } else if (paramValue instanceof Collection) {
+ Collection<?> valueList = (Collection<?>) paramValue;
+
+ if (!valueList.isEmpty()) {
+ String queryString = valueList.stream().map(v -> "(" + escapeLucene(v.toString().trim().toLowerCase()) + ")").collect(Collectors.joining(" OR "));
+
+ return Map.of("query_string", Map.of("query", queryString, "default_field", fieldName));
+ }
+ } else if (searchType == SearchField.SEARCH_TYPE.PARTIAL) {
+ String value = paramValue.toString().trim();
+
+ if (!value.isEmpty()) {
+ return Map.of("query_string", Map.of("query", "*" + escapeLucene(value.toLowerCase()) + "*", "default_field", fieldName));
+ }
+ } else {
+ String value = paramValue.toString().trim();
+
+ if (!value.isEmpty()) {
+ return Map.of("match_phrase", Map.of(fieldName, escapeLucene(value.toLowerCase())));
+ }
+ }
+
+ return null;
+ }
+
+ private Map<String, Object> buildDateRange(String fieldName, Date fromDate, Date toDate) {
+ Map<String, Object> rangeParams = new LinkedHashMap<>();
+
+ rangeParams.put("format", DATE_FORMAT_STR);
+
+ if (fromDate != null) {
+ rangeParams.put("gte", DATE_FORMAT.get().format(fromDate));
+ }
+
+ if (toDate != null) {
+ rangeParams.put("lte", DATE_FORMAT.get().format(toDate));
+ }
+
+ return Map.of("range", Map.of(fieldName, rangeParams));
+ }
+
+ private String[] resolveSortField(SearchCriteria searchCriteria, List<SortField> sortFields) {
+ String sortBy = searchCriteria.getSortBy();
+ String querySortBy = null;
+
+ if (sortBy != null && !sortBy.trim().isEmpty()) {
+ sortBy = sortBy.trim();
+
+ for (SortField sortField : sortFields) {
+ if (sortBy.equalsIgnoreCase(sortField.getParamName())) {
+ querySortBy = sortField.getFieldName();
+
+ searchCriteria.setSortBy(sortField.getParamName());
+
+ break;
+ }
+ }
+ }
+
+ if (querySortBy == null) {
+ for (SortField sortField : sortFields) {
+ if (sortField.isDefault()) {
+ querySortBy = sortField.getFieldName();
+
+ searchCriteria.setSortBy(sortField.getParamName());
+ searchCriteria.setSortType(sortField.getDefaultOrder().name());
+
+ break;
+ }
+ }
+ }
+
+ if (querySortBy != null) {
+ String order = "desc".equalsIgnoreCase(searchCriteria.getSortType()) ? "desc" : "asc";
+
+ return new String[] {querySortBy, order};
+ }
+
+ return null;
+ }
+
+ static String escapeLucene(String value) {
+ StringBuilder sb = new StringBuilder();
+
+ for (int i = 0; i < value.length(); i++) {
+ char c = value.charAt(i);
+
+ if (LUCENE_SPECIAL_CHARS.indexOf(c) >= 0) {
+ sb.append('\\');
+ }
+
+ sb.append(c);
+ }
+
+ return sb.toString();
+ }
+
+ public static class OpenSearchSearchResult {
+ private final long totalHits;
+ private final List<Map<String, Object>> sources;
+
+ public OpenSearchSearchResult(long totalHits, List<Map<String, Object>> sources) {
+ this.totalHits = totalHits;
+ this.sources = sources;
+ }
+
+ public long getTotalHits() {
+ return totalHits;
+ }
+
+ public List<Map<String, Object>> getSources() {
+ return sources;
+ }
+ }
+}
diff --git a/security-admin/src/main/resources/conf.dist/ranger-admin-default-site.xml b/security-admin/src/main/resources/conf.dist/ranger-admin-default-site.xml
index 76ca23bd6..a69d5886d 100644
--- a/security-admin/src/main/resources/conf.dist/ranger-admin-default-site.xml
+++ b/security-admin/src/main/resources/conf.dist/ranger-admin-default-site.xml
@@ -516,6 +516,34 @@
<name>ranger.audit.elasticsearch.bootstrap.enabled</name>
<value>true</value>
</property>
+ <property>
+ <name>ranger.audit.opensearch.urls</name>
+ <value></value>
+ </property>
+ <property>
+ <name>ranger.audit.opensearch.protocol</name>
+ <value>http</value>
+ </property>
+ <property>
+ <name>ranger.audit.opensearch.port</name>
+ <value>9200</value>
+ </property>
+ <property>
+ <name>ranger.audit.opensearch.user</name>
+ <value></value>
+ </property>
+ <property>
+ <name>ranger.audit.opensearch.password</name>
+ <value></value>
+ </property>
+ <property>
+ <name>ranger.audit.opensearch.index</name>
+ <value>ranger_audits</value>
+ </property>
+ <property>
+ <name>ranger.audit.opensearch.bootstrap.enabled</name>
+ <value>true</value>
+ </property>
<property>
<name>ranger.audit.solr.max.retry</name>
<value>30</value>
diff --git a/security-admin/src/main/resources/conf.dist/ranger-admin-site.xml b/security-admin/src/main/resources/conf.dist/ranger-admin-site.xml
index a75cd05a3..4effba2da 100644
--- a/security-admin/src/main/resources/conf.dist/ranger-admin-site.xml
+++ b/security-admin/src/main/resources/conf.dist/ranger-admin-site.xml
@@ -79,6 +79,49 @@
<name>ranger.audit.elasticsearch.bootstrap.enabled</name>
<value>true</value>
</property>
+ <property>
+ <name>ranger.audit.opensearch.urls</name>
+ <value></value>
+ </property>
+ <property>
+ <name>ranger.audit.opensearch.protocol</name>
+ <value>http</value>
+ </property>
+ <property>
+ <name>ranger.audit.opensearch.port</name>
+ <value>9200</value>
+ </property>
+ <property>
+ <name>ranger.audit.opensearch.user</name>
+ <value></value>
+ </property>
+ <property>
+ <name>ranger.audit.opensearch.password</name>
+ <value></value>
+ </property>
+ <property>
+ <name>ranger.audit.opensearch.index</name>
+ <value>ranger_audits</value>
+ </property>
+ <property>
+ <name>ranger.audit.opensearch.authentication.type</name>
+ <value></value>
+ <description>OpenSearch authentication scheme: kerberos, basic, or none. When left empty it is inferred (kerberos if the password/keytab points to a keytab file, basic if user and password are set, otherwise none).</description>
+ </property>
+ <property>
+ <name>ranger.audit.opensearch.kerberos.principal</name>
+ <value></value>
+ <description>Kerberos principal used when authentication.type=kerberos. Falls back to ranger.audit.opensearch.user when unset.</description>
+ </property>
+ <property>
+ <name>ranger.audit.opensearch.kerberos.keytab</name>
+ <value></value>
+ <description>Path to the Kerberos keytab used when authentication.type=kerberos. Falls back to ranger.audit.opensearch.password when unset.</description>
+ </property>
+ <property>
+ <name>ranger.audit.opensearch.bootstrap.enabled</name>
+ <value>true</value>
+ </property>
<property>
<name>ranger.audit.amazon_cloudwatch.region</name>
<value>us-east-2</value>
diff --git a/security-admin/src/test/java/org/apache/ranger/opensearch/OpenSearchAccessAuditsServiceTest.java b/security-admin/src/test/java/org/apache/ranger/opensearch/OpenSearchAccessAuditsServiceTest.java
new file mode 100644
index 000000000..12975f390
--- /dev/null
+++ b/security-admin/src/test/java/org/apache/ranger/opensearch/OpenSearchAccessAuditsServiceTest.java
@@ -0,0 +1,185 @@
+/*
+ * 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.opensearch;
+
+import org.apache.ranger.common.MessageEnums;
+import org.apache.ranger.common.RESTErrorUtil;
+import org.apache.ranger.common.SearchCriteria;
+import org.apache.ranger.db.RangerDaoManager;
+import org.apache.ranger.db.XXServiceDao;
+import org.apache.ranger.db.XXServiceDefDao;
+import org.apache.ranger.opensearch.OpenSearchUtil.OpenSearchSearchResult;
+import org.apache.ranger.view.VXAccessAuditList;
+import org.elasticsearch.client.RestClient;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import javax.ws.rs.WebApplicationException;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class OpenSearchAccessAuditsServiceTest {
+ @InjectMocks
+ OpenSearchAccessAuditsService service;
+
+ @Mock
+ OpenSearchMgr openSearchMgr;
+
+ @Mock
+ OpenSearchUtil openSearchUtil;
+
+ @Mock
+ RESTErrorUtil restErrorUtil;
+
+ @Mock
+ RangerDaoManager daoManager;
+
+ @Mock
+ XXServiceDao xxServiceDao;
+
+ @Mock
+ XXServiceDefDao xxServiceDefDao;
+
+ @Mock
+ RestClient restClient;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ service.setRestErrorUtil(restErrorUtil);
+
+ java.lang.reflect.Field daoField = org.apache.ranger.AccessAuditsService.class.getDeclaredField("daoManager");
+
+ daoField.setAccessible(true);
+ daoField.set(service, daoManager);
+ }
+
+ @Test
+ void searchXAccessAudits_clientNull_throwsException() {
+ when(openSearchMgr.getClient()).thenReturn(null);
+ when(restErrorUtil.createRESTException(anyString(), any(MessageEnums.class)))
+ .thenReturn(new WebApplicationException(500));
+
+ SearchCriteria criteria = new SearchCriteria();
+
+ assertThrows(WebApplicationException.class, () -> service.searchXAccessAudits(criteria));
+ }
+
+ @Test
+ void searchXAccessAudits_ioException_throwsException() throws Exception {
+ when(openSearchMgr.getClient()).thenReturn(restClient);
+ when(openSearchMgr.getIndex()).thenReturn("ranger_audits");
+ when(openSearchUtil.searchResources(any(), any(), any(), eq(restClient), eq("ranger_audits")))
+ .thenThrow(new IOException("Connection refused"));
+ when(restErrorUtil.createRESTException(anyString(), any(MessageEnums.class)))
+ .thenReturn(new WebApplicationException(500));
+
+ SearchCriteria criteria = new SearchCriteria();
+
+ assertThrows(WebApplicationException.class, () -> service.searchXAccessAudits(criteria));
+ }
+
+ @Test
+ void searchXAccessAudits_success() throws Exception {
+ when(openSearchMgr.getClient()).thenReturn(restClient);
+ when(openSearchMgr.getIndex()).thenReturn("ranger_audits");
+
+ Map<String, Object> doc1 = new HashMap<>();
+
+ doc1.put("id", "test-id-001");
+ doc1.put("reqUser", "testuser1");
+ doc1.put("resource", "/tmp/test");
+ doc1.put("access", "read");
+ doc1.put("result", 1);
+ doc1.put("repo", "dev_hdfs");
+ doc1.put("repoType", 1);
+ doc1.put("action", "read");
+
+ List<Map<String, Object>> sources = new ArrayList<>();
+
+ sources.add(doc1);
+
+ OpenSearchSearchResult mockResult = new OpenSearchSearchResult(1, sources);
+
+ when(openSearchUtil.searchResources(any(), any(), any(), eq(restClient), anyString()))
+ .thenReturn(mockResult);
+ when(daoManager.getXXService()).thenReturn(xxServiceDao);
+ when(daoManager.getXXServiceDef()).thenReturn(xxServiceDefDao);
+
+ SearchCriteria criteria = new SearchCriteria();
+
+ criteria.setMaxRows(25);
+ criteria.setStartIndex(0);
+
+ VXAccessAuditList result = service.searchXAccessAudits(criteria);
+
+ assertNotNull(result);
+ assertEquals(1, result.getTotalCount());
+ assertEquals(1, result.getResultSize());
+ assertNotNull(result.getVXAccessAudits());
+ assertEquals(1, result.getVXAccessAudits().size());
+ assertEquals("testuser1", result.getVXAccessAudits().get(0).getRequestUser());
+ assertEquals("/tmp/test", result.getVXAccessAudits().get(0).getResourcePath());
+ assertEquals("read", result.getVXAccessAudits().get(0).getAccessType());
+ }
+
+ @Test
+ void searchXAccessAudits_emptyResults() throws Exception {
+ when(openSearchMgr.getClient()).thenReturn(restClient);
+ when(openSearchMgr.getIndex()).thenReturn("ranger_audits");
+
+ OpenSearchSearchResult mockResult = new OpenSearchSearchResult(0, new ArrayList<>());
+
+ when(openSearchUtil.searchResources(any(), any(), any(), eq(restClient), anyString()))
+ .thenReturn(mockResult);
+
+ SearchCriteria criteria = new SearchCriteria();
+
+ criteria.setMaxRows(25);
+ criteria.setStartIndex(0);
+
+ VXAccessAuditList result = service.searchXAccessAudits(criteria);
+
+ assertNotNull(result);
+ assertEquals(0, result.getTotalCount());
+ assertEquals(0, result.getResultSize());
+ assertTrue(result.getVXAccessAudits().isEmpty());
+ }
+
+ private static void assertTrue(boolean condition) {
+ org.junit.jupiter.api.Assertions.assertTrue(condition);
+ }
+}
diff --git a/security-admin/src/test/java/org/apache/ranger/opensearch/OpenSearchMgrTest.java b/security-admin/src/test/java/org/apache/ranger/opensearch/OpenSearchMgrTest.java
new file mode 100644
index 000000000..61731da28
--- /dev/null
+++ b/security-admin/src/test/java/org/apache/ranger/opensearch/OpenSearchMgrTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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.opensearch;
+
+import org.elasticsearch.client.RestClientBuilder;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+class OpenSearchMgrTest {
+ @Test
+ void buildRestClientBuilder_basicAuth() {
+ RestClientBuilder builder = OpenSearchMgr.buildRestClientBuilder("localhost", "http", "admin", "password", 9200);
+
+ assertNotNull(builder);
+ }
+
+ @Test
+ void buildRestClientBuilder_noAuth() {
+ RestClientBuilder builder = OpenSearchMgr.buildRestClientBuilder("localhost", "http", "", "", 9200);
+
+ assertNotNull(builder);
+ }
+
+ @Test
+ void buildRestClientBuilder_noneCredentials() {
+ RestClientBuilder builder = OpenSearchMgr.buildRestClientBuilder("localhost", "https", "NONE", "NONE", 9200);
+
+ assertNotNull(builder);
+ }
+
+ @Test
+ void buildRestClientBuilder_multipleHosts() {
+ RestClientBuilder builder = OpenSearchMgr.buildRestClientBuilder("host1,host2,host3", "http", "user", "pass", 9200);
+
+ assertNotNull(builder);
+ }
+}
diff --git a/security-admin/src/test/java/org/apache/ranger/opensearch/OpenSearchUtilTest.java b/security-admin/src/test/java/org/apache/ranger/opensearch/OpenSearchUtilTest.java
new file mode 100644
index 000000000..4da67cab8
--- /dev/null
+++ b/security-admin/src/test/java/org/apache/ranger/opensearch/OpenSearchUtilTest.java
@@ -0,0 +1,219 @@
+/*
+ * 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.opensearch;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.ranger.common.SearchCriteria;
+import org.apache.ranger.common.SearchField;
+import org.apache.ranger.common.SortField;
+import org.apache.ranger.common.StringUtil;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.when;
+
+class OpenSearchUtilTest {
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ @InjectMocks
+ OpenSearchUtil openSearchUtil;
+
+ @Mock
+ StringUtil stringUtil;
+
+ @BeforeEach
+ void setUp() {
+ MockitoAnnotations.openMocks(this);
+
+ when(stringUtil.isEmpty(anyString())).thenAnswer(inv -> {
+ String s = inv.getArgument(0);
+ return s == null || s.trim().isEmpty();
+ });
+ }
+
+ @Test
+ void buildSearchBody_emptyParams() throws Exception {
+ SearchCriteria criteria = new SearchCriteria();
+
+ criteria.setMaxRows(25);
+ criteria.setStartIndex(0);
+
+ List<SearchField> searchFields = new ArrayList<>();
+ List<SortField> sortFields = List.of(new SortField("eventTime", "evtTime", true, SortField.SORT_ORDER.DESC));
+ String body = openSearchUtil.buildSearchBody(criteria, searchFields, sortFields);
+ JsonNode root = MAPPER.readTree(body);
+
+ assertNotNull(root.get("query"));
+ assertEquals(0, root.get("from").asInt());
+ assertEquals(25, root.get("size").asInt());
+ }
+
+ @Test
+ void buildSearchBody_partialStringSearch() throws Exception {
+ SearchCriteria criteria = new SearchCriteria();
+
+ criteria.setMaxRows(10);
+ criteria.setStartIndex(0);
+ criteria.addParam("requestUser", "testuser");
+
+ List<SearchField> searchFields = List.of(new SearchField("requestUser", "reqUser", SearchField.DATA_TYPE.STRING, SearchField.SEARCH_TYPE.PARTIAL));
+ List<SortField> sortFields = List.of(new SortField("eventTime", "evtTime", true, SortField.SORT_ORDER.DESC));
+ String body = openSearchUtil.buildSearchBody(criteria, searchFields, sortFields);
+ JsonNode root = MAPPER.readTree(body);
+ JsonNode mustClauses = root.at("/query/bool/must");
+ String bodyStr = body.toLowerCase();
+
+ assertTrue(mustClauses.isArray());
+ assertTrue(mustClauses.size() > 0);
+
+ assertTrue(bodyStr.contains("testuser"));
+ assertTrue(bodyStr.contains("query_string"));
+ }
+
+ @Test
+ void buildSearchBody_fullStringMatch() throws Exception {
+ SearchCriteria criteria = new SearchCriteria();
+
+ criteria.setMaxRows(10);
+ criteria.setStartIndex(0);
+ criteria.addParam("accessType", "read");
+
+ List<SearchField> searchFields = List.of(new SearchField("accessType", "access", SearchField.DATA_TYPE.STRING, SearchField.SEARCH_TYPE.FULL));
+ List<SortField> sortFields = List.of(new SortField("eventTime", "evtTime", true, SortField.SORT_ORDER.DESC));
+ String body = openSearchUtil.buildSearchBody(criteria, searchFields, sortFields);
+
+ assertTrue(body.contains("match_phrase"));
+ assertTrue(body.contains("access"));
+ assertTrue(body.contains("read"));
+ }
+
+ @Test
+ void buildSearchBody_dateRange() throws Exception {
+ SearchCriteria criteria = new SearchCriteria();
+
+ criteria.setMaxRows(10);
+ criteria.setStartIndex(0);
+ criteria.addParam("startDate", new Date(1700000000000L));
+ criteria.addParam("endDate", new Date(1700100000000L));
+
+ List<SearchField> searchFields = List.of(
+ new SearchField("startDate", "evtTime", SearchField.DATA_TYPE.DATE, SearchField.SEARCH_TYPE.GREATER_EQUAL_THAN),
+ new SearchField("endDate", "evtTime", SearchField.DATA_TYPE.DATE, SearchField.SEARCH_TYPE.LESS_EQUAL_THAN));
+ List<SortField> sortFields = List.of(new SortField("eventTime", "evtTime", true, SortField.SORT_ORDER.DESC));
+
+ String body = openSearchUtil.buildSearchBody(criteria, searchFields, sortFields);
+
+ assertTrue(body.contains("range"));
+ assertTrue(body.contains("evtTime"));
+ assertTrue(body.contains("gte"));
+ assertTrue(body.contains("lte"));
+ }
+
+ @Test
+ void buildSearchBody_collectionOrQuery() throws Exception {
+ SearchCriteria criteria = new SearchCriteria();
+
+ criteria.setMaxRows(10);
+ criteria.setStartIndex(0);
+ criteria.addParam("requestUser", Arrays.asList("user1", "user2", "user3"));
+
+ List<SearchField> searchFields = List.of(new SearchField("requestUser", "reqUser", SearchField.DATA_TYPE.STR_LIST, SearchField.SEARCH_TYPE.FULL));
+ List<SortField> sortFields = List.of(new SortField("eventTime", "evtTime", true, SortField.SORT_ORDER.DESC));
+ String body = openSearchUtil.buildSearchBody(criteria, searchFields, sortFields);
+
+ assertTrue(body.contains("query_string"));
+ assertTrue(body.contains("OR"));
+ assertTrue(body.contains("user1"));
+ assertTrue(body.contains("user2"));
+ assertTrue(body.contains("user3"));
+ }
+
+ @Test
+ void buildSearchBody_negation() throws Exception {
+ SearchCriteria criteria = new SearchCriteria();
+
+ criteria.setMaxRows(10);
+ criteria.setStartIndex(0);
+ criteria.addParam("excludeUser", "serviceuser");
+
+ List<SearchField> searchFields = List.of(new SearchField("excludeUser", "-reqUser", SearchField.DATA_TYPE.STRING, SearchField.SEARCH_TYPE.FULL));
+ List<SortField> sortFields = List.of(new SortField("eventTime", "evtTime", true, SortField.SORT_ORDER.DESC));
+ String body = openSearchUtil.buildSearchBody(criteria, searchFields, sortFields);
+
+ assertTrue(body.contains("must_not"));
+ assertTrue(body.contains("reqUser"));
+ }
+
+ @Test
+ void buildSearchBody_sorting() throws Exception {
+ SearchCriteria criteria = new SearchCriteria();
+
+ criteria.setMaxRows(10);
+ criteria.setStartIndex(5);
+ criteria.setSortBy("eventTime");
+ criteria.setSortType("asc");
+
+ List<SearchField> searchFields = new ArrayList<>();
+ List<SortField> sortFields = List.of(new SortField("eventTime", "evtTime", true, SortField.SORT_ORDER.DESC));
+ String body = openSearchUtil.buildSearchBody(criteria, searchFields, sortFields);
+ JsonNode root = MAPPER.readTree(body);
+ JsonNode sortNode = root.get("sort");
+
+ assertEquals(5, root.get("from").asInt());
+ assertEquals(10, root.get("size").asInt());
+
+ assertNotNull(sortNode);
+ assertTrue(sortNode.isArray());
+ assertTrue(sortNode.get(0).has("evtTime"));
+ assertEquals("asc", sortNode.get(0).at("/evtTime/order").asText());
+ }
+
+ @Test
+ void escapeLucene_specialCharacters() {
+ String input = "test+value:with*special?chars";
+ String escaped = OpenSearchUtil.escapeLucene(input);
+
+ assertTrue(escaped.contains("\\+"));
+ assertTrue(escaped.contains("\\:"));
+ assertTrue(escaped.contains("\\*"));
+ assertTrue(escaped.contains("\\?"));
+ }
+
+ @Test
+ void escapeLucene_noSpecialChars() {
+ String input = "simplevalue";
+ String escaped = OpenSearchUtil.escapeLucene(input);
+
+ assertEquals("simplevalue", escaped);
+ }
+}