v12 - feat(plugins/rule-flood-guard): auto-disable flooding correlation rules and notify via bell

.github/workflows/v12-deployment-pipeline.yml

@@ -391,6 +391,7 @@ jobs:
           - ad-audit
           - soar
           - playground
+          - rule-flood-guard
     steps:
       - name: Check out code into the right branch
         uses: actions/checkout@v4

backend/modules/eventprocessing/dto/correlation_rule_internal.go

@@ -0,0 +1,9 @@
+package dto
+
+type InternalDeactivateRuleRequest struct {
+	RuleName string `json:"ruleName" binding:"required"`
+}
+
+type InternalDeactivateRuleResponse struct {
+	Changed bool `json:"changed"`
+}

backend/modules/eventprocessing/handler/correlation_rule_internal.go

@@ -0,0 +1,65 @@
+package handler
+
+import (
+	"net/http"
+	"strings"
+
+	"github.com/gin-gonic/gin"
+	"github.com/utmstack/utmstack/backend/modules/eventprocessing/dto"
+)
+
+// @Summary     Deactivate a correlation rule by name (internal)
+// @Tags        Correlation Rules
+// @Accept      json
+// @Produce     json
+// @Param       input body dto.InternalDeactivateRuleRequest true "Rule to deactivate"
+// @Success     200 {object} dto.InternalDeactivateRuleResponse
+// @Failure     400 {object} map[string]string
+// @Failure     404 {object} map[string]string
+// @Failure     500 {object} map[string]string
+// @Router      /eventprocessing/internal/correlation-rule/deactivate [put]
+func (h *CorrelationRuleHandler) InternalDeactivate(c *gin.Context) {
+	var req dto.InternalDeactivateRuleRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+
+	ctx := c.Request.Context()
+
+	// RuleName filtering in List is a case-insensitive partial match, so the
+	// exact (case-insensitive) match is re-applied here to avoid disabling
+	// unrelated rules whose name merely contains the requested substring.
+	result, err := h.usecase.List(ctx, dto.CorrelationRuleFilters{RuleName: req.RuleName})
+	if err != nil {
+		writeCorrelationError(c, err)
+		return
+	}
+
+	var matches []dto.CorrelationRuleResponse
+	for _, r := range result.Items {
+		if strings.EqualFold(r.RuleName, req.RuleName) {
+			matches = append(matches, r)
+		}
+	}
+	if len(matches) == 0 {
+		c.JSON(http.StatusNotFound, gin.H{"error": "correlation rule not found"})
+		return
+	}
+
+	// Two or more rules sharing an exact display name is an accepted edge
+	// case (see design's Open Questions) — disable every exact match.
+	changed := false
+	for _, r := range matches {
+		if !r.RuleActive {
+			continue
+		}
+		if err := h.usecase.SetActive(ctx, r.RelPath, false); err != nil {
+			writeCorrelationError(c, err)
+			return
+		}
+		changed = true
+	}
+
+	c.JSON(http.StatusOK, dto.InternalDeactivateRuleResponse{Changed: changed})
+}

backend/modules/eventprocessing/routes.go

