mneethiraj commented on code in PR #847:
URL: https://github.com/apache/ranger/pull/847#discussion_r2888269055
##########
agents-audit/core/src/main/java/org/apache/ranger/audit/provider/AuditHandler.java:
##########
@@ -29,6 +29,8 @@ public interface AuditHandler {
boolean log(Collection<AuditEventBase> events);
+ boolean log(Collection<AuditEventBase> events, String batchKey);
Review Comment:
consider implementing as `default` method, to avoid update in all
implementation classes. This is critical to avoid breaking any implementation
that may not be in Ranger git repo.
```
default boolean log(Collection<AuditEventBase> events, String batchKey) {
log(events);
}
```
With this default method in place, please remove this method from
implementation classes - BaseAuditHandler, DummyAuditProvider
##########
agents-audit/core/src/main/java/org/apache/ranger/audit/model/AuthzAuditEvent.java:
##########
@@ -19,16 +19,28 @@
package org.apache.ranger.audit.model;
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.databind.annotation.JsonSerialize;
import org.apache.commons.lang3.StringUtils;
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlRootElement;
+
import java.util.Date;
import java.util.HashSet;
import java.util.Set;
+@JsonAutoDetect(fieldVisibility = JsonAutoDetect.Visibility.ANY)
+@JsonInclude(JsonInclude.Include.NON_EMPTY)
+@JsonIgnoreProperties(ignoreUnknown = true)
@JsonSerialize
+@XmlRootElement
Review Comment:
@rameeshm - we removed all `@Xml` annotations a while back (RANGER-4350).
Is it necessary to bring them back here?
##########
agents-audit/core/pom.xml:
##########
@@ -63,6 +63,11 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>javax.xml.bind</groupId>
Review Comment:
If `javax.xml.bind:jaxb-api` was added for `@Xml` annotations, consider
removing this dependency.
##########
agents-audit/dest-auditserver/src/main/java/org/apache/ranger/audit/destination/RangerAuditServerDestination.java:
##########
@@ -0,0 +1,429 @@
+/*
+ * 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.sun.jersey.api.client.ClientResponse;
+import com.sun.jersey.api.client.WebResource;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.ranger.audit.model.AuditEventBase;
+import org.apache.ranger.audit.provider.MiscUtil;
+import org.apache.ranger.plugin.util.RangerRESTClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.servlet.http.HttpServletResponse;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.Charset;
+import java.security.PrivilegedExceptionAction;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+public class RangerAuditServerDestination extends AuditDestination {
+ public static final String PROP_AUDITSERVER_URL =
"xasecure.audit.destination.auditserver.url";
+ public static final String PROP_AUDITSERVER_USER_NAME =
"xasecure.audit.destination.auditserver.username";
+ public static final String PROP_AUDITSERVER_USER_PASSWORD =
"xasecure.audit.destination.auditserver.password";
+ public static final String PROP_AUDITSERVER_AUTH_TYPE =
"xasecure.audit.destination.auditserver.authentication.type";
+ public static final String PROP_AUDITSERVER_JWT_TOKEN =
"xasecure.audit.destination.auditserver.jwt.token";
+ public static final String PROP_AUDITSERVER_JWT_TOKEN_FILE =
"xasecure.audit.destination.auditserver.jwt.token.file";
+ public static final String PROP_AUDITSERVER_CLIENT_CONN_TIMEOUT_MS =
"xasecure.audit.destination.auditserver.connection.timeout.ms";
+ public static final String PROP_AUDITSERVER_CLIENT_READ_TIMEOUT_MS =
"xasecure.audit.destination.auditserver.read.timeout.ms";
+ public static final String PROP_AUDITSERVER_SSL_CONFIG_FILE =
"xasecure.audit.destination.auditserver.ssl.config.file";
+ public static final String PROP_AUDITSERVER_MAX_RETRY_ATTEMPTS =
"xasecure.audit.destination.auditserver.max.retry.attempts";
+ public static final String PROP_AUDITSERVER_RETRY_INTERVAL_MS =
"xasecure.audit.destination.auditserver.retry.interval.ms";
+ public static final String PROP_SERVICE_TYPE =
"ranger.plugin.audit.service.type";
Review Comment:
Why are serviceType and appId needed for audit destination? How about using
existing fields in `AuthzAuditEvent` - `repoType` (serviceType),
`repositoryName` (serviceName), `agentType` (appId) instead, if necessary?
It will help avoid additional configurations that need to be setup in plugin.
##########
agents-audit/dest-auditserver/src/main/java/org/apache/ranger/audit/destination/RangerAuditServerDestination.java:
##########
@@ -0,0 +1,429 @@
+/*
+ * 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.sun.jersey.api.client.ClientResponse;
+import com.sun.jersey.api.client.WebResource;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.ranger.audit.model.AuditEventBase;
+import org.apache.ranger.audit.provider.MiscUtil;
+import org.apache.ranger.plugin.util.RangerRESTClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.servlet.http.HttpServletResponse;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.Charset;
+import java.security.PrivilegedExceptionAction;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+public class RangerAuditServerDestination extends AuditDestination {
+ public static final String PROP_AUDITSERVER_URL =
"xasecure.audit.destination.auditserver.url";
+ public static final String PROP_AUDITSERVER_USER_NAME =
"xasecure.audit.destination.auditserver.username";
+ public static final String PROP_AUDITSERVER_USER_PASSWORD =
"xasecure.audit.destination.auditserver.password";
+ public static final String PROP_AUDITSERVER_AUTH_TYPE =
"xasecure.audit.destination.auditserver.authentication.type";
+ public static final String PROP_AUDITSERVER_JWT_TOKEN =
"xasecure.audit.destination.auditserver.jwt.token";
+ public static final String PROP_AUDITSERVER_JWT_TOKEN_FILE =
"xasecure.audit.destination.auditserver.jwt.token.file";
+ public static final String PROP_AUDITSERVER_CLIENT_CONN_TIMEOUT_MS =
"xasecure.audit.destination.auditserver.connection.timeout.ms";
+ public static final String PROP_AUDITSERVER_CLIENT_READ_TIMEOUT_MS =
"xasecure.audit.destination.auditserver.read.timeout.ms";
+ public static final String PROP_AUDITSERVER_SSL_CONFIG_FILE =
"xasecure.audit.destination.auditserver.ssl.config.file";
+ public static final String PROP_AUDITSERVER_MAX_RETRY_ATTEMPTS =
"xasecure.audit.destination.auditserver.max.retry.attempts";
+ public static final String PROP_AUDITSERVER_RETRY_INTERVAL_MS =
"xasecure.audit.destination.auditserver.retry.interval.ms";
+ public static final String PROP_SERVICE_TYPE =
"ranger.plugin.audit.service.type";
+ public static final String PROP_APP_ID =
"ranger.plugin.audit.app.id";
+ public static final String REST_RELATIVE_PATH_POST =
"/api/audit/access";
+ public static final String QUERY_PARAM_SERVICE_NAME =
"serviceName";
+ public static final String QUERY_PARAM_APP_ID =
"appId";
+
+ // Authentication types
+ public static final String AUTH_TYPE_KERBEROS =
"kerberos";
+ public static final String AUTH_TYPE_BASIC =
"basic";
+ public static final String AUTH_TYPE_JWT =
"jwt";
+
+ private static final Logger LOG =
LoggerFactory.getLogger(RangerAuditServerDestination.class);
+ private RangerRESTClient restClient;
+ private String authType;
+ private String serviceType;
+ private String appId;
+
+ @Override
+ public void init(Properties props, String propPrefix) {
+ LOG.info("==> RangerAuditServerDestination:init()");
+ super.init(props, propPrefix);
+
+ String url = MiscUtil.getStringProperty(props,
PROP_AUDITSERVER_URL);
+ String sslConfigFileName = MiscUtil.getStringProperty(props,
PROP_AUDITSERVER_SSL_CONFIG_FILE);
+ String userName = MiscUtil.getStringProperty(props,
PROP_AUDITSERVER_USER_NAME);
+ String password = MiscUtil.getStringProperty(props,
PROP_AUDITSERVER_USER_PASSWORD);
+ int connTimeoutMs = MiscUtil.getIntProperty(props,
PROP_AUDITSERVER_CLIENT_CONN_TIMEOUT_MS, 120000);
+ int readTimeoutMs = MiscUtil.getIntProperty(props,
PROP_AUDITSERVER_CLIENT_READ_TIMEOUT_MS, 30000);
+ int maxRetryAttempts = MiscUtil.getIntProperty(props,
PROP_AUDITSERVER_MAX_RETRY_ATTEMPTS, 3);
+ int retryIntervalMs = MiscUtil.getIntProperty(props,
PROP_AUDITSERVER_RETRY_INTERVAL_MS, 1000);
+
+ this.authType = MiscUtil.getStringProperty(props,
PROP_AUDITSERVER_AUTH_TYPE);
+ this.serviceType = MiscUtil.getStringProperty(props,
PROP_SERVICE_TYPE);
+ this.appId = MiscUtil.getStringProperty(props, PROP_APP_ID);
+
+ if (StringUtils.isEmpty(authType)) {
+ // Authentication priority: JWT → Kerberos → Basic
+ try {
+ if (StringUtils.isNotEmpty(MiscUtil.getStringProperty(props,
PROP_AUDITSERVER_JWT_TOKEN)) ||
+
StringUtils.isNotEmpty(MiscUtil.getStringProperty(props,
PROP_AUDITSERVER_JWT_TOKEN_FILE))) {
+ this.authType = AUTH_TYPE_JWT;
+ } else if (isKerberosAuthenticated()) {
+ this.authType = AUTH_TYPE_KERBEROS;
+ } else if (StringUtils.isNotEmpty(userName)) {
+ this.authType = AUTH_TYPE_BASIC;
+ }
+ } catch (Exception e) {
+ LOG.warn("Failed to auto-detect authentication type", e);
+ }
+ }
+
+ LOG.info("Audit destination authentication type: {}", authType);
+
+ if (StringUtils.isEmpty(this.serviceType)) {
+ LOG.error("Service type not available in audit properties. This is
a configuration error. Audit destination will not function correctly.");
+ LOG.error("Ensure that RangerBasePlugin is properly initialized
with a valid serviceType (hdfs, hive, kafka, etc.)");
+ } else {
+ LOG.info("Audit destination configured for service type: {}",
this.serviceType);
+ }
+
+ if (StringUtils.isNotEmpty(this.appId)) {
+ LOG.info("Audit destination configured with appId: {}",
this.appId);
+ } else {
+ LOG.warn("appId not available in audit properties. Batch
processing may not be optimal.");
+ }
+
+ if (AUTH_TYPE_KERBEROS.equalsIgnoreCase(authType)) {
+ preAuthenticateKerberos();
+ }
+
+ Configuration config = createConfigurationFromProperties(props,
authType, userName, password);
+ this.restClient = new RangerRESTClient(url, sslConfigFileName, config);
+ this.restClient.setRestClientConnTimeOutMs(connTimeoutMs);
+ this.restClient.setRestClientReadTimeOutMs(readTimeoutMs);
+ this.restClient.setMaxRetryAttempts(maxRetryAttempts);
+ this.restClient.setRetryIntervalMs(retryIntervalMs);
+
+ LOG.info("<== RangerAuditServerDestination:init()");
+ }
+
+ private String initJwtToken() {
+ LOG.info("==> RangerAuditServerDestination:initJwtToken()");
+
+ String jwtToken = MiscUtil.getStringProperty(props,
PROP_AUDITSERVER_JWT_TOKEN);
+ if (StringUtils.isEmpty(jwtToken)) {
+ String jwtTokenFile = MiscUtil.getStringProperty(props,
PROP_AUDITSERVER_JWT_TOKEN_FILE);
+ if (StringUtils.isNotEmpty(jwtTokenFile)) {
+ try {
+ jwtToken = readJwtTokenFromFile(jwtTokenFile);
+ LOG.info("JWT token loaded from file: {}", jwtTokenFile);
+ } catch (Exception e) {
+ LOG.error("Failed to read JWT token from file: {}",
jwtTokenFile, e);
+ }
+ }
+ }
+
+ if (StringUtils.isEmpty(jwtToken)) {
+ LOG.warn("JWT authentication configured but no token found.
Configure {} or {}", PROP_AUDITSERVER_JWT_TOKEN,
PROP_AUDITSERVER_JWT_TOKEN_FILE);
+ } else {
+ LOG.info("JWT authentication initialized successfully");
+ }
+
+ LOG.info("<== RangerAuditServerDestination:initJwtToken()");
+
+ return jwtToken;
+ }
+
+ private String readJwtTokenFromFile(String tokenFile) throws IOException {
+ try (InputStream in = getFileInputStream(tokenFile)) {
+ if (in != null) {
+ return IOUtils.toString(in, Charset.defaultCharset()).trim();
+ } else {
+ throw new IOException("Unable to read JWT token file: " +
tokenFile);
+ }
+ }
+ }
+
+ private void preAuthenticateKerberos() {
Review Comment:
`preAuthenticateKerberos()` => `initKerberos()`
##########
agents-audit/dest-auditserver/pom.xml:
##########
@@ -0,0 +1,73 @@
+<?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-auditserver</artifactId>
+ <packaging>jar</packaging>
+ <name>Ranger Audit Destination - auditserver</name>
+ <description>Ranger Audit Destination - auditserver</description>
+ <properties>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <securesm.version>1.2</securesm.version>
+ </properties>
+ <dependencies>
+ <dependency>
+ <groupId>com.fasterxml.jackson.jaxrs</groupId>
+ <artifactId>jackson-jaxrs-json-provider</artifactId>
+ <version>${fasterxml.jackson.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.google.code.gson</groupId>
+ <artifactId>gson</artifactId>
Review Comment:
`gson` doesn't seem to be used. Please review and remove if unused.
##########
agents-audit/dest-auditserver/src/main/java/org/apache/ranger/audit/destination/RangerAuditServerDestination.java:
##########
@@ -0,0 +1,429 @@
+/*
+ * 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.sun.jersey.api.client.ClientResponse;
+import com.sun.jersey.api.client.WebResource;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.ranger.audit.model.AuditEventBase;
+import org.apache.ranger.audit.provider.MiscUtil;
+import org.apache.ranger.plugin.util.RangerRESTClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.servlet.http.HttpServletResponse;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.Charset;
+import java.security.PrivilegedExceptionAction;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+public class RangerAuditServerDestination extends AuditDestination {
+ public static final String PROP_AUDITSERVER_URL =
"xasecure.audit.destination.auditserver.url";
+ public static final String PROP_AUDITSERVER_USER_NAME =
"xasecure.audit.destination.auditserver.username";
+ public static final String PROP_AUDITSERVER_USER_PASSWORD =
"xasecure.audit.destination.auditserver.password";
+ public static final String PROP_AUDITSERVER_AUTH_TYPE =
"xasecure.audit.destination.auditserver.authentication.type";
+ public static final String PROP_AUDITSERVER_JWT_TOKEN =
"xasecure.audit.destination.auditserver.jwt.token";
+ public static final String PROP_AUDITSERVER_JWT_TOKEN_FILE =
"xasecure.audit.destination.auditserver.jwt.token.file";
+ public static final String PROP_AUDITSERVER_CLIENT_CONN_TIMEOUT_MS =
"xasecure.audit.destination.auditserver.connection.timeout.ms";
+ public static final String PROP_AUDITSERVER_CLIENT_READ_TIMEOUT_MS =
"xasecure.audit.destination.auditserver.read.timeout.ms";
+ public static final String PROP_AUDITSERVER_SSL_CONFIG_FILE =
"xasecure.audit.destination.auditserver.ssl.config.file";
+ public static final String PROP_AUDITSERVER_MAX_RETRY_ATTEMPTS =
"xasecure.audit.destination.auditserver.max.retry.attempts";
+ public static final String PROP_AUDITSERVER_RETRY_INTERVAL_MS =
"xasecure.audit.destination.auditserver.retry.interval.ms";
+ public static final String PROP_SERVICE_TYPE =
"ranger.plugin.audit.service.type";
+ public static final String PROP_APP_ID =
"ranger.plugin.audit.app.id";
+ public static final String REST_RELATIVE_PATH_POST =
"/api/audit/access";
+ public static final String QUERY_PARAM_SERVICE_NAME =
"serviceName";
+ public static final String QUERY_PARAM_APP_ID =
"appId";
+
+ // Authentication types
+ public static final String AUTH_TYPE_KERBEROS =
"kerberos";
+ public static final String AUTH_TYPE_BASIC =
"basic";
+ public static final String AUTH_TYPE_JWT =
"jwt";
+
+ private static final Logger LOG =
LoggerFactory.getLogger(RangerAuditServerDestination.class);
+ private RangerRESTClient restClient;
+ private String authType;
+ private String serviceType;
+ private String appId;
+
+ @Override
+ public void init(Properties props, String propPrefix) {
+ LOG.info("==> RangerAuditServerDestination:init()");
+ super.init(props, propPrefix);
+
+ String url = MiscUtil.getStringProperty(props,
PROP_AUDITSERVER_URL);
+ String sslConfigFileName = MiscUtil.getStringProperty(props,
PROP_AUDITSERVER_SSL_CONFIG_FILE);
+ String userName = MiscUtil.getStringProperty(props,
PROP_AUDITSERVER_USER_NAME);
+ String password = MiscUtil.getStringProperty(props,
PROP_AUDITSERVER_USER_PASSWORD);
+ int connTimeoutMs = MiscUtil.getIntProperty(props,
PROP_AUDITSERVER_CLIENT_CONN_TIMEOUT_MS, 120000);
+ int readTimeoutMs = MiscUtil.getIntProperty(props,
PROP_AUDITSERVER_CLIENT_READ_TIMEOUT_MS, 30000);
+ int maxRetryAttempts = MiscUtil.getIntProperty(props,
PROP_AUDITSERVER_MAX_RETRY_ATTEMPTS, 3);
+ int retryIntervalMs = MiscUtil.getIntProperty(props,
PROP_AUDITSERVER_RETRY_INTERVAL_MS, 1000);
+
+ this.authType = MiscUtil.getStringProperty(props,
PROP_AUDITSERVER_AUTH_TYPE);
+ this.serviceType = MiscUtil.getStringProperty(props,
PROP_SERVICE_TYPE);
+ this.appId = MiscUtil.getStringProperty(props, PROP_APP_ID);
+
+ if (StringUtils.isEmpty(authType)) {
+ // Authentication priority: JWT → Kerberos → Basic
+ try {
+ if (StringUtils.isNotEmpty(MiscUtil.getStringProperty(props,
PROP_AUDITSERVER_JWT_TOKEN)) ||
+
StringUtils.isNotEmpty(MiscUtil.getStringProperty(props,
PROP_AUDITSERVER_JWT_TOKEN_FILE))) {
+ this.authType = AUTH_TYPE_JWT;
+ } else if (isKerberosAuthenticated()) {
+ this.authType = AUTH_TYPE_KERBEROS;
+ } else if (StringUtils.isNotEmpty(userName)) {
+ this.authType = AUTH_TYPE_BASIC;
+ }
+ } catch (Exception e) {
+ LOG.warn("Failed to auto-detect authentication type", e);
+ }
+ }
+
+ LOG.info("Audit destination authentication type: {}", authType);
+
+ if (StringUtils.isEmpty(this.serviceType)) {
+ LOG.error("Service type not available in audit properties. This is
a configuration error. Audit destination will not function correctly.");
+ LOG.error("Ensure that RangerBasePlugin is properly initialized
with a valid serviceType (hdfs, hive, kafka, etc.)");
+ } else {
+ LOG.info("Audit destination configured for service type: {}",
this.serviceType);
+ }
+
+ if (StringUtils.isNotEmpty(this.appId)) {
+ LOG.info("Audit destination configured with appId: {}",
this.appId);
+ } else {
+ LOG.warn("appId not available in audit properties. Batch
processing may not be optimal.");
+ }
+
+ if (AUTH_TYPE_KERBEROS.equalsIgnoreCase(authType)) {
+ preAuthenticateKerberos();
+ }
+
+ Configuration config = createConfigurationFromProperties(props,
authType, userName, password);
+ this.restClient = new RangerRESTClient(url, sslConfigFileName, config);
+ this.restClient.setRestClientConnTimeOutMs(connTimeoutMs);
+ this.restClient.setRestClientReadTimeOutMs(readTimeoutMs);
+ this.restClient.setMaxRetryAttempts(maxRetryAttempts);
+ this.restClient.setRetryIntervalMs(retryIntervalMs);
+
+ LOG.info("<== RangerAuditServerDestination:init()");
+ }
+
+ private String initJwtToken() {
+ LOG.info("==> RangerAuditServerDestination:initJwtToken()");
+
+ String jwtToken = MiscUtil.getStringProperty(props,
PROP_AUDITSERVER_JWT_TOKEN);
+ if (StringUtils.isEmpty(jwtToken)) {
+ String jwtTokenFile = MiscUtil.getStringProperty(props,
PROP_AUDITSERVER_JWT_TOKEN_FILE);
+ if (StringUtils.isNotEmpty(jwtTokenFile)) {
+ try {
+ jwtToken = readJwtTokenFromFile(jwtTokenFile);
+ LOG.info("JWT token loaded from file: {}", jwtTokenFile);
+ } catch (Exception e) {
+ LOG.error("Failed to read JWT token from file: {}",
jwtTokenFile, e);
+ }
+ }
+ }
+
+ if (StringUtils.isEmpty(jwtToken)) {
+ LOG.warn("JWT authentication configured but no token found.
Configure {} or {}", PROP_AUDITSERVER_JWT_TOKEN,
PROP_AUDITSERVER_JWT_TOKEN_FILE);
+ } else {
+ LOG.info("JWT authentication initialized successfully");
+ }
+
+ LOG.info("<== RangerAuditServerDestination:initJwtToken()");
+
+ return jwtToken;
+ }
+
+ private String readJwtTokenFromFile(String tokenFile) throws IOException {
+ try (InputStream in = getFileInputStream(tokenFile)) {
+ if (in != null) {
+ return IOUtils.toString(in, Charset.defaultCharset()).trim();
+ } else {
+ throw new IOException("Unable to read JWT token file: " +
tokenFile);
+ }
+ }
+ }
+
+ private void preAuthenticateKerberos() {
+ LOG.info("==> RangerAuditServerDestination:preAuthenticateKerberos()");
+
+ try {
+ UserGroupInformation ugi = UserGroupInformation.getLoginUser();
+
+ if (ugi == null) {
+ LOG.warn("No UserGroupInformation available for Kerberos
pre-authentication");
+ return;
+ }
+
+ if (!ugi.hasKerberosCredentials()) {
+ LOG.warn("User {} does not have Kerberos credentials for
pre-authentication", ugi.getUserName());
+ return;
+ }
+
+ LOG.info("Pre-authenticating Kerberos for user: {}, authMethod:
{}", ugi.getUserName(), ugi.getAuthenticationMethod());
+
+ ugi.checkTGTAndReloginFromKeytab();
+ LOG.debug("TGT verified and refreshed if needed for user: {}",
ugi.getUserName());
+
+ LOG.info("Kerberos pre-authentication completed successfully");
+ } catch (Exception e) {
+ LOG.warn("Kerberos pre-authentication failed. First request will
retry authentication", e);
+ }
+
+ LOG.info("<== RangerAuditServerDestination:preAuthenticateKerberos()");
+ }
+
+ @Override
+ public void stop() {
+ LOG.info("==> RangerAuditServerDestination.stop() called..");
+ logStatus();
+ if (restClient != null) {
+ restClient.resetClient();
+ restClient = null;
+ }
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.ranger.audit.provider.AuditProvider#flush()
+ */
+ @Override
+ public void flush() {
+ }
+
+ @Override
+ public boolean log(Collection<AuditEventBase> events) {
+ boolean ret = false;
+ try {
+ logStatusIfRequired();
+ addTotalCount(events.size());
+
+ if (restClient == null) {
+ LOG.error("REST client is not initialized. Cannot send audit
events");
+ addDeferredCount(events.size());
+ return ret;
+ }
+
+ ret = logAsBatch(events);
+ } catch (Throwable t) {
+ addDeferredCount(events.size());
+ logError("Error sending audit to Audit Server", t);
+ }
+ return ret;
+ }
+
+ public boolean isAsync() {
+ return true;
+ }
+
+ private boolean logAsBatch(Collection<AuditEventBase> events) {
+ boolean batchSuccess = false;
+ int totalEvents = events.size();
+
+ LOG.debug("==> logAsBatch() Sending batch of {} events to Audit
Server....", totalEvents);
+
+ batchSuccess = sendBatch(events);
+ if (batchSuccess) {
+ addSuccessCount(totalEvents);
+ } else {
+ LOG.error("Failed to send batch of {} events", totalEvents);
+ addFailedCount(totalEvents);
+ }
+
+ LOG.debug("<== logAsBatch() Batch processing complete: {}/{} events
sent successfully", batchSuccess ? totalEvents : 0, totalEvents);
+
+ return batchSuccess;
+ }
+
+ private boolean sendBatch(Collection<AuditEventBase> events) {
+ boolean ret = false;
+ Map<String, String> queryParams = new HashMap<>();
+
+ // Add serviceName to query parameters
+ if (StringUtils.isNotEmpty(serviceType)) {
+ queryParams.put(QUERY_PARAM_SERVICE_NAME, serviceType);
Review Comment:
Parameter `QUERY_PARAM_SERVICE_NAME` is set to `serviceType`? Perhaps the
parameter should be renamed to `QUERY_PARAM_SERVICE_TYPE`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
