Yandex S3 output

README.md

@@ -56,6 +56,7 @@ It works as a single endpoint for as many as you want `Falco` instances :
 
 - [**AWS S3**](https://aws.amazon.com/s3/features/)
 - [**GCP Storage**](https://cloud.google.com/storage)
+- [**Yandex S3 Storage**](https://cloud.yandex.com/en-ru/services/storage)
 
 ### FaaS / Serverless
 
@@ -406,6 +407,15 @@ grafana:
 
 webui:
   url: "" # WebUI URL, if not empty, WebUI output is enabled
+
+yandex:
+  aws:
+  # accesskeyid: "" # aws access key (optional if you use EC2 Instance Profile)
+  # secretaccesskey: "" # aws secret access key (optional if you use EC2 Instance Profile)
+  s3:
+    # bucket: "falcosidekick" # AWS S3, bucket name
+    # prefix : "" # name of prefix, keys will have format: s3://<bucket>/<prefix>/YYYY-MM-DD/YYYY-MM-DDTHH:mm:ss.s+01:00.json
+    # minimumpriority: "" # minimum priority of event for using this output, order is emergency|alert|critical|erro
 ```
 
 Usage :

config.go

@@ -266,6 +266,14 @@ func getConfig() *types.Configuration {
 	v.SetDefault("Grafana.MutualTls", false)
 	v.SetDefault("Grafana.CheckCert", true)
 
+	v.SetDefault("Yandex.AccessKeyID", "")
+	v.SetDefault("Yandex.SecretAccessKey", "")
+	v.SetDefault("Yandex.Endpoint", "https://storage.yandexcloud.net")
+	v.SetDefault("Yandex.Region", "ru-central1")
+	v.SetDefault("Yandex.S3.Bucket", "")
+	v.SetDefault("Yandex.S3.Prefix", "falco")
+	v.SetDefault("Yamdex.S3.MinimumPriority", "")
+
 	v.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
 	v.AutomaticEnv()
 	if *configFile != "" {
@@ -358,6 +366,7 @@ func getConfig() *types.Configuration {
 	c.Fission.MinimumPriority = checkPriority(c.Fission.MinimumPriority)
 	c.Rabbitmq.MinimumPriority = checkPriority(c.Rabbitmq.MinimumPriority)
 	c.Wavefront.MinimumPriority = checkPriority(c.Wavefront.MinimumPriority)
+	c.Yandex.S3.MinimumPriority = checkPriority(c.Wavefront.MinimumPriority)
 
 	c.Slack.MessageFormatTemplate = getMessageFormatTemplate("Slack", c.Slack.MessageFormat)
 	c.Rocketchat.MessageFormatTemplate = getMessageFormatTemplate("Rocketchat", c.Rocketchat.MessageFormat)

handlers.go

@@ -260,4 +260,8 @@ func forwardEvent(falcopayload types.FalcoPayload) {
 	if config.Fission.Function != "" && (falcopayload.Priority >= types.Priority(config.Fission.MinimumPriority) || falcopayload.Rule == testRule) {
 		go fissionClient.FissionCall(falcopayload)
 	}
+
+	if config.Yandex.S3.Bucket != "" && (falcopayload.Priority >= types.Priority(config.Yandex.S3.MinimumPriority) || falcopayload.Rule == testRule) {
+		go yandexS3Client.UploadYandexS3(falcopayload)
+	}
 }

main.go

@@ -47,6 +47,7 @@ var (
 	wavefrontClient     *outputs.Client
 	fissionClient       *outputs.Client
 	grafanaClient       *outputs.Client
+	yandexS3Client      *outputs.Client
 
 	statsdClient, dogstatsdClient *statsd.Client
 	config                        *types.Configuration
@@ -450,6 +451,20 @@ func init() {
 		}
 		log.Printf("[INFO]  : Enabled Outputs : %s\n", outputs.EnabledOutputs)
 	}
+
+	if config.Yandex.S3.Bucket != "" {
+		var err error
+		yandexS3Client, err = outputs.NewYandexS3Client(config, stats, promStats, statsdClient, dogstatsdClient)
+		if err != nil {
+			log.Printf("[ERROR] : Yandex - %v\n", err)
+		} else {
+			if config.Yandex.S3.Bucket != "" {
+				outputs.EnabledOutputs = append(outputs.EnabledOutputs, "YandexS3Client")
+			}
+		}
+		log.Printf("[INFO]  : Enabled Outputs : %s\n", outputs.EnabledOutputs)
+	}
+
 }
 
 func main() {

outputs/yandex.go

@@ -0,0 +1,77 @@
+package outputs
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log"
+	"os"
+	"time"
+
+	"github.com/DataDog/datadog-go/statsd"
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/s3"
+
+	"github.com/falcosecurity/falcosidekick/types"
+)
+
+// NewAWSClient returns a new output.Client for accessing the AWS API.
+func NewYandexS3Client(config *types.Configuration, stats *types.Statistics, promStats *types.PromStatistics, statsdClient, dogstatsdClient *statsd.Client) (*Client, error) {
+
+	if config.Yandex.AccessKeyID != "" && config.Yandex.SecretAccessKey != "" {
+		err1 := os.Setenv("AWS_ACCESS_KEY_ID", config.Yandex.AccessKeyID)
+		err2 := os.Setenv("AWS_SECRET_ACCESS_KEY", config.Yandex.SecretAccessKey)
+		if err1 != nil || err2 != nil {
+			log.Printf("[ERROR] : AWS - Error setting AWS env vars")
+			return nil, errors.New("Error setting AWS env vars")
+		}
+	}
+	sess, err := session.NewSession(&aws.Config{
+		Region:   aws.String(config.Yandex.Region),
+		Endpoint: aws.String(config.Yandex.Endpoint)})
+	if err != nil {
+		log.Printf("[ERROR] : AWS - %v\n", "Error while creating AWS Session")
+		return nil, errors.New("Error while creating AWS Session")
+	}
+
+	log.Printf("[INFO] : Yandex S3 session has been configured successfully")
+
+	return &Client{
+		OutputType:      "YandexS3",
+		Config:          config,
+		AWSSession:      sess,
+		Stats:           stats,
+		PromStats:       promStats,
+		StatsdClient:    statsdClient,
+		DogstatsdClient: dogstatsdClient,
+	}, nil
+}
+
+// UploadS3 upload payload to S3
+func (c *Client) UploadYandexS3(falcopayload types.FalcoPayload) {
+	f, _ := json.Marshal(falcopayload)
+	prefix := ""
+	t := time.Now()
+	if c.Config.Yandex.S3.Prefix != "" {
+		prefix = c.Config.Yandex.S3.Prefix
+	}
+	key := fmt.Sprintf("%s/%s/%s.json", prefix, t.Format("2006-01-02"), t.Format(time.RFC3339Nano))
+	_, err := s3.New(c.AWSSession).PutObject(&s3.PutObjectInput{
+		Bucket: aws.String(c.Config.Yandex.S3.Bucket),
+		Key:    aws.String(key),
+		Body:   bytes.NewReader(f),
+	})
+	if err != nil {
+		go c.CountMetric("outputs", 1, []string{"output:awss3", "status:error"})
+		c.PromStats.Outputs.With(map[string]string{"destination": "awss3", "status": Error}).Inc()
+		log.Printf("[ERROR] : %v S3 - %v\n", c.OutputType, err.Error())
+		return
+	}
+
+	log.Printf("[INFO]  : %v S3 - Upload payload OK\n", c.OutputType)
+
+	go c.CountMetric("outputs", 1, []string{"output:awss3", "status:ok"})
+	c.PromStats.Outputs.With(map[string]string{"destination": "awss3", "status": "ok"}).Inc()
+}

stats.go

@@ -60,6 +60,7 @@ func getInitStats() *types.Statistics {
 		Wavefront:         getOutputNewMap("wavefront"),
 		Fission:           getOutputNewMap("fission"),
 		Grafana:           getOutputNewMap("grafana"),
+		YandexS3:          getOutputNewMap("yandexs3"),
 	}
 	stats.Falco.Add(outputs.Emergency, 0)
 	stats.Falco.Add(outputs.Alert, 0)

types/types.go

@@ -56,6 +56,7 @@ type Configuration struct {
 	Wavefront          WavefrontOutputConfig
 	Fission            fissionConfig
 	Grafana            grafanaOutputConfig
+	Yandex             YandexOutputConfig
 }
 
 // SlackOutputConfig represents parameters for Slack
@@ -386,6 +387,19 @@ type grafanaOutputConfig struct {
 	MinimumPriority string
 }
 
+type YandexOutputConfig struct {
+	AccessKeyID     string
+	SecretAccessKey string
+	Endpoint        string
+	Region          string
+	S3              YandexS3Config
+}
+type YandexS3Config struct {
+	Prefix          string
+	Bucket          string
+	MinimumPriority string
+}
+
 // Statistics is a struct to store stastics
 type Statistics struct {
 	Requests          *expvar.Map
@@ -430,6 +444,7 @@ type Statistics struct {
 	Wavefront         *expvar.Map
 	Fission           *expvar.Map
 	Grafana           *expvar.Map
+	YandexS3          *expvar.Map
 }
 
 // PromStatistics is a struct to store prometheus metrics