Merge remote-tracking branch 'origin/release/v11' into release/v11

Author: mjabascal10

plugins/crowdStrike/README.md

@@ -0,0 +1,39 @@
+# UTMStack Plugin for CrowdStrike Falcon
+
+## Description
+
+UTMStack Plugin for CrowdStrike Falcon is a connector developed in Golang that receives real-time events from `CrowdStrike Falcon Event Streams` and sends them to the `UTMStack` processing server for further processing.
+
+This connector uses a `GRPC` client to communicate with the UTMStack processing server. The client connects through a `Unix socket` that is created in the UTMStack working directory. 
+
+To obtain the events, `CrowdStrike GoFalcon SDK` is used to communicate with the Falcon Event Streams API, providing real-time security event data from your CrowdStrike environment.
+
+Please note that the connector requires valid CrowdStrike Falcon API credentials to run. The connector will not work without proper authentication.
+
+## Configuration
+
+The plugin requires the following configuration parameters:
+
+- **client_id**: OAuth2 Client ID for CrowdStrike Falcon API
+- **client_secret**: OAuth2 Client Secret for CrowdStrike Falcon API  
+- **member_cid**: (Optional) Member CID for MSSP environments
+- **cloud**: Falcon cloud region (us-1, us-2, eu-1, us-gov-1)
+
+## Features
+
+- Real-time event streaming from CrowdStrike Falcon
+- Automatic stream discovery and processing
+- Error handling and retry mechanisms
+- Event batching to optimize performance
+- Timeout controls to prevent blocking
+- Structured JSON event formatting
+
+## Authentication
+
+This plugin uses OAuth2 authentication with CrowdStrike Falcon API. You need to:
+
+1. Create API client credentials in your CrowdStrike console
+2. Ensure the client has the required scopes for Event Streams API
+3. Configure the credentials in UTMStack module configuration
+
+For more information on creating API credentials, visit: https://falcon.crowdstrike.com/support/api-clients-and-keys

plugins/crowdStrike/check.go

@@ -0,0 +1,76 @@
+package main
+
+import (
+	"fmt"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/threatwinds/go-sdk/catcher"
+)
+
+func connectionChecker(url string) error {
+	checkConn := func() error {
+		if err := checkConnection(url); err != nil {
+			return fmt.Errorf("connection failed: %v", err)
+		}
+		return nil
+	}
+
+	if err := infiniteRetryIfXError(checkConn, "connection failed"); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func checkConnection(url string) error {
+	client := &http.Client{
+		Timeout: 30 * time.Second,
+	}
+
+	req, err := http.NewRequest(http.MethodGet, url, nil)
+	if err != nil {
+		return err
+	}
+
+	resp, err := client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		err := resp.Body.Close()
+		if err != nil {
+			_ = catcher.Error("error closing response body: %v", err, nil)
+		}
+	}()
+
+	return nil
+}
+
+func infiniteRetryIfXError(f func() error, exception string) error {
+	var xErrorWasLogged bool
+
+	for {
+		err := f()
+		if err != nil && is(err, exception) {
+			if !xErrorWasLogged {
+				_ = catcher.Error("An error occurred (%s), will keep retrying indefinitely...", err, nil)
+				xErrorWasLogged = true
+			}
+			time.Sleep(wait)
+			continue
+		}
+
+		return err
+	}
+}
+
+func is(e error, args ...string) bool {
+	for _, arg := range args {
+		if strings.Contains(e.Error(), arg) {
+			return true
+		}
+	}
+	return false
+}

plugins/crowdStrike/config/config.go

@@ -0,0 +1,141 @@
+package config
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"strings"
+	sync "sync"
+	"time"
+
+	"github.com/threatwinds/go-sdk/catcher"
+	"github.com/threatwinds/go-sdk/plugins"
+	"google.golang.org/grpc"
+	codes "google.golang.org/grpc/codes"
+	"google.golang.org/grpc/connectivity"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+)
+
+const (
+	reconnectDelay = 5 * time.Second
+	maxMessageSize = 1024 * 1024 * 1024
+)
+
+var (
+	cnf *ConfigurationSection
+	mu  sync.Mutex
+
+	internalKey       string
+	modulesConfigHost string
+)
+
+func GetConfig() *ConfigurationSection {
+	mu.Lock()
+	defer mu.Unlock()
+	if cnf == nil {
+		return &ConfigurationSection{}
+	}
+	return cnf
+}
+
+func StartConfigurationSystem() {
+	for {
+		pluginConfig := plugins.PluginCfg("com.utmstack", false)
+		if !pluginConfig.Exists() {
+			_ = catcher.Error("plugin configuration not found", nil, nil)
+			time.Sleep(reconnectDelay)
+			continue
+		}
+		internalKey = pluginConfig.Get("internalKey").String()
+		modulesConfigHost = pluginConfig.Get("modulesConfig").String()
+
+		if internalKey == "" || modulesConfigHost == "" {
+			fmt.Println("Internal key or Modules Config Host is not set, skipping UTMStack plugin execution")
+			time.Sleep(reconnectDelay)
+			continue
+		}
+		break
+	}
+
+	for {
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		ctx = metadata.AppendToOutgoingContext(ctx, "internal-key", internalKey)
+		conn, err := grpc.NewClient(
+			modulesConfigHost,
+			grpc.WithTransportCredentials(insecure.NewCredentials()),
+			grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMessageSize)),
+		)
+
+		if err != nil {
+			catcher.Error("Failed to connect to server", err, nil)
+			cancel()
+			time.Sleep(reconnectDelay)
+			continue
+		}
+
+		state := conn.GetState()
+		if state == connectivity.Shutdown || state == connectivity.TransientFailure {
+			catcher.Error("Connection is in shutdown or transient failure state", nil, nil)
+			cancel()
+			time.Sleep(reconnectDelay)
+			continue
+		}
+
+		client := NewConfigServiceClient(conn)
+		stream, err := client.StreamConfig(ctx)
+		if err != nil {
+			catcher.Error("Failed to create stream", err, nil)
+			conn.Close()
+			cancel()
+			time.Sleep(reconnectDelay)
+			continue
+		}
+
+		err = stream.Send(&BiDirectionalMessage{
+			Payload: &BiDirectionalMessage_PluginInit{
+				PluginInit: &PluginInit{Type: PluginType_CROWDSTRIKE},
+			},
+		})
+		if err != nil {
+			catcher.Error("Failed to send PluginInit", err, nil)
+			conn.Close()
+			cancel()
+			time.Sleep(reconnectDelay)
+			continue
+		}
+
+		for {
+			in, err := stream.Recv()
+			if err != nil {
+				if strings.Contains(err.Error(), "EOF") {
+					catcher.Info("Stream closed by server, reconnecting...", nil)
+					conn.Close()
+					cancel()
+					time.Sleep(reconnectDelay)
+					break
+				}
+				st, ok := status.FromError(err)
+				if ok && (st.Code() == codes.Unavailable || st.Code() == codes.Canceled) {
+					catcher.Error("Stream error: "+st.Message(), err, nil)
+					conn.Close()
+					cancel()
+					time.Sleep(reconnectDelay)
+					break
+				} else {
+					catcher.Error("Stream receive error", err, nil)
+					time.Sleep(reconnectDelay)
+					continue
+				}
+			}
+
+			switch message := in.Payload.(type) {
+			case *BiDirectionalMessage_Config:
+				log.Printf("Received configuration update: %v", message.Config)
+				cnf = message.Config
+			}
+		}
+	}
+}