fix(kaspersky): classify final actions and recover endpoint IPs

plugins/alerts/kaspersky_action_result_test.go

@@ -0,0 +1,61 @@
+package main
+
+import (
+	"encoding/json"
+	"os"
+	"testing"
+
+	"github.com/threatwinds/go-sdk/plugins"
+	"github.com/tidwall/gjson"
+)
+
+// Synthetic CEF and KSC records cover final decisions and IP roles. The
+// isolated EventProcessor parser is replayed separately on the same inputs.
+func TestKasperskyActionResultRaw(t *testing.T) {
+	var cases []struct {
+		Name       string            `json:"name"`
+		Raw        string            `json:"raw"`
+		DataSource string            `json:"dataSource"`
+		Result     string            `json:"result"`
+		Expected   map[string]string `json:"expected"`
+		Absent     []string          `json:"absent"`
+		Initial    map[string]any    `json:"initial"`
+	}
+	data, err := os.ReadFile("testdata/kaspersky_action_result.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := json.Unmarshal(data, &cases); err != nil {
+		t.Fatal(err)
+	}
+	if len(cases) < 17 {
+		t.Fatal("outcome and parser regression classes are missing")
+	}
+	config := kaspConfig(t)
+	cache := plugins.NewCELCache("kaspersky-final-outcome")
+	for _, tc := range cases {
+		t.Run(tc.Name, func(t *testing.T) {
+			out := kaspParse(t, config, tc.Raw, tc.DataSource, cache, tc.Initial)
+			if got := gjson.Get(out, "actionResult").String(); got != tc.Result {
+				t.Errorf("actionResult = %q; want %q", got, tc.Result)
+			}
+			for path, want := range tc.Expected {
+				got := gjson.Get(out, path)
+				if !got.Exists() || got.String() != want {
+					t.Errorf("%s = %q; want %q", path, got.String(), want)
+				}
+			}
+			for _, path := range tc.Absent {
+				if gjson.Get(out, path).Exists() {
+					t.Errorf("unexpected %s", path)
+				}
+			}
+			for _, result := range []string{"success", "failure", "denied"} {
+				matched, err := cache.Eval(`equals("actionResult","`+result+`")`, out)
+				if err != nil || matched != (tc.Result == result) {
+					t.Errorf("outcome predicate %s = %v (%v)", result, matched, err)
+				}
+			}
+		})
+	}
+}

plugins/alerts/kaspersky_contract_test.go

@@ -1,9 +1,8 @@
 package main
 
 // Offline Kaspersky extraction model, not the closed EventProcessor.
-// Explicit YAML grok/rename/cast/trim/add/delete and documented KV splitting are
-// modeled. CEL, Event serialization, placeholder expansion, query creation and
-// history thresholds use SDK v1.1.31. External geolocation is not executed.
+// Grok patterns are consumed in order, with whitespace trimmed before each
+// match, as the EventProcessor grok plugin does. External geolocation is not run.
 import (
 	"bytes"
 	"encoding/json"
@@ -77,21 +76,13 @@ func kaspConfig(t *testing.T) *plugins.Config {
 	}
 	return c
 }
-func kaspRegex(t *testing.T, g *plugins.Grok, cfg *plugins.Config) *regexp.Regexp {
+func kaspRegex(t *testing.T, pattern string, cfg *plugins.Config) *regexp.Regexp {
 	t.Helper()
-	var pattern strings.Builder
-	for i, p := range g.Patterns {
-		if p.FieldName != "" {
-			fmt.Fprintf(&pattern, "(?P<f%d>%s)", i, p.Pattern)
-		} else {
-			pattern.WriteString("(?:" + p.Pattern + ")")
-		}
-	}
 	pats := map[string]string{"greedy": ".*", "data": ".*?", "word": "[A-Za-z0-9_-]+", "space": "\\s+"}
 	for k, v := range cfg.Patterns {
 		pats[k] = v
 	}
-	tmpl, e := template.New("grok").Option("missingkey=error").Parse(pattern.String())
+	tmpl, e := template.New("grok").Option("missingkey=error").Parse(pattern)
 	if e != nil {
 		t.Fatal(e)
 	}
@@ -105,9 +96,25 @@ func kaspRegex(t *testing.T, g *plugins.Grok, cfg *plugins.Config) *regexp.Regex
 	}
 	return r
 }
