Dp auditlog CDAP-20852 Milestone2

cdap-app-fabric/src/main/java/io/cdap/cdap/app/guice/DistributedProgramContainerModule.java

@@ -23,6 +23,7 @@
 import com.google.inject.Provider;
 import com.google.inject.Scopes;
 import com.google.inject.util.Modules;
+import io.cdap.cdap.api.auditlogging.AuditLogPublisherService;
 import io.cdap.cdap.app.runtime.Arguments;
 import io.cdap.cdap.app.runtime.ProgramOptions;
 import io.cdap.cdap.app.runtime.ProgramStateWriter;
@@ -73,6 +74,7 @@
 import io.cdap.cdap.proto.id.ProgramRunId;
 import io.cdap.cdap.runtime.spi.RuntimeMonitorType;
 import io.cdap.cdap.security.auth.context.AuthenticationContextModules;
+import io.cdap.cdap.security.auth.service.DefaultAuditLogPublisherService;
 import io.cdap.cdap.security.authorization.AuthorizationEnforcementModule;
 import io.cdap.cdap.security.guice.CoreSecurityModule;
 import io.cdap.cdap.security.guice.CoreSecurityRuntimeModule;
@@ -193,6 +195,7 @@ protected void configure() {
 
         bind(PreferencesFetcher.class).to(RemotePreferencesFetcherInternal.class)
             .in(Scopes.SINGLETON);
+        bind(AuditLogPublisherService.class).to(DefaultAuditLogPublisherService.class);
       }
     });
 

cdap-app-fabric/src/main/java/io/cdap/cdap/app/preview/DefaultPreviewRunnerManager.java

@@ -27,6 +27,7 @@
 import com.google.inject.Scopes;
 import com.google.inject.name.Named;
 import com.google.inject.util.Modules;
+import io.cdap.cdap.api.auditlogging.AuditLogPublisherService;
 import io.cdap.cdap.api.security.store.SecureStore;
 import io.cdap.cdap.app.guice.ProgramRunnerRuntimeModule;
 import io.cdap.cdap.common.NotFoundException;
@@ -55,6 +56,7 @@
 import io.cdap.cdap.metrics.guice.MetricsClientRuntimeModule;
 import io.cdap.cdap.proto.id.ApplicationId;
 import io.cdap.cdap.security.auth.context.AuthenticationContextModules;
+import io.cdap.cdap.security.auth.service.DefaultAuditLogPublisherService;
 import io.cdap.cdap.security.guice.CoreSecurityRuntimeModule;
 import io.cdap.cdap.security.guice.preview.PreviewSecureStoreModule;
 import java.net.InetAddress;
@@ -216,6 +218,12 @@ public InetAddress providesHostname(CConfiguration cConf) {
             String address = cConf.get(Constants.Preview.ADDRESS);
             return Networks.resolve(address, new InetSocketAddress("localhost", 0).getAddress());
           }
+        },
+        new AbstractModule() {
+          @Override
+          protected void configure() {
+            bind(AuditLogPublisherService.class).to(DefaultAuditLogPublisherService.class);
+          }
         }
     );
   }

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/preview/DefaultPreviewManager.java

@@ -31,6 +31,7 @@
 import com.google.inject.name.Names;
 import com.google.inject.util.Modules;
 import io.cdap.cdap.api.annotation.Name;
+import io.cdap.cdap.api.auditlogging.AuditLogPublisherService;
 import io.cdap.cdap.api.metrics.MetricsCollectionService;
 import io.cdap.cdap.api.security.AccessException;
 import io.cdap.cdap.app.preview.PreviewConfigModule;
@@ -91,6 +92,7 @@
 import io.cdap.cdap.proto.id.ProgramRunId;
 import io.cdap.cdap.proto.security.ApplicationPermission;
 import io.cdap.cdap.security.auth.context.AuthenticationContextModules;
+import io.cdap.cdap.security.auth.service.DefaultAuditLogPublisherService;
 import io.cdap.cdap.security.authorization.AccessControllerInstantiator;
 import io.cdap.cdap.security.authorization.DefaultContextAccessEnforcer;
 import io.cdap.cdap.security.guice.CoreSecurityRuntimeModule;
@@ -388,6 +390,7 @@ protected void configure() {
             bind(LevelDBTableService.class).toInstance(previewLevelDBTableService);
             bind(RemoteExecutionLogProcessor.class).to(LogAppenderLogProcessor.class)
                 .in(Scopes.SINGLETON);
+            bind(AuditLogPublisherService.class).to(DefaultAuditLogPublisherService.class);
           }
 
           @Provides

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/runtime/distributed/AbstractProgramTwillRunnable.java

