[ranger] branch master updated: RANGER-4400: Implemented processResults(Collection results) for RangerKafkaAuditHandler to prevent audit of cluster resource level topic creation while creating topic (#281)

[rmani]

This is an automated email from the ASF dual-hosted git repository.

rmani 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 da68108bf RANGER-4400: Implemented 
processResults(Collection<RangerAccessResult> results) for 
RangerKafkaAuditHandler to prevent audit of cluster resource level topic 
creation while creating topic (#281)
da68108bf is described below

commit da68108bfa3b7a451d6c6964406148f14185f127
Author: Fateh Singh <fateh...@gmail.com>
AuthorDate: Wed Sep 13 09:30:53 2023 -0700

    RANGER-4400: Implemented processResults(Collection<RangerAccessResult> 
results) for RangerKafkaAuditHandler to prevent audit of cluster resource level 
topic creation while creating topic (#281)
---
 .../kafka/authorizer/RangerKafkaAuditHandler.java  | 62 +++++++++++++++++++++-
 1 file changed, 60 insertions(+), 2 deletions(-)

diff --git 
a/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/RangerKafkaAuditHandler.java
 
b/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/RangerKafkaAuditHandler.java
index 57a888e9a..459e874f1 100644
--- 
a/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/RangerKafkaAuditHandler.java
+++ 
b/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/RangerKafkaAuditHandler.java
@@ -28,11 +28,16 @@ import 
org.apache.ranger.plugin.policyengine.RangerAccessResult;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.ArrayList;
+import java.util.Collection;
+
 public class RangerKafkaAuditHandler extends RangerDefaultAuditHandler {
     private static final Logger LOG = 
LoggerFactory.getLogger(RangerKafkaAuditHandler.class);
 
     private AuthzAuditEvent auditEvent      = null;
 
+    private ArrayList<AuthzAuditEvent> auditEventList = new ArrayList<>();
+
     public RangerKafkaAuditHandler(){
     }
 
@@ -40,13 +45,53 @@ public class RangerKafkaAuditHandler extends 
RangerDefaultAuditHandler {
     public void processResult(RangerAccessResult result) {
         // If Cluster Resource Level Topic Creation is not Allowed we don't 
audit.
         // Subsequent call from Kafka for Topic Creation at Topic resource 
Level will be audited.
+        if(LOG.isDebugEnabled()) {
+            LOG.debug("==> RangerKafkaAuditHandler.processResult()");
+        }
         if (!isAuditingNeeded(result)) {
             return;
         }
         auditEvent = super.getAuthzEvents(result);
+        if(LOG.isDebugEnabled()) {
+            LOG.debug("<== RangerKafkaAuditHandler.processResult()");
+        }
+    }
+    @Override
+    public void processResults(Collection<RangerAccessResult> results) {
+        if(LOG.isDebugEnabled()) {
+            LOG.debug("==> RangerKafkaAuditHandler.processResults(" + results 
+ ")");
+        }
+        for(RangerAccessResult res: results){
+            if (isAuditingNeeded(res)){
+                AuthzAuditEvent event = super.getAuthzEvents(res);
+                if(event!=null){
+                    if(LOG.isDebugEnabled()) {
+                        LOG.debug("Got event=" + event + " for 
RangerAccessResult=" + res);
+                    }
+                    auditEventList.add(event);
+                }
+                else{
+                    if(LOG.isDebugEnabled()) {
+                        LOG.debug("No audit event for :" + res);
+                    }
+                }
+            }
+            else {
+                if(LOG.isDebugEnabled()) {
+                    LOG.debug("Auditing not required for :"+res);
+                }
+            }
+        }
+
+        if(LOG.isDebugEnabled()) {
+            LOG.debug("<== RangerKafkaAuditHandler.processResults(" + results 
+ ")");
+        }
     }
 
     private boolean isAuditingNeeded(final RangerAccessResult result) {
+        if(LOG.isDebugEnabled()) {
+            LOG.debug("==> RangerKafkaAuditHandler.isAuditingNeeded()");
+        }
         boolean ret = true;
         boolean                            isAllowed = result.getIsAllowed();
         RangerAccessRequest request = result.getAccessRequest();
@@ -57,18 +102,31 @@ public class RangerKafkaAuditHandler extends 
RangerDefaultAuditHandler {
                 ret = false;
             }
         }
+        if(LOG.isDebugEnabled()) {
+            LOG.debug("RangerKafkaAuditHandler: isAuditingNeeded()");
+            LOG.debug("request:"+request);
+            LOG.debug("resource:"+resource);
+            LOG.debug("resourceName:"+resourceName);
+            LOG.debug("request.getAccessType():"+request.getAccessType());
+            LOG.debug("isAllowed:"+isAllowed);
+            LOG.debug("ret="+ret);
+            LOG.debug("<== RangerKafkaAuditHandler.isAuditingNeeded() = 
"+ret+" for result="+result);
+        }
         return ret;
     }
 
     public void flushAudit() {
         if(LOG.isDebugEnabled()) {
-            LOG.info("==> RangerYarnAuditHandler.flushAudit(" + "AuditEvent: " 
+ auditEvent + ")");
+            LOG.debug("==> RangerKafkaAuditHandler.flushAudit(" + "AuditEvent: 
" + auditEvent +" list="+ auditEventList+ ")");
         }
         if (auditEvent != null) {
             super.logAuthzAudit(auditEvent);
         }
+        else if (auditEventList.size()>0){
+            super.logAuthzAudits(auditEventList);
+        }
         if(LOG.isDebugEnabled()) {
-            LOG.info("<== RangerYarnAuditHandler.flushAudit(" + "AuditEvent: " 
+ auditEvent + ")");
+            LOG.debug("<== RangerKafkaAuditHandler.flushAudit()");
         }
     }
 }