Improve system test coverage - 1

packages/akamai/_dev/deploy/docker/docker-compose.yml

@@ -1,4 +1,4 @@
-version: '2.3'
+version: "2.3"
 services:
   akamai-siem-emulator:
     hostname: akamai-siem-emulator
@@ -13,3 +13,17 @@ services:
       - -access-token=at-6b8c7217-8748-490d-b0f5-bfeb72b2e7cd
       - -client-secret=cs-0d15cfd9-764a-48e6-a822-22756180ddb8
       - -client-token=ct-f625f0b8-9c8f-44ce-8250-eaf17bc93051
+  gcs-mock-service:
+    image: golang:1.24.7-alpine
+    working_dir: /app
+    volumes:
+      - ./gcs-mock-service:/app
+      - ./files/manifest.yml:/files/manifest.yml:ro
+      - ./sample_logs/:/data:ro
+    ports:
+      - "4443/tcp"
+    healthcheck:
+      test: "wget --no-verbose --tries=1 --spider http://localhost:4443/health || exit 1"
+      interval: 10s
+      timeout: 5s
+    command: go run main.go -manifest /files/manifest.yml

packages/akamai/_dev/deploy/docker/files/manifest.yml

@@ -0,0 +1,5 @@
+buckets:
+  testbucket:
+    files:
+      - path: /data/siem.log
+        content-type: application/x-ndjson

packages/akamai/_dev/deploy/docker/gcs-mock-service/go.mod

@@ -0,0 +1,5 @@
+module gcs-mock-service
+
+go 1.24.7
+
+require gopkg.in/yaml.v3 v3.0.1

packages/akamai/_dev/deploy/docker/gcs-mock-service/go.sum

@@ -0,0 +1,4 @@
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

packages/akamai/_dev/deploy/docker/gcs-mock-service/main.go