@@ -38,6 +38,11 @@ func RegisterRoutes(api *gin.RouterGroup, m *Module, userAuth gin.HandlerFunc) {
 	cr.GET("/find", read, crh.GetByID)
 	cr.DELETE("", write, crh.Delete)
 
+	// Internal-only: rule-flood-guard plugin disables a rule it identified as
+	// flooding the alerts list, by display name.
+	icr := g.Group("/internal/correlation-rule", middleware.RequireInternal())
+	icr.PUT("/deactivate", crh.InternalDeactivate)
+
 	// Filters (file-backed, pipeline: YAML). Identity = relPath query param.
 	f := g.Group("/filters")
 	f.POST("", write, fh.Create)

backend/modules/notifications/domain/enums.go

@@ -18,7 +18,7 @@ const (
 func (s NotificationSource) Valid() bool {
 	switch s {
 	case SourceAS400, SourceBitDefender, SourceAWS, SourceAzure,
-		SourceOffice365, SourceSophos, SourceGoogle, SourceEmailSetting, SourceAlerts:
+		SourceOffice365, SourceSophos, SourceGoogle, SourceEmailSetting, SourceAlerts, SourceSystem:
 		return true
 	}
 	return false

backend/modules/notifications/dto/notification.go

@@ -7,7 +7,7 @@ import (
 )
 
 type CreateNotificationRequest struct {
-	Source  domain.NotificationSource `json:"source"  binding:"required,oneof=AS400 BIT_DEFENDER AWS AZURE OFFICE_365 SOPHOS GOOGLE EMAIL_SETTING ALERTS"`
+	Source  domain.NotificationSource `json:"source"  binding:"required,oneof=AS400 BIT_DEFENDER AWS AZURE OFFICE_365 SOPHOS GOOGLE EMAIL_SETTING ALERTS SYSTEM"`
 	Type    domain.NotificationType   `json:"type"    binding:"required,oneof=INFO WARNING ERROR"`
 	Message string                    `json:"message" binding:"required,min=1"`
 }

plugins/rule-flood-guard/README.md

@@ -0,0 +1,35 @@
+# UTMStack Rule Flood Guard Plugin
+
+Safety net against alert fatigue: when a correlation rule floods the alerts
+list with individual, non-deduplicated alerts, this plugin automatically
+disables the rule and notifies the user through the notification bell,
+suggesting they mark the alerts as false positives (Alert Tag Rules) or
+fine-tune the rule (`deduplicateBy`/`groupBy`).
+
+It runs as its own independent plugin, separate from `plugins/alerts`.
+
+## Configuration
+
+Settings live in `system_plugins_rule-flood-guard.yaml`, in the same pipeline
+config folder as the other plugins' config files. It reloads automatically
+when the file changes — no restart needed.
+
+```yaml
+plugins:
+  rule-flood-guard:
+    enabled: true
+    threshold: 50
+    windowHours: 24
+    intervalSeconds: 300
+```
+
+| Field | Default | Meaning |
+|---|---|---|
+| `enabled` | `true` | Turns the guard on or off. |
+| `threshold` | `50` | How many alerts trigger the auto-disable. |
+| `windowHours` | `24` | Time window used to count alerts. |
+| `intervalSeconds` | `300` | How often the guard checks. |
+
+If the file doesn't exist, the guard just runs with these defaults.
+
+

plugins/rule-flood-guard/aggregation.go

@@ -0,0 +1,83 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+
+	sdkos "github.com/threatwinds/go-sdk/os"
+)
+
+const alertIndex = "v11-alert-*"
+
+const falsePositiveTag = "False positive"
+
+const ruleNameAgg = "by_rule_name"
+
+type ruleBucket struct {
+	RuleName string
+	Count    int64
+}
+
+func buildFloodQuery(window time.Duration, now time.Time) map[string]any {
+	gte := now.Add(-window).UTC().Format(time.RFC3339)
+	return map[string]any{
+		"size": 0,
+		"query": map[string]any{
+			"bool": map[string]any{
+				"must": []map[string]any{
+					{"term": map[string]any{"status": 2}},
+					{"range": map[string]any{"@timestamp": map[string]any{"gte": gte}}},
+				},
+				"must_not": []map[string]any{
+					{"term": map[string]any{"tags": falsePositiveTag}},
+					{"exists": map[string]any{"field": "parentId.keyword"}},
+				},
+			},
+		},
+		"aggs": map[string]any{
+			ruleNameAgg: map[string]any{
+				"terms": map[string]any{
+					"field": "name.keyword",
+					"size":  1000,
+				},
+			},
+		},
+	}
+}
+
+type termsBucketsResponse struct {
+	Buckets []struct {
+		Key      string `json:"key"`
+		DocCount int64  `json:"doc_count"`
+	} `json:"buckets"`
+}
+
+func parseRuleBuckets(aggs map[string]interface{}) ([]ruleBucket, error) {
+	raw, ok := aggs[ruleNameAgg]
+	if !ok {
+		return nil, nil
+	}
+	b, err := json.Marshal(raw)
+	if err != nil {
+		return nil, err
+	}
+	var parsed termsBucketsResponse
+	if err := json.Unmarshal(b, &parsed); err != nil {
+		return nil, err
+	}
+	buckets := make([]ruleBucket, 0, len(parsed.Buckets))
+	for _, bucket := range parsed.Buckets {
+		buckets = append(buckets, ruleBucket{RuleName: bucket.Key, Count: bucket.DocCount})
+	}
+	return buckets, nil
+}
+
+func searchRuleBuckets(ctx context.Context, window time.Duration) ([]ruleBucket, error) {
+	query := buildFloodQuery(window, time.Now())
+	result, err := sdkos.RawSearch(ctx, []string{alertIndex}, query)
+	if err != nil {
+		return nil, err
+	}
+	return parseRuleBuckets(result.Aggregations)
+}

plugins/rule-flood-guard/backend.go

@@ -0,0 +1,108 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+
+	"github.com/threatwinds/go-sdk/catcher"
+)
+
+const notificationMessageTemplate = "Correlation rule '%s' generated over %d open, un-deduplicated alerts in the last 24h and was automatically disabled to prevent alert flooding. If this volume is expected, use an Alert Tag Rule to mark it 'False positive', or add deduplicateBy/groupBy to the rule, then re-enable it."
+
+type backendClient struct {
+	baseURL     string
+	internalKey string
+	httpClient  *http.Client
+}
+
+func newBackendClient(baseURL, internalKey string) *backendClient {
+	return &backendClient{
+		baseURL:     baseURL,
+		internalKey: internalKey,
+		httpClient:  &http.Client{Timeout: 15 * time.Second},
+	}
+}
+
+type deactivateRequest struct {
+	RuleName string `json:"ruleName"`
+}
+
+type deactivateResponse struct {
+	Changed bool `json:"changed"`
+}
+
+func (c *backendClient) Deactivate(ctx context.Context, ruleName string) (bool, error) {
+	payload, err := json.Marshal(deactivateRequest{RuleName: ruleName})
+	if err != nil {
+		return false, err
+	}
+	url := c.baseURL + "/api/v1/eventprocessing/internal/correlation-rule/deactivate"
+	req, err := http.NewRequestWithContext(ctx, http.MethodPut, url, bytes.NewReader(payload))
+	if err != nil {
+		return false, err
+	}
+	req.Header.Set("X-Internal-Key", c.internalKey)
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return false, err
+	}
+	defer func() { _ = resp.Body.Close() }()
+
+	if resp.StatusCode >= 400 {
+		body, _ := io.ReadAll(resp.Body)
+		return false, catcher.Error("deactivate call returned error status", nil, map[string]any{
+			"status": resp.StatusCode, "body": string(body), "ruleName": ruleName,
+		})
+	}
+
+	var out deactivateResponse
+	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+		return false, err
+	}
+	return out.Changed, nil
+}
+
+type notifyRequest struct {
+	Source  string `json:"source"`
+	Type    string `json:"type"`
+	Message string `json:"message"`
+}
+
+func (c *backendClient) Notify(ctx context.Context, message string) error {
+	payload, err := json.Marshal(notifyRequest{Source: "SYSTEM", Type: "WARNING", Message: message})
+	if err != nil {
+		return err
+	}
+	url := c.baseURL + "/api/v1/notifications"
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(payload))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("X-Internal-Key", c.internalKey)
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = resp.Body.Close() }()
+
+	if resp.StatusCode >= 400 {
+		body, _ := io.ReadAll(resp.Body)
+		return catcher.Error("notify call returned error status", nil, map[string]any{
+			"status": resp.StatusCode, "body": string(body),
+		})
+	}
+	return nil
+}
+
+func floodNotificationMessage(ruleName string, threshold int64) string {
+	return fmt.Sprintf(notificationMessageTemplate, ruleName, threshold)
+}