Map Azure action results to final operation outcomes

plugins/alerts/azure_action_result_test.go

@@ -0,0 +1,57 @@
+package main
+
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/threatwinds/go-sdk/plugins"
+	"github.com/tidwall/gjson"
+)
+
+// Fabricated raw records exercise overlapping final and intermediate vendor signals.
+// The parser here is the existing offline model; EventProcessor replay is separate.
+func TestAzureActionResultRaw(t *testing.T) {
+	content, err := os.ReadFile("testdata/azure_action_result.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var cases []struct {
+		Name   string `json:"name"`
+		Raw    string `json:"raw"`
+		Result string `json:"result"`
+		Rule   string `json:"rule"`
+		Match  bool   `json:"match"`
+	}
+	if err := json.Unmarshal(content, &cases); err != nil {
+		t.Fatal(err)
+	}
+	if len(cases) < 20 {
+		t.Fatalf("unexpected outcome coverage: %d", len(cases))
+	}
+	config, cache, rules := azureConfig(t), plugins.NewCELCache("azure-final-outcome"), azureRules(t)
+	for _, c := range cases {
+		t.Run(c.Name, func(t *testing.T) {
+			got := azureParse(t, config, c.Raw, "synthetic-collector", cache)
+			if value := gjson.Get(got, "actionResult"); value.String() != c.Result {
+				t.Errorf("actionResult = %q, want %q; kind=%q statusCode=%q vendor=%q api=%q kubeKind=%q stage=%q", value.String(), c.Result, gjson.Get(got, "log.azureKind").String(), gjson.Get(got, "statusCode").String(), gjson.Get(got, "log.azureProperties.ScStatus").String(), gjson.Get(got, "log.azureKubernetes.apiVersion").String(), gjson.Get(got, "log.azureKubernetes.kind").String(), gjson.Get(got, "log.azureKubernetes.stage").String())
+			}
+			if c.Rule != "" {
+				key := strings.TrimSuffix(filepath.Base(c.Rule), filepath.Ext(c.Rule))
+				rule := rules[key]
+				if rule == nil {
+					t.Fatalf("unknown rule %s", c.Rule)
+				}
+				match, err := cache.Eval(rule.Where, got)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if match != c.Match {
+					t.Errorf("%s matched %v, want %v", key, match, c.Match)
+				}
+			}
+		})
+	}
+}

plugins/alerts/azure_contract_test.go

@@ -1,8 +1,8 @@
 package main
 
-// Offline Azure extraction model, not the closed EventProcessor.
+// Offline Azure extraction model, separate from actual EventProcessor playground runs.
 // Explicit YAML JSON/key sanitization, grok, rename, add and delete steps are modeled.
-// CEL and Event serialization use SDK v1.1.31. History requests are tested separately
+// CEL and Event serialization use the reviewed module's SDK v1.1.33. History requests are tested separately
 // with that SDK. External geolocation is mocked only when a fixture declares it.
 import (
 	"bytes"
@@ -118,11 +118,20 @@ func azureRegex(t *testing.T, g *plugins.Grok, cfg *plugins.Config) *regexp.Rege
 	}
 	return r
 }
+
+// azureStoredName is the name the parser plugins store for a grok, rename or add target:
+// utils.SanitizeField keeps only letters, digits and dots.
+func azureStoredName(name string) string {
+	utils.SanitizeField(&name)
+	return name
+}
+
 func azureParse(t *testing.T, cfg *plugins.Config, raw string, dataSource string, cache *plugins.CELCache, enrichment ...map[string]any) string {
 	return azureParseMode(t, cfg, raw, dataSource, cache, false, enrichment...)
 }
 