@@ -0,0 +1,297 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package main
+
+import (
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"os"
+	"strconv"
+	"strings"
+
+	"gopkg.in/yaml.v3"
+)
+
+func main() {
+	host := flag.String("host", "0.0.0.0", "host to listen on")
+	port := flag.String("port", "4443", "port to listen on")
+	manifest := flag.String("manifest", "", "path to YAML manifest file for preloading buckets and objects")
+	flag.Parse()
+
+	addr := fmt.Sprintf("%s:%s", *host, *port)
+
+	fmt.Printf("Starting mock GCS server on http://%s\n", addr)
+	if *manifest != "" {
+		m, err := readManifest(*manifest)
+		if err != nil {
+			log.Fatalf("error reading manifest: %v", err)
+		}
+		if err := processManifest(m); err != nil {
+			log.Fatalf("error processing manifest: %v", err)
+		}
+	} else {
+		fmt.Println("Store is empty. Create buckets and objects via API calls.")
+	}
+
+	// setup HTTP handlers
+	mux := http.NewServeMux()
+	// health check
+	mux.HandleFunc("/health", healthHandler)
+	// standard gcs api calls
+	mux.HandleFunc("GET /storage/v1/b/{bucket}/o", handleListObjects)
+	mux.HandleFunc("GET /storage/v1/b/{bucket}/o/{object...}", handleGetObject)
+	mux.HandleFunc("POST /storage/v1/b", handleCreateBucket)
+	mux.HandleFunc("POST /upload/storage/v1/b/{bucket}/o", handleUploadObject)
+	mux.HandleFunc("POST /upload/storage/v1/b/{bucket}/o/{object...}", handleUploadObject)
+	// direct path-style gcs sdk calls
+	mux.HandleFunc("GET /{bucket}/o/{object...}", handleGetObject)
+	mux.HandleFunc("GET /{bucket}/{object...}", handleGetObject)
+	// debug: log all requests
+	loggedMux := loggingMiddleware(mux)
+
+	if err := http.ListenAndServe(addr, loggedMux); err != nil {
+		log.Fatalf("failed to start server: %v", err)
+	}
+}
+
+// loggingMiddleware logs incoming HTTP requests.
+func loggingMiddleware(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Printf("%s %s\n", r.Method, r.URL.Path)
+		next.ServeHTTP(w, r)
+	})
+}
+
+// readManifest reads and parses the YAML manifest file.
+func readManifest(path string) (*Manifest, error) {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read manifest: %w", err)
+	}
+
+	var manifest Manifest
+	if err := yaml.Unmarshal(data, &manifest); err != nil {
+		return nil, fmt.Errorf("failed to parse manifest: %w", err)
+	}
+
+	return &manifest, nil
+}
+
+// processManifest creates buckets and uploads objects as specified in the manifest.
+func processManifest(manifest *Manifest) error {
+	for bucketName, bucket := range manifest.Buckets {
+		for _, file := range bucket.Files {
+			fmt.Printf("preloading data for bucket: %s | path: %s | content-type: %s...\n",
+				bucketName, file.Path, file.ContentType)
+
+			if err := createBucket(bucketName); err != nil {
+				return fmt.Errorf("failed to create bucket '%s': %w", bucketName, err)
+			}
+			data, err := os.ReadFile(file.Path)
+			if err != nil {
+				return fmt.Errorf("failed to read bucket data file '%s': %w", file.Path, err)
+			}
+			pathParts := strings.Split(file.Path, "/")
+			if _, err := uploadObject(bucketName, pathParts[len(pathParts)-1], data, file.ContentType); err != nil {
+				return fmt.Errorf("failed to create object '%s' in bucket '%s': %w", file.Path, bucketName, err)
+			}
+		}
+	}
+	return nil
+}
+
+// healthHandler responds with a simple "OK" message for health checks.
+func healthHandler(w http.ResponseWriter, r *http.Request) {
+	w.WriteHeader(http.StatusOK)
+	fmt.Fprint(w, "OK")
+}
+
+// handleListObjects lists all objects in the specified bucket.
+func handleListObjects(w http.ResponseWriter, r *http.Request) {
+	bucketName := r.PathValue("bucket")
+
+	if bucket, ok := inMemoryStore[bucketName]; ok {
+		response := GCSListResponse{
+			Kind:  "storage#objects",
+			Items: []GCSObject{},
+		}
+		for name, object := range bucket {
+			item := GCSObject{
+				Kind:        "storage#object",
+				Name:        name,
+				Bucket:      bucketName,
+				Size:        strconv.Itoa(len(object.Data)),
+				ContentType: object.ContentType,
+			}
+			response.Items = append(response.Items, item)
+		}
+		w.Header().Set("Content-Type", "application/json")
+		json.NewEncoder(w).Encode(response)
+		return
+	}
+	http.Error(w, "not found", http.StatusNotFound)
+}
+
+// handleGetObject retrieves a specific object from a bucket.
+func handleGetObject(w http.ResponseWriter, r *http.Request) {
+	bucketName := r.PathValue("bucket")
+	objectName := r.PathValue("object")
+
+	if bucketName == "" || objectName == "" {
+		http.Error(w, "not found: invalid URL format", http.StatusNotFound)
+		return
+	}
+
+	if bucket, ok := inMemoryStore[bucketName]; ok {
+		if object, ok := bucket[objectName]; ok {
+			w.Header().Set("Content-Type", object.ContentType)
+			w.Write(object.Data)
+			return
+		}
+	}
+	http.Error(w, "not found", http.StatusNotFound)
+}
+
+// handleCreateBucket creates a new bucket.
+func handleCreateBucket(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.NotFound(w, r)
+		return
+	}
+
+	var bucketInfo struct {
+		Name string `json:"name"`
+	}
+	if err := json.NewDecoder(r.Body).Decode(&bucketInfo); err != nil {
+		http.Error(w, "invalid JSON body", http.StatusBadRequest)
+		return
+	}
+	if bucketInfo.Name == "" {
+		http.Error(w, "bucket name is required", http.StatusBadRequest)
+		return
+	}
+
+	if err := createBucket(bucketInfo.Name); err != nil {
+		http.Error(w, err.Error(), http.StatusConflict)
+		return
+	}
+
+	w.WriteHeader(http.StatusOK)
+	json.NewEncoder(w).Encode(bucketInfo)
+}
+
+// handleUploadObject uploads an object to a specified bucket.
+func handleUploadObject(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.NotFound(w, r)
+		return
+	}
+
+	bucketName := r.PathValue("bucket")
+	objectName := r.URL.Query().Get("name")
+	if objectName == "" {
+		objectName = r.PathValue("object")
+	}
+
+	if bucketName == "" || objectName == "" {
+		http.Error(w, "missing bucket or object name", http.StatusBadRequest)
+		return
+	}
+
+	data, err := io.ReadAll(r.Body)
+	if err != nil {
+		http.Error(w, "failed to read request body", http.StatusInternalServerError)
+		return
+	}
+	defer r.Body.Close()
+
+	contentType := r.Header.Get("Content-Type")
+	if contentType == "" {
+		contentType = "application/octet-stream"
+	}
+
+	response, err := uploadObject(bucketName, objectName, data, contentType)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusNotFound)
+		return
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(response)
+}
+
+func createBucket(bucketName string) error {
+	if _, exists := inMemoryStore[bucketName]; exists {
+		return fmt.Errorf("bucket already exists")
+	}
+	inMemoryStore[bucketName] = make(map[string]ObjectData)
+	log.Printf("created bucket: %s", bucketName)
+	return nil
+}
+
+func uploadObject(bucketName, objectName string, data []byte, contentType string) (*GCSObject, error) {
+	if _, ok := inMemoryStore[bucketName]; !ok {
+		return nil, fmt.Errorf("bucket not found")
+	}
+
+	inMemoryStore[bucketName][objectName] = ObjectData{
+		Data:        data,
+		ContentType: contentType,
+	}
+	log.Printf("created object '%s' in bucket '%s' with Content-Type '%s'",
+		objectName, bucketName, contentType)
+
+	return &GCSObject{
+		Kind:        "storage#object",
+		Name:        objectName,
+		Bucket:      bucketName,
+		Size:        strconv.Itoa(len(data)),
+		ContentType: contentType,
+	}, nil
+}
+
+// The in-memory store to hold ObjectData structs.
+var inMemoryStore = make(map[string]map[string]ObjectData)
+
+// ObjectData stores the raw data and its content type.
+type ObjectData struct {
+	Data        []byte
+	ContentType string
+}
+
+// GCSListResponse mimics the structure of a real GCS object list response.
+type GCSListResponse struct {
+	Kind  string      `json:"kind"`
+	Items []GCSObject `json:"items"`
+}
+
+// GCSObject mimics the structure of a GCS object resource with ContentType.
+type GCSObject struct {
+	Kind        string `json:"kind"`
+	Name        string `json:"name"`
+	Bucket      string `json:"bucket"`
+	Size        string `json:"size"`
+	ContentType string `json:"contentType"`
+}
+
+// Manifest represents the top-level structure of the YAML file
+type Manifest struct {
+	Buckets map[string]Bucket `yaml:"buckets"`
+}
+
+// Bucket represents each bucket and its files
+type Bucket struct {
+	Files []File `yaml:"files"`
+}
+
+// File represents each file entry inside a bucket
+type File struct {
+	Path        string `yaml:"path"`
+	ContentType string `yaml:"content-type"`
+}