@@ -30,6 +30,7 @@
 import com.google.inject.Key;
 import com.google.inject.Module;
 import io.cdap.cdap.api.app.ApplicationSpecification;
+import io.cdap.cdap.api.auditlogging.AuditLogPublisherService;
 import io.cdap.cdap.api.metrics.MetricsCollectionService;
 import io.cdap.cdap.app.guice.ClusterMode;
 import io.cdap.cdap.app.guice.DistributedProgramContainerModule;
@@ -454,6 +455,10 @@ private Deque<Service> createCoreServices(Injector injector, ProgramOptions prog
         MetricsCollectionService.class);
     services.add(metricsCollectionService);
 
+    AuditLogPublisherService auditLogPublisherService = injector.getInstance(
+      AuditLogPublisherService.class);
+    services.add(auditLogPublisherService);
+
     if (ProgramRunners.getClusterMode(programOptions) != ClusterMode.ON_PREMISE) {
       services.add(injector.getInstance(LogAppenderLoaderService.class));
     }
@@ -483,6 +488,8 @@ private void addOnPremiseServices(Injector injector, ProgramOptions programOptio
   private void startCoreServices() {
     // Starts the core services
     for (Service service : coreServices) {
+      LOG.warn("SANKET_LOG_1 : STARTING SERVICE : " + service.toString());
+      LOG.warn("SANKET_LOG_1 : STARTING SERVICE : " + service.getClass());
       service.startAndWait();
     }
   }

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/worker/TaskWorkerTwillRunnable.java

@@ -25,6 +25,7 @@
 import com.google.inject.Guice;
 import com.google.inject.Injector;
 import com.google.inject.Module;
+import io.cdap.cdap.api.auditlogging.AuditLogPublisherService;
 import io.cdap.cdap.api.metrics.MetricsCollectionService;
 import io.cdap.cdap.common.conf.CConfiguration;
 import io.cdap.cdap.common.conf.Constants;
@@ -48,6 +49,7 @@
 import io.cdap.cdap.metrics.guice.MetricsClientRuntimeModule;
 import io.cdap.cdap.proto.id.NamespaceId;
 import io.cdap.cdap.security.auth.context.AuthenticationContextModules;
+import io.cdap.cdap.security.auth.service.DefaultAuditLogPublisherService;
 import io.cdap.cdap.security.guice.CoreSecurityModule;
 import io.cdap.cdap.security.guice.CoreSecurityRuntimeModule;
 import java.io.File;
@@ -118,6 +120,14 @@ protected void configure() {
         }
       });
       modules.add(new RemoteLogAppenderModule());
+      modules.add(
+        new AbstractModule() {
+          @Override
+          protected void configure() {
+            bind(AuditLogPublisherService.class).to(DefaultAuditLogPublisherService.class);
+          }
+        }
+      );
 
       if (coreSecurityModule.requiresZKClient()) {
         modules.add(new ZkClientModule());

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/worker/sidecar/ArtifactLocalizerTwillRunnable.java

@@ -25,6 +25,7 @@
 import com.google.inject.Guice;
 import com.google.inject.Injector;
 import com.google.inject.Module;
+import io.cdap.cdap.api.auditlogging.AuditLogPublisherService;
 import io.cdap.cdap.api.feature.FeatureFlagsProvider;
 import io.cdap.cdap.app.guice.DistributedArtifactManagerModule;
 import io.cdap.cdap.common.conf.CConfiguration;
@@ -53,6 +54,7 @@
 import io.cdap.cdap.proto.id.NamespaceId;
 import io.cdap.cdap.security.auth.TokenManager;
 import io.cdap.cdap.security.auth.context.AuthenticationContextModules;
+import io.cdap.cdap.security.auth.service.DefaultAuditLogPublisherService;
 import io.cdap.cdap.security.guice.CoreSecurityModule;
 import io.cdap.cdap.security.guice.CoreSecurityRuntimeModule;
 import java.io.File;
@@ -137,6 +139,12 @@ protected void configure() {
       });
       modules.add(new RemoteLogAppenderModule());
       modules.add(new LocalLocationModule());
