Merge pull request #88 from AikidoSec/add-stats

Store monitored sink stats

lib/agent/aikido_types/stats.go

@@ -10,6 +10,8 @@ type StatsDataType struct {
 	RequestsAborted int
 	Attacks         int
 	AttacksBlocked  int
+
+	MonitoredSinkStats map[string]MonitoredSinkStats
 }
 
 type RateLimitingConfig struct {

lib/agent/cloud/cloud.go

@@ -1,6 +1,7 @@
 package cloud
 
 import (
+	. "main/aikido_types"
 	"main/globals"
 	"main/utils"
 	"time"
@@ -19,6 +20,7 @@ func Init() {
 	utils.StartPollingRoutine(ConfigPollingRoutineChannel, ConfigPollingTicker, CheckConfigUpdatedAt)
 
 	globals.StatsData.StartedAt = utils.GetTime()
+	globals.StatsData.MonitoredSinkStats = make(map[string]MonitoredSinkStats)
 	globals.MiddlewareInstalled = 0
 }
 

lib/agent/cloud/event_heartbeat.go

@@ -61,7 +61,7 @@ func GetStatsAndClear() Stats {
 	defer globals.StatsData.StatsMutex.Unlock()
 
 	stats := Stats{
-		Sinks:     make(map[string]MonitoredSinkStats),
+		Sinks:     globals.StatsData.MonitoredSinkStats,
 		StartedAt: globals.StatsData.StartedAt,
 		EndedAt:   utils.GetTime(),
 		Requests: Requests{
@@ -79,6 +79,7 @@ func GetStatsAndClear() Stats {
 	globals.StatsData.RequestsAborted = 0
 	globals.StatsData.Attacks = 0
 	globals.StatsData.AttacksBlocked = 0
+	globals.StatsData.MonitoredSinkStats = make(map[string]MonitoredSinkStats)
 
 	return stats
 }

lib/agent/go.mod

@@ -6,7 +6,7 @@ toolchain go1.23.3
 
 require (
 	github.com/stretchr/testify v1.9.0
-	google.golang.org/grpc v1.69.0
+	google.golang.org/grpc v1.69.2
 	google.golang.org/protobuf v1.35.1
 )
 

lib/agent/go.sum

@@ -38,6 +38,8 @@ google.golang.org/grpc v1.68.1 h1:oI5oTa11+ng8r8XMMN7jAOmWfPZWbYpCFaMUTACxkM0=
 google.golang.org/grpc v1.68.1/go.mod h1:+q1XYFJjShcqn0QZHvCyeR4CXPA+llXIeUIfIe00waw=
 google.golang.org/grpc v1.69.0 h1:quSiOM1GJPmPH5XtU+BCoVXcDVJJAzNcoyfC2cCjGkI=
 google.golang.org/grpc v1.69.0/go.mod h1:vyjdE6jLBI76dgpDojsFGNaHlxdjXN9ghpnd2o7JGZ4=
+google.golang.org/grpc v1.69.2 h1:U3S9QEtbXC0bYNvRtcoklF3xGtLViumSYxWykJS+7AU=
+google.golang.org/grpc v1.69.2/go.mod h1:vyjdE6jLBI76dgpDojsFGNaHlxdjXN9ghpnd2o7JGZ4=
 google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
 google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
 google.golang.org/protobuf v1.35.1 h1:m3LfL6/Ca+fqnjnlqQXNpFPABW1UD7mjh8KO2mKFytA=

lib/agent/grpc/request.go

@@ -16,6 +16,44 @@ func storeStats() {
 	globals.StatsData.Requests += 1
 }
 
+func storeAttackStats(req *protos.AttackDetected) {
+	globals.StatsData.StatsMutex.Lock()
+	defer globals.StatsData.StatsMutex.Unlock()
+
+	globals.StatsData.Attacks += 1
+	if req.GetAttack().GetBlocked() {
+		globals.StatsData.AttacksBlocked += 1
+	}
+}
+
+func getCompressedTiming(protoCompressedTiming *protos.CompressedTiming) CompressedTiming {
+	return CompressedTiming{
+		AverageInMS:  protoCompressedTiming.GetAverageInMs(),
+		Percentiles:  protoCompressedTiming.GetPercentiles(),
+		CompressedAt: utils.GetTime(),
+	}
+}
+
+func storeSinkStats(protoSinkStats *protos.MonitoredSinkStats) {
+	globals.StatsData.StatsMutex.Lock()
+	defer globals.StatsData.StatsMutex.Unlock()
+
+	sink := protoSinkStats.GetSink()
+	monitoredSinkStats, found := globals.StatsData.MonitoredSinkStats[sink]
+	if !found {
+		monitoredSinkStats = MonitoredSinkStats{}
+	}
+
+	monitoredSinkStats.AttacksDetected.Total += int(protoSinkStats.GetAttacksDetected())
+	monitoredSinkStats.AttacksDetected.Blocked += int(protoSinkStats.GetAttacksBlocked())
+	monitoredSinkStats.InterceptorThrewError += int(protoSinkStats.GetInterceptorThrewError())
+	monitoredSinkStats.WithoutContext += int(protoSinkStats.GetWithoutContext())
+	monitoredSinkStats.Total += int(protoSinkStats.GetTotal())
+	monitoredSinkStats.CompressedTimings = append(monitoredSinkStats.CompressedTimings, getCompressedTiming(protoSinkStats.GetCompressedTiming()))
+
+	globals.StatsData.MonitoredSinkStats[sink] = monitoredSinkStats
+}
+
 func getApiSpecData(apiSpec *protos.APISpec) (*protos.DataSchema, string, *protos.DataSchema, []*protos.APIAuthType) {
 	if apiSpec == nil {
 		return nil, "", nil, nil

lib/agent/grpc/server.go

@@ -70,7 +70,13 @@ func (s *server) OnUser(ctx context.Context, req *protos.User) (*emptypb.Empty,
 }
 
 func (s *server) OnAttackDetected(ctx context.Context, req *protos.AttackDetected) (*emptypb.Empty, error) {
-	go cloud.SendAttackDetectedEvent(req)
+	cloud.SendAttackDetectedEvent(req)
+	storeAttackStats(req)
+	return &emptypb.Empty{}, nil
+}
+
+func (s *server) OnMonitoredSinkStats(ctx context.Context, req *protos.MonitoredSinkStats) (*emptypb.Empty, error) {
+	storeSinkStats(req)
 	return &emptypb.Empty{}, nil
 }
 

lib/ipc.proto

@@ -14,6 +14,7 @@ service Aikido {
   rpc GetCloudConfig(CloudConfigUpdatedAt) returns (CloudConfig);
   rpc OnUser(User) returns (google.protobuf.Empty);
   rpc OnAttackDetected(AttackDetected) returns (google.protobuf.Empty);
+  rpc OnMonitoredSinkStats(MonitoredSinkStats) returns (google.protobuf.Empty);
   rpc OnMiddlewareInstalled(google.protobuf.Empty) returns (google.protobuf.Empty);
 }
 
@@ -46,6 +47,21 @@ message RequestMetadataShutdown {
   APISpec apiSpec = 6;
 }
 
+message CompressedTiming {
+  double averageInMs = 1;
+  map<string, double> percentiles = 2;
+}
+
+message MonitoredSinkStats {
+  string sink = 1;
+  int32 attacksDetected = 2;
+  int32 attacksBlocked = 3;
+  int32 interceptorThrewError = 4;
+  int32 withoutContext = 5;
+  int32 total = 6;
+  CompressedTiming compressedTiming = 7;
+}
+
 message RateLimiting {
   bool enabled = 1;
 }

lib/php-extension/Action.cpp

@@ -54,6 +54,10 @@ ACTION_STATUS Action::Execute(std::string &event) {
     return CONTINUE;
 }
 
+bool Action::IsDetection(std::string &event) {
+    return !event.empty();
+}
+
 void Action::Reset() {
     exit = false;
     block = false;

lib/php-extension/Aikido.cpp

@@ -53,6 +53,8 @@ PHP_MSHUTDOWN_FUNCTION(aikido) {
 }
 
 PHP_RINIT_FUNCTION(aikido) {
+    ScopedTimer scopedTimer("request_init");
+
     AIKIDO_LOG_DEBUG("RINIT started!\n");
 
     if (AIKIDO_GLOBAL(disable) == true) {
@@ -66,6 +68,8 @@ PHP_RINIT_FUNCTION(aikido) {
 }
 
 PHP_RSHUTDOWN_FUNCTION(aikido) {
+    ScopedTimer scopedTimer("request_shutdown");
+
     AIKIDO_LOG_DEBUG("RSHUTDOWN started!\n");
 
     if (AIKIDO_GLOBAL(disable) == true) {

lib/php-extension/Environment.cpp

@@ -1,5 +1,9 @@
 #include "Includes.h"
 
+// Minimum value to use for reporting once every X requests the collected stats to Agent
+// As the report_stats_interval_to_agent is configurable, this define is used to ensure that the configured interval is NEVER less that 50 requests
+#define MIN_REPORT_STATS_INTERVAL_TO_AGENT 50
+
 std::string GetLaravelEnvVariable(const std::string& env_key) {
     zval env_value;
     if (!CallPhpFunctionWithOneParam("getenv", env_key, &env_value) || Z_TYPE(env_value) != IS_STRING) {
@@ -44,6 +48,20 @@ bool GetEnvBool(const std::string& env_key, bool default_value) {
     return default_value;
 }
 
+unsigned int GetEnvNumber(const std::string& env_key, unsigned int default_value) {
+    std::string env_value = GetEnvVariable(env_key.c_str());
+    if (!env_value.empty()) {
+        try {
+            unsigned int number = std::stoi(env_value);
+            if (number <= MIN_REPORT_STATS_INTERVAL_TO_AGENT) {
+                return MIN_REPORT_STATS_INTERVAL_TO_AGENT;
+            }
+        }
+        catch (...) {}
+    }
+    return default_value;
+}
+
 void LoadEnvironment() {
     if (GetEnvBool("AIKIDO_DEBUG", false)) {
         AIKIDO_GLOBAL(log_level_str) = "DEBUG";
@@ -62,4 +80,5 @@ void LoadEnvironment() {
     AIKIDO_GLOBAL(token) = GetEnvString("AIKIDO_TOKEN", "");
     AIKIDO_GLOBAL(endpoint) = GetEnvString("AIKIDO_ENDPOINT", "https://guard.aikido.dev/");
     AIKIDO_GLOBAL(config_endpoint) = GetEnvString("AIKIDO_REALTIME_ENDPOINT", "https://runtime.aikido.dev/");
+    AIKIDO_GLOBAL(report_stats_interval_to_agent) = GetEnvNumber("AIKIDO_REPORT_STATS_INTERVAL", 10000);
 }

lib/php-extension/GoWrappers.cpp

@@ -1,9 +1,12 @@
 #include "Includes.h"
 
-GoString GoCreateString(std::string& s) {
+GoString GoCreateString(const std::string& s) {
     return GoString{s.c_str(), s.length()};
 }
 
+GoSlice GoCreateSlice(const std::vector<int64_t>& v) {
+    return GoSlice{ (void*)v.data(), v.size(), v.capacity() };
+}
 /*
     Callback wrapper called by the RequestProcessor (GO) whenever it needs data from PHP (C++ extension).
 */

lib/php-extension/Handle.cpp

@@ -1,11 +1,15 @@
 #include "Includes.h"
+#include "include/Stats.h"
 
 ZEND_NAMED_FUNCTION(aikido_generic_handler) {
+    ScopedTimer scopedTimer;
+
     AIKIDO_LOG_DEBUG("Aikido generic handler started!\n");
 
     zif_handler original_handler = nullptr;
     aikido_handler post_handler = nullptr;
 
+    std::string sink;
     std::string outputEvent;
     bool caughtException = false;
 
@@ -53,6 +57,9 @@ ZEND_NAMED_FUNCTION(aikido_generic_handler) {
             return;
         }
 
+        sink = scope_name;
+        scopedTimer.SetSink(sink);
+
         AIKIDO_LOG_DEBUG("Calling handler for \"%s\"!\n", scope_name.c_str());
 
         EVENT_ID eventId = NO_EVENT_ID;
@@ -68,10 +75,14 @@ ZEND_NAMED_FUNCTION(aikido_generic_handler) {
         if (eventId != NO_EVENT_ID) {
             std::string outputEvent;
             requestProcessor.SendEvent(eventId, outputEvent);
-            if (requestProcessor.IsBlockingEnabled() && action.Execute(outputEvent) == BLOCK) {
-                // exit generic handler and do not call the original handler
-                // thus blocking the execution
-                return;
+            if (action.IsDetection(outputEvent)) {
+                stats[sink].IncrementAttacksDetected();
+                if (requestProcessor.IsBlockingEnabled() && action.Execute(outputEvent) == BLOCK) {
+                    stats[sink].IncrementAttacksBlocked();
+                    // exit generic handler and do not call the original handler
+                    // thus blocking the execution
+                    return;
+                }
             }
         }
     } catch (const std::exception& e) {
@@ -80,7 +91,9 @@ ZEND_NAMED_FUNCTION(aikido_generic_handler) {
     }
 
     if (original_handler) {
+        scopedTimer.Stop();
         original_handler(INTERNAL_FUNCTION_PARAM_PASSTHRU);
+        scopedTimer.Start();
 
         if (!caughtException && post_handler) {
             EVENT_ID eventId = NO_EVENT_ID;
@@ -95,9 +108,12 @@ ZEND_NAMED_FUNCTION(aikido_generic_handler) {
             if (eventId != NO_EVENT_ID) {
                 std::string output;
                 requestProcessor.SendEvent(eventId, output);
-                if (requestProcessor.IsBlockingEnabled()) {
-                    action.Execute(output);
-                }
+                if (action.IsDetection(output)) {
+                    stats[sink].IncrementAttacksDetected();
+                    if (requestProcessor.IsBlockingEnabled() && action.Execute(output) == BLOCK) {
+                        stats[sink].IncrementAttacksBlocked();
+                    }
+                }   
             }
         }
     }

lib/php-extension/HandleShouldBlockRequest.cpp

@@ -15,6 +15,8 @@ bool GetBlockingStatus() {
 }
 
 ZEND_FUNCTION(should_block_request) {
+    ScopedTimer scopedTimer("should_block_request");
+
     if (AIKIDO_GLOBAL(disable) == true) {
         return;
     }

lib/php-extension/HandleUsers.cpp

@@ -19,6 +19,8 @@ bool SendUserEvent(std::string id, std::string username) {
 // Receives two parameters: "id" (string) and "name" (string, optional).
 // Returns true if the setting of the user succeeded, false otherwise.
 ZEND_FUNCTION(set_user) {
+    ScopedTimer scopedTimer("set_user");
+
     if (AIKIDO_GLOBAL(disable) == true) {
         RETURN_BOOL(false);
     }

lib/php-extension/RequestProcessor.cpp

@@ -77,6 +77,17 @@ bool RequestProcessor::IsBlockingEnabled() {
     return ret;
 }
 
+bool RequestProcessor::ReportStats() {
+    AIKIDO_LOG_INFO("Reporting stats to Aikido Request Processor...\n");
+
+    for (const auto& [sink, sinkStats] : stats) {
+        AIKIDO_LOG_INFO("Reporting stats for sink \"%s\" to Aikido Request Processor...\n", sink.c_str());
+        requestProcessorReportStatsFn(GoCreateString(sink), sinkStats.attacksDetected, sinkStats.attacksBlocked, sinkStats.interceptorThrewError, sinkStats.withoutContext, sinkStats.timings.size(), GoCreateSlice(sinkStats.timings));
+    }
+    stats.clear();
+    return true;
+}
+
 bool RequestProcessor::Init() {
     if (this->initFailed) {
         return false;
@@ -101,12 +112,14 @@ bool RequestProcessor::Init() {
     this->requestProcessorConfigUpdateFn = (RequestProcessorConfigUpdateFn)dlsym(libHandle, "RequestProcessorConfigUpdate");
     this->requestProcessorOnEventFn = (RequestProcessorOnEventFn)dlsym(libHandle, "RequestProcessorOnEvent");
     this->requestProcessorGetBlockingModeFn = (RequestProcessorGetBlockingModeFn)dlsym(libHandle, "RequestProcessorGetBlockingMode");
+    this->requestProcessorReportStatsFn = (RequestProcessorReportStats)dlsym(libHandle, "RequestProcessorReportStats");
     this->requestProcessorUninitFn = (RequestProcessorUninitFn)dlsym(libHandle, "RequestProcessorUninit");
     if (!requestProcessorInitFn ||
         !this->requestProcessorContextInitFn ||
         !this->requestProcessorConfigUpdateFn ||
         !this->requestProcessorOnEventFn ||
         !this->requestProcessorGetBlockingModeFn ||
+        !this->requestProcessorReportStatsFn ||
         !this->requestProcessorUninitFn) {
         AIKIDO_LOG_ERROR("Error loading symbols from the Aikido Request Processor library!\n");
         this->initFailed = true;
@@ -135,10 +148,15 @@ bool RequestProcessor::RequestInit() {
         return false;
     }
 
-    requestInitialized = true;
+    this->requestInitialized = true;
+    this->numberOfRequests++;
 
     ContextInit();
     SendPreRequestEvent();
+
+    if ((this->numberOfRequests % AIKIDO_GLOBAL(report_stats_interval_to_agent)) == 0) {
+        requestProcessor.ReportStats();
+    }
     return true;
 }
 
@@ -161,14 +179,17 @@ void RequestProcessor::RequestShutdown() {
 
     LoadConfigOnce();
     SendPostRequestEvent();
-    requestInitialized = false;
+    this->requestInitialized = false;
 }
 
 void RequestProcessor::Uninit() {
     if (!this->libHandle) {
         return;
     }
     if (!this->initFailed && this->requestProcessorUninitFn) {
+        AIKIDO_LOG_INFO("Reporting final stats to Aikido Request Processor...\n");
+        this->ReportStats();
+
         AIKIDO_LOG_INFO("Calling uninit for Aikido Request Processor...\n");
         this->requestProcessorUninitFn();
     }