packages/akamai/_dev/deploy/docker/sample_logs/siem.log

@@ -0,0 +1 @@
+{"attackData":{"clientIP":"89.160.20.156","configId":"6724","policyId":"scoe_5426","ruleActions":"QUxFUlQ;REVOWQ==","ruleData":"YWxlcnQo;Y3VybA==","ruleMessages":"Q3Jvc3Mtc2l0ZSBTY3 JpcHRpbmcgKFhTUykgQXR0YWNr; UmVxdWVzdCBJbmRpY2F0ZXMgYW4 gYXV0b21hdGVkIHByb2 dyYW0gZXhwbG9yZWQgdGhlIHNpdGU=","ruleSelectors":"QVJHUzph;UkVRVUVTVF9IRU FERVJTOlVzZXItQWdlbnQ=","ruleTags":"V0VCX0FUVEFDSy9YU1M=;QV VUT01BVElPTi9NSVND","ruleVersions":";","rules":"OTUwMDA0;OTkwMDEx"},"botData":{"botScore":"100","responseSegment":"3"},"clientData":{"appBundleId":"com.mydomain.myapp","appVersion":"1.23","sdkVersion":"4.7.1","telemetryType":"2"},"format":"json","geo":{"asn":"12271","city":"NEWYORK","continent":"NA","country":"US","regionCode":"NY"},"httpMessage":{"bytes":"34523","host":"www.example.com","method":"POST","path":"/examples/1/","port":"80","protocol":"http/2","query":"a%3D..%2F..%2F..%2Fetc%2Fpasswd","requestHeaders":"User-Agent%3a%20BOT%2f0.1%20(BOT%20for%20JCE)%0d%0aAccept%3a%20text%2fhtml,application%2fxhtml+xml","requestId":"2ab418ac8515f33","responseHeaders":"Server%3a%20AkamaiGHost%0d%0aMime-Version%3a%201.0%0d%0aContent-Type%3a%20text%2fhtml","start":"1470923133.026","status":"301","tls":"TLSv1.2"},"type":"akamai_siem","userRiskData":{"allow":"0","general":"duc_1h:10|duc_1d:30","risk":"udfp:1325gdg4g4343g/M|unp:74256/H","score":"75","status":"0","trust":"ugp:US","uuid":"964d54b7-0821-413a-a4d6-8131770ec8d5"},"version":"1.0"}