+      modules.add(new AbstractModule() {
+        @Override
+        protected void configure() {
+          bind(AuditLogPublisherService.class).to(DefaultAuditLogPublisherService.class);
+        }
+      });
 
       if (coreSecurityModule.requiresZKClient()) {
         modules.add(new ZkClientModule());

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/worker/system/SystemWorkerTwillRunnable.java

@@ -31,6 +31,7 @@
 import com.google.inject.multibindings.OptionalBinder;
 import com.google.inject.util.Modules;
 import io.cdap.cdap.api.artifact.ArtifactManager;
+import io.cdap.cdap.api.auditlogging.AuditLogPublisherService;
 import io.cdap.cdap.api.metrics.MetricsCollectionService;
 import io.cdap.cdap.app.guice.AppFabricServiceRuntimeModule;
 import io.cdap.cdap.app.guice.AuthorizationModule;
@@ -83,6 +84,7 @@
 import io.cdap.cdap.proto.id.NamespaceId;
 import io.cdap.cdap.security.auth.KeyManager;
 import io.cdap.cdap.security.auth.context.AuthenticationContextModules;
+import io.cdap.cdap.security.auth.service.DefaultAuditLogPublisherService;
 import io.cdap.cdap.security.authorization.AuthorizationEnforcementModule;
 import io.cdap.cdap.security.guice.CoreSecurityModule;
 import io.cdap.cdap.security.guice.FileBasedCoreSecurityModule;
@@ -190,6 +192,12 @@ protected void configure() {
             bind(MetadataPublisher.class).to(MessagingMetadataPublisher.class);
             bind(MetadataServiceClient.class).to(DefaultMetadataServiceClient.class);
           }
+        },
+        new AbstractModule() {
+          @Override
+          protected void configure() {
+            bind(AuditLogPublisherService.class).to(DefaultAuditLogPublisherService.class);
+          }
         }
     ));
 

cdap-app-fabric/src/main/java/io/cdap/cdap/master/environment/k8s/AbstractServiceMain.java

@@ -26,6 +26,7 @@
 import com.google.inject.Injector;
 import com.google.inject.Key;
 import com.google.inject.Module;
+import io.cdap.cdap.api.auditlogging.AuditLogPublisherService;
 import io.cdap.cdap.api.metrics.MetricsCollectionService;
 import io.cdap.cdap.app.preview.PreviewConfigModule;
 import io.cdap.cdap.common.app.MainClassLoader;
@@ -55,6 +56,7 @@
 import io.cdap.cdap.metrics.guice.MetricsClientRuntimeModule;
 import io.cdap.cdap.security.auth.TokenManager;
 import io.cdap.cdap.security.auth.context.AuthenticationContextModules;
+import io.cdap.cdap.security.auth.service.DefaultAuditLogPublisherService;
 import io.cdap.cdap.security.guice.CoreSecurityModule;
 import io.cdap.cdap.security.guice.CoreSecurityRuntimeModule;
 import io.cdap.cdap.security.impersonation.SecurityUtil;
@@ -178,6 +180,14 @@ protected void configure() {
                 new SupplierProviderBridge<>(masterEnv.getDiscoveryServiceClientSupplier()));
       }
     });
+
+    modules.add(new AbstractModule() {
+      @Override
+      protected void configure() {
+        bind(AuditLogPublisherService.class).to(DefaultAuditLogPublisherService.class);
+      }
+    });
+
     modules.add(getLogAppenderModule());
 
     CoreSecurityModule coreSecurityModule = CoreSecurityRuntimeModule.getDistributedModule(cConf);