-// Both modes model the unresolved nested-key behavior of the closed JSON step.
+// Both modes cover nested-key compatibility. The separately pinned public JSON
+// plugin preserves nested keys; deployed extractor versions can differ.
 func azureParseMode(t *testing.T, cfg *plugins.Config, raw string, dataSource string, cache *plugins.CELCache, preserveNested bool, enrichment ...map[string]any) string {
 	t.Helper()
 	draft := map[string]any{"raw": raw, "dataType": "azure", "dataSource": dataSource, "log": map[string]any{}}
@@ -181,13 +190,13 @@ func azureParseMode(t *testing.T, cfg *plugins.Config, raw string, dataSource st
 					}
 					for i, p := range g.Patterns {
 						if p.FieldName != "" {
-							azurePut(draft, p.FieldName, m[r.SubexpIndex(fmt.Sprintf("f%d", i))], false)
+							azurePut(draft, azureStoredName(p.FieldName), m[r.SubexpIndex(fmt.Sprintf("f%d", i))], false)
 						}
 					}
 				case "rename":
 					for _, p := range s.Rename.From {
 						if v, ok := azureGet(draft, p); ok {
-							azurePut(draft, s.Rename.To, v, false)
+							azurePut(draft, azureStoredName(s.Rename.To), v, false)
 							azurePut(draft, p, nil, true)
 							break
 						}
@@ -196,7 +205,7 @@ func azureParseMode(t *testing.T, cfg *plugins.Config, raw string, dataSource st
 					if s.Add.Function != "string" {
 						t.Fatalf("unsupported add function %s", s.Add.Function)
 					}
-					azurePut(draft, s.Add.Params["key"].GetStringValue(), s.Add.Params["value"].AsInterface(), false)
+					azurePut(draft, azureStoredName(s.Add.Params["key"].GetStringValue()), s.Add.Params["value"].AsInterface(), false)
 				case "delete":
 					for _, p := range s.Delete.Fields {
 						azurePut(draft, p, nil, true)
@@ -228,12 +237,17 @@ func azureParseMode(t *testing.T, cfg *plugins.Config, raw string, dataSource st
 					if !ok {
 						continue
 					}
-					str, ok := source.(string)
-					if !ok {
-						t.Fatalf("JSON source is not a string")
+					var encoded []byte
+					if str, ok := source.(string); ok {
+						encoded = []byte(str)
+					} else {
+						encoded, e = json.Marshal(source)
+						if e != nil {
+							t.Fatal(e)
+						}
 					}
 					var parsed map[string]any
-					if e := json.Unmarshal([]byte(str), &parsed); e != nil {
+					if e := json.Unmarshal(encoded, &parsed); e != nil {
 						t.Fatal(e)
 					}
 					normalized := azureSanitizeJSON(parsed)

plugins/alerts/azure_history_test.go

@@ -37,7 +37,10 @@ func TestAzureSDKHistory(t *testing.T) {
 	paths := []string{"dataSource", "log.azureScopeType", "log.azureScope", "log.azureActorType", "log.azureActor", "origin.ip"}
 	for name, r := range rules {
 		if len(r.Correlation) > 0 {
-			paths = append(paths, "log.correlationCandidate."+name)
+			if azureHistoryMarkers[name] == "" {
+				t.Fatalf("history rule %s has no correlation marker", name)
+			}
+			paths = append(paths, "log.correlationCandidate."+azureHistoryMarkers[name])
 		}
 	}
 	for _, path := range paths {
@@ -206,7 +209,8 @@ func TestAzureSDKHistory(t *testing.T) {
 			if yes, e := cache.Eval(r.Where, out); e != nil || !yes {
 				t.Fatalf("raw trigger failed: %v %v", yes, e)
 			}
-			marker := "log.correlationCandidate." + tc.rule
+			// The filter stores this marker and the rule counts it under the same name.
+			marker := "log.correlationCandidate." + azureHistoryMarkers[tc.rule]
 			terms = map[string]string{"dataSource": "collector-test", "log.azureScopeType": "directory", "log.azureScope": "directory-test", marker: "true"}
 			notTerms = map[string]string{}
 			if tc.rule == "azure_kubernetes_secret_access" || tc.rule == "application_gateway_waf_alerts" {
@@ -272,7 +276,7 @@ func TestAzureSDKHistory(t *testing.T) {
 			if e := json.Unmarshal([]byte(f.Raw), &raw); e != nil {
 				t.Fatal(e)
 			}
-			raw["correlationCandidate"] = map[string]any{tc.rule: "true"}
+			raw["correlationCandidate"] = map[string]any{azureHistoryMarkers[tc.rule]: "true"}
 			raw["category"] = "AppServiceConsoleLogs"
 			raw["operationName"] = "Microsoft.Web/sites/log"
 			delete(raw, "properties")

plugins/alerts/azure_marker_names_test.go

@@ -0,0 +1,172 @@
+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.key_vault_access_spikes is stored as
+// ...keyvaultaccessspikes, 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"
+)
+
+// azureHistoryMarkers maps each Azure history rule (file name without extension) to the
+// correlation marker that filters/azure/azure-eventhub.yml adds and the rule counts.
+var azureHistoryMarkers = map[string]string{
+	"aks_security_threats":           "aksSecurityThreats",
+	"app_registration_abuse":         "appRegistrationAbuse",
+	"application_gateway_waf_alerts": "applicationGatewayWafAlerts",
+	"azure_ad_password_spray":        "azureAdPasswordSpray",
+	"azure_bulk_role_changes":        "azureBulkRoleChanges",
+	"azure_kubernetes_secret_access": "azureKubernetesSecretAccess",
+	"azure_laps_credential_dump":     "azureLapsCredentialDump",
+	"azure_ropc_authentication":      "azureRopcAuthentication",
+	"key_vault_access_spikes":        "keyVaultAccessSpikes",
+	"pim_role_activation_abuse":      "pimRoleActivationAbuse",
+}
+
+// Every marker the Azure filter adds and every marker an Azure 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 TestAzureCorrelationMarkerNames(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 azureConfig(t).Pipeline {
+		if !slices.Contains(stage.DataTypes, "azure") {
+			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, "azure") {
+			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 Azure filter does not add", path, marker)
+				}
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	expected := map[string]bool{}
+	for rule, marker := range azureHistoryMarkers {
+		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+azureHistoryMarkers[rule] != marker {
+				t.Errorf("%s reads %s, expected only %s%s", rule, marker, prefix, azureHistoryMarkers[rule])
+			}
+		}
+	}
+	for marker := range written {
+		if !expected[marker] {
+			t.Errorf("filter adds %s, which no Azure history rule counts", marker)
+		}
+	}
+	if len(written) != 10 || len(azureHistoryMarkers) != 10 {
+		t.Errorf("marker coverage: filter adds %d, table lists %d", len(written), len(azureHistoryMarkers))
+	}
+}