packages/akamai/data_stream/siem/_dev/test/system/test-gcs-config.yml

@@ -0,0 +1,16 @@
+deployer: docker
+service: gcs-mock-service
+input: gcs
+vars:
+data_stream:
+  vars:
+    project_id: fake-gcs-project
+    alternative_host: "http://{{Hostname}}:{{Port}}"
+    number_of_workers: 1
+    poll: true
+    poll_interval: 15s
+    service_account_key: "{\"type\":\"service_account\",\"project_id\":\"fake-gcs-project\"}"
+    buckets: |
+      - name: testbucket
+assert:
+  hit_count: 1

packages/akamai/data_stream/siem/agent/stream/gcs.yml.hbs

@@ -1,6 +1,9 @@
 {{#if project_id}}
 project_id: {{project_id}}
 {{/if}}
+{{#if alternative_host}}
+alternative_host: {{alternative_host}}
+{{/if}}
 {{#if service_account_key}}
 auth.credentials_json.account_key: {{service_account_key}}
 {{/if}}

packages/akamai/data_stream/siem/fields/beats.yml

@@ -7,3 +7,15 @@
 - name: log.offset
   type: long
   description: Offset of the entry in the log file.
+- name: gcs.storage
+  type: group
+  fields:
+    - name: bucket.name
+      type: keyword
+      description: The name of the Google Cloud Storage Bucket.
+    - name: object.name
+      type: keyword
+      description: The content type of the Google Cloud Storage object.
+    - name: object.content_type
+      type: keyword
+      description: The content type of the Google Cloud Storage object.

packages/akamai/data_stream/siem/manifest.yml

@@ -199,6 +199,13 @@ streams:
           #  poll: true
           #  poll_interval: 10s
           #  bucket_timeout: 30s
+      - name: alternative_host
+        type: text
+        title: Alternative Host
+        description: Used to override the default host for the storage client (default is storage.googleapis.com)
+        required: false
+        multi: false
+        show_user: false
       - name: processors
         type: yaml
         title: Processors