@@ -202,6 +212,7 @@ protected void configure() {
 
     // Add Services
     services.add(injector.getInstance(MetricsCollectionService.class));
+    services.add(injector.getInstance(AuditLogPublisherService.class));
     if (SecurityUtil.isInternalAuthEnabled(cConf)) {
       services.add(injector.getInstance(TokenManager.class));
     }

cdap-common/src/main/java/io/cdap/cdap/api/auditlogging/AuditLogPublisherService.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright © 2024 Cask Data, Inc.
+ *
+ * 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.
+ */
+
+package io.cdap.cdap.api.auditlogging;
+
+
+import com.google.common.util.concurrent.Service;
+import io.cdap.cdap.security.spi.authorization.AuditLogContext;
+
+import java.util.Queue;
+
+
+/**
+ * Service to batch and publish audit log to external auth service.
+ */
+public interface AuditLogPublisherService extends Service {
+
+  /**
+   * pushes the log entry to respective external service
+   */
+  void publish();
+
+  /**
+   * add to service's pending list for publishing
+   */
+  void addAuditContexts(Queue<AuditLogContext> auditLogContextQueue);
+}
+

cdap-common/src/main/java/io/cdap/cdap/common/http/AuthenticationChannelHandler.java

@@ -16,13 +16,15 @@
 
 package io.cdap.cdap.common.http;
 
+import io.cdap.cdap.api.auditlogging.AuditLogPublisherService;
 import io.cdap.cdap.common.conf.Constants;
 import io.cdap.cdap.proto.security.Credential;
 import io.cdap.cdap.security.spi.authentication.SecurityRequestContext;
 import io.cdap.cdap.security.spi.authentication.UnauthenticatedException;
+import io.netty.channel.ChannelDuplexHandler;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
 import io.netty.handler.codec.http.DefaultFullHttpResponse;
 import io.netty.handler.codec.http.HttpRequest;
 import io.netty.handler.codec.http.HttpResponse;
@@ -32,11 +34,13 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.concurrent.LinkedBlockingDeque;
+
 /**
  * An UpstreamHandler that verifies the userId in a request header and updates the {@code
  * SecurityRequestContext}.
  */
-public class AuthenticationChannelHandler extends ChannelInboundHandlerAdapter {
+public class AuthenticationChannelHandler extends ChannelDuplexHandler {
 
   private static final Logger LOG = LoggerFactory.getLogger(AuthenticationChannelHandler.class);
 
@@ -47,9 +51,11 @@ public class AuthenticationChannelHandler extends ChannelInboundHandlerAdapter {
   private static final String EMPTY_USER_IP = "CDAP-empty-user-ip";
 
   private final boolean internalAuthEnabled;
+  private final AuditLogPublisherService auditLogPublisherService;
 
-  public AuthenticationChannelHandler(boolean internalAuthEnabled) {
+  public AuthenticationChannelHandler(boolean internalAuthEnabled, AuditLogPublisherService auditLogPublisherService) {
     this.internalAuthEnabled = internalAuthEnabled;
+    this.auditLogPublisherService = auditLogPublisherService;
   }
 
   /**
@@ -118,13 +124,33 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
       SecurityRequestContext.setUserIp(currentUserIp);
     }
 
+    LOG.warn("SANKET_LOG : read1");
     try {
       ctx.fireChannelRead(msg);
     } finally {
       SecurityRequestContext.reset();
     }
   }
 
+  @Override
+  public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
+
+    LOG.warn("SANKET_LOG : write1");
+    if (msg instanceof HttpRequest) {
+      LOG.warn("SANKET_LOG : HttpRequest");
+    }
+
+    if (msg instanceof HttpResponse) {
+      LOG.warn("SANKET_LOG : HttpResponse " + (HttpResponse) msg);
+    }
+
+
+    auditLogPublisherService.addAuditContexts(SecurityRequestContext.getAuditLogQueue());
+    auditLogPublisherService.publish();
+    super.write(ctx, msg, promise);
+    LOG.warn("SANKET_LOG : write2");
+  }
+
   @Override
   public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
     LOG.error("Got exception: {}", cause.getMessage(), cause);

cdap-common/src/main/java/io/cdap/cdap/common/http/CommonNettyHttpServiceBuilder.java

@@ -15,11 +15,13 @@
  */
 package io.cdap.cdap.common.http;
 
+import io.cdap.cdap.api.auditlogging.AuditLogPublisherService;
 import io.cdap.cdap.api.metrics.MetricsCollectionService;
 import io.cdap.cdap.common.HttpExceptionHandler;
 import io.cdap.cdap.common.conf.CConfiguration;
 import io.cdap.cdap.common.conf.Constants;
 import io.cdap.cdap.common.metrics.MetricsReporterHook;
+import io.cdap.cdap.common.metrics.NoOpMetricsCollectionService;
 import io.cdap.http.ChannelPipelineModifier;
 import io.cdap.http.NettyHttpService;
 import io.netty.channel.ChannelPipeline;
@@ -38,7 +40,7 @@ public class CommonNettyHttpServiceBuilder extends NettyHttpService.Builder {
   private ChannelPipelineModifier additionalModifier;
 
   public CommonNettyHttpServiceBuilder(CConfiguration cConf, String serviceName,
-      MetricsCollectionService metricsCollectionService) {
+      MetricsCollectionService metricsCollectionService, AuditLogPublisherService auditLogPublisherService) {
     super(serviceName);
     if (cConf.getBoolean(Constants.Security.ENABLED)) {
       pipelineModifier = new ChannelPipelineModifier() {
@@ -51,7 +53,7 @@ public void modify(ChannelPipeline pipeline) {
           EventExecutor executor = pipeline.context("dispatcher").executor();
           pipeline.addBefore(executor, "dispatcher", AUTHENTICATOR_NAME,
               new AuthenticationChannelHandler(cConf.getBoolean(Constants.Security
-                  .INTERNAL_AUTH_ENABLED)));
+                  .INTERNAL_AUTH_ENABLED), auditLogPublisherService));
         }
       };
     }
@@ -60,6 +62,12 @@ public void modify(ChannelPipeline pipeline) {
         new MetricsReporterHook(cConf, metricsCollectionService, serviceName)));
   }
 
+  //TODO : Remove , this is for compiling test classes
+  public CommonNettyHttpServiceBuilder(CConfiguration cConf, String serviceName,
+                                       MetricsCollectionService metricsCollectionService) {
+    this(cConf, serviceName, metricsCollectionService, null);
+  }
+
   /**
    * Sets pipeline modifier, preserving the security one installed in constructor.
    */