-func kaspParse(t *testing.T, cfg *plugins.Config, raw string, dataSource string, cache *plugins.CELCache) string {
+
+// kaspStoredName is the name the parser plugins store for a grok, rename or add target:
+// utils.SanitizeField keeps only letters, digits and dots.
+func kaspStoredName(name string) string {
+	utils.SanitizeField(&name)
+	return name
+}
+
+func kaspParse(t *testing.T, cfg *plugins.Config, raw string, dataSource string, cache *plugins.CELCache, initial ...map[string]any) string {
 	t.Helper()
 	draft := map[string]any{"raw": raw, "dataType": "antivirus-kaspersky", "dataSource": dataSource, "log": map[string]any{}}
+	if len(initial) > 0 {
+		for key, value := range initial[0] {
+			if key != "action" && key != "actionResult" {
+				t.Fatalf("unexpected seeded field %s", key)
+			}
+			draft[key] = value
+		}
+	}
 	for _, stage := range cfg.Pipeline {
 		matched := false
 		for _, dataType := range stage.DataTypes {
@@ -156,20 +163,34 @@ func kaspParse(t *testing.T, cfg *plugins.Config, raw string, dataSource string,
 					if !ok {
 						t.Fatalf("non-string grok source %s", src)
 					}
-					r := kaspRegex(t, g, cfg)
-					m := r.FindStringSubmatch(str)
-					if m == nil {
-						continue
-					}
-					for i, p := range g.Patterns {
+					fields := make(map[string]string)
+					matched := 0
+					for _, p := range g.Patterns {
+						str = strings.TrimSpace(str)
+						if str == "" {
+							break
+						}
+						r := kaspRegex(t, p.Pattern, cfg)
+						loc := r.FindStringIndex(str)
+						if loc == nil || loc[0] != 0 || loc[1] == 0 {
+							break
+						}
+						value := str[:loc[1]]
 						if p.FieldName != "" {
-							kaspPut(draft, p.FieldName, m[r.SubexpIndex(fmt.Sprintf("f%d", i))], false)
+							fields[p.FieldName] = strings.TrimSpace(value)
+						}
+						str = str[loc[1]:]
+						matched++
+					}
+					if matched == len(g.Patterns) {
+						for field, value := range fields {
+							kaspPut(draft, kaspStoredName(field), value, false)
 						}
 					}
 				case "rename":
 					for _, p := range s.Rename.From {
 						if v, ok := kaspGet(draft, p); ok {
-							kaspPut(draft, s.Rename.To, v, false)
+							kaspPut(draft, kaspStoredName(s.Rename.To), v, false)
 							kaspPut(draft, p, nil, true)
 							break
 						}
@@ -196,7 +217,7 @@ func kaspParse(t *testing.T, cfg *plugins.Config, raw string, dataSource string,
 					if s.Add.Function != "string" {
 						t.Fatalf("unsupported add function %s", s.Add.Function)
 					}
-					kaspPut(draft, s.Add.Params["key"].GetStringValue(), s.Add.Params["value"].AsInterface(), false)
+					kaspPut(draft, kaspStoredName(s.Add.Params["key"].GetStringValue()), s.Add.Params["value"].AsInterface(), false)
 				case "delete":
 					for _, p := range s.Delete.Fields {
 						kaspPut(draft, p, nil, true)

plugins/alerts/kaspersky_history_test.go

@@ -37,7 +37,10 @@ func TestKasperskySDKHistory(t *testing.T) {
 	paths := []string{"dataSource", "log.endpointKeyType", "log.endpointKey", "origin.ip", "target.ip", "log.cat"}
 	for name, r := range rules {
 		if len(r.Correlation) > 0 {
-			paths = append(paths, "log.correlationCandidate."+name)
+			if kasperskyHistoryMarkers[name] == "" {
+				t.Fatalf("history rule %s has no correlation marker", name)
+			}
+			paths = append(paths, "log.correlationCandidate."+kasperskyHistoryMarkers[name])
 		}
 	}
 	for _, path := range paths {
@@ -195,7 +198,8 @@ func TestKasperskySDKHistory(t *testing.T) {
 			if ok, e := cache.Eval(r.Where, out); e != nil || !ok {
 				t.Fatalf("raw trigger failed: %v %v", ok, e)
 			}
-			marker := "log.correlationCandidate." + tc.rule
+			// The filter stores this marker and the rule counts it under the same name.
+			marker := "log.correlationCandidate." + kasperskyHistoryMarkers[tc.rule]
 			terms = map[string]string{"dataSource": "collector-test", "log.endpointKeyType": "ip", marker: "match"}
 			notTerms = map[string]string{}
 			if tc.cross {

plugins/alerts/kaspersky_marker_names_test.go

@@ -0,0 +1,166 @@
+package main
+
+// The parser plugins keep only letters, digits and dots in the field names they write
+// (go-sdk utils.SanitizeField), while rules look names up exactly as written. A marker
+// added as log.correlationCandidate.lateral_movement_indicators is stored as
+// ...lateralmovementindicators, so a history rule counting the underscored name never fires.
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"regexp"
+	"slices"
+	"strings"
+	"testing"
+
+	"github.com/threatwinds/go-sdk/plugins"
+	"github.com/threatwinds/go-sdk/utils"
+	"google.golang.org/protobuf/encoding/protojson"
+)
+
+// kasperskyHistoryMarkers maps each Kaspersky history rule (file name without extension)
+// to the correlation marker that filters/antivirus/kaspersky.yml adds and the rule counts.
+var kasperskyHistoryMarkers = map[string]string{
+	"data_exfiltration_attempts":    "dataExfiltrationAttempts",
+	"kaspersky_ransomware_behavior": "kasperskyRansomwareBehavior",
+	"lateral_movement_indicators":   "lateralMovementIndicators",
+	"suspicious_network_activity":   "suspiciousNetworkActivity",
+}
+
+// Every marker the Kaspersky filter adds and every marker a Kaspersky rule reads (where,
+// history fields and placeholders, groupBy, deduplicateBy) must be the same string, made
+// only of letters, digits and dots, so the stored name is the name the rule looks up.
+func TestKasperskyCorrelationMarkerNames(t *testing.T) {
+	const prefix = "log.correlationCandidate."
+	clean := regexp.MustCompile(`^[A-Za-z0-9.]+$`)
+	kept := func(name string) bool {
+		stored := name
+		utils.SanitizeField(&stored)
+		return stored == name && clean.MatchString(name)
+	}
+
+	written := map[string]string{}
+	for _, stage := range kaspConfig(t).Pipeline {
+		if !slices.Contains(stage.DataTypes, "antivirus-kaspersky") {
+			continue
+		}
+		for _, step := range stage.Steps {
+			names := []string{}
+			if s := step.Grok; s != nil {
+				for _, p := range s.Patterns {
+					if p.FieldName != "" {
+						names = append(names, p.FieldName)
+					}
+				}
+			}
+			if s := step.Rename; s != nil {
+				names = append(names, s.To)
+			}
+			if s := step.Csv; s != nil {
+				names = append(names, s.Headers...)
+			}
+			if s := step.Add; s != nil {
+				key := s.Params["key"].GetStringValue()
+				names = append(names, key)
+				if strings.HasPrefix(key, prefix) {
+					value := s.Params["value"].GetStringValue()
+					if previous, ok := written[key]; ok && previous != value {
+						t.Errorf("filter adds %s as %q and %q", key, previous, value)
+					}
+					written[key] = value
+				}
+			}
+			for _, name := range names {
+				if !kept(name) {
+					t.Errorf("filter writes %q, which the parser stores under another name", name)
+				}
+			}
+		}
+	}
+
+	reference := regexp.MustCompile(`log\.correlationCandidate\.[^"'\s,()\[\]{}]*`)
+	read := map[string]map[string]bool{}
+	err := filepath.WalkDir("../../rules", func(path string, d os.DirEntry, err error) error {
+		if err != nil || d.IsDir() || (filepath.Ext(path) != ".yml" && filepath.Ext(path) != ".yaml") {
+			return err
+		}
+		b, err := utils.ReadPbYaml(path)
+		if err != nil {
+			return err
+		}
+		var head struct {
+			DataTypes []string `json:"dataTypes"`
+		}
+		if err = json.Unmarshal(b, &head); err != nil || !slices.Contains(head.DataTypes, "antivirus-kaspersky") {
+			return err
+		}
+		rule := new(plugins.Rule)
+		if err = protojson.Unmarshal(b, rule); err != nil {
+			return err
+		}
+		name := strings.TrimSuffix(filepath.Base(path), filepath.Ext(path))
+		texts := []string{rule.Where}
+		texts = append(texts, rule.GroupBy...)
+		texts = append(texts, rule.DeduplicateBy...)
+		var searches func([]*plugins.SearchRequest)
+		searches = func(list []*plugins.SearchRequest) {
+			for _, search := range list {
+				for _, term := range search.With {
+					value := term.Value.GetStringValue()
+					texts = append(texts, term.Field, value)
+					if strings.HasPrefix(term.Field, prefix) && written[term.Field] != value {
+						t.Errorf("%s counts %s = %q; the filter adds %q", path, term.Field, value, written[term.Field])
+					}
+				}
+				searches(search.Or)
+			}
+		}
+		searches(rule.AfterEvents)
+		searches(rule.Correlation)
+		for _, text := range texts {
+			for _, marker := range reference.FindAllString(text, -1) {
+				marker = strings.TrimSuffix(marker, ".keyword")
+				if read[marker] == nil {
+					read[marker] = map[string]bool{}
+				}
+				read[marker][name] = true
+				if !kept(marker) {
+					t.Errorf("%s reads %q, which the parser never stores under that name", path, marker)
+				}
+				if _, ok := written[marker]; !ok {
+					t.Errorf("%s reads %q, which the Kaspersky filter does not add", path, marker)
+				}
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	expected := map[string]bool{}
+	for rule, marker := range kasperskyHistoryMarkers {
+		expected[prefix+marker] = true
+		if _, ok := written[prefix+marker]; !ok {
+			t.Errorf("filter does not add %s%s for %s", prefix, marker, rule)
+		}
+		if !read[prefix+marker][rule] {
+			t.Errorf("%s does not read %s%s", rule, prefix, marker)
+		}
+	}
+	for marker, rules := range read {
+		for rule := range rules {
+			if prefix+kasperskyHistoryMarkers[rule] != marker {
+				t.Errorf("%s reads %s, expected only %s%s", rule, marker, prefix, kasperskyHistoryMarkers[rule])
+			}
+		}
+	}
+	for marker := range written {
+		if !expected[marker] {
+			t.Errorf("filter adds %s, which no Kaspersky history rule counts", marker)
+		}
+	}
+	if len(written) != 4 || len(kasperskyHistoryMarkers) != 4 {
+		t.Errorf("marker coverage: filter adds %d, table lists %d", len(written), len(kasperskyHistoryMarkers))
+	}
+}

plugins/alerts/testdata/filter-contracts/kaspersky.json

@@ -34,7 +34,7 @@
           "format": "cef",
           "endpointKey": "forged",
           "correlationCandidate": {
-            "kaspersky_ransomware_behavior": "match"
+            "kasperskyRansomwareBehavior": "match"
           }
         }
       },