Add Signal struct and enhance WAF detection handling in LogConsumerService

This update introduces a new Signal struct to encapsulate details related to WAF detections, including severity, tags, and context. The genWafDetectionSignal function has been implemented to generate signals based on attacked logs, improving the overall detection process. Additionally, the handling of bulk indexing for signals has been integrated into the existing workflow, ensuring that signals are properly indexed alongside events.

internal/model/log.go

@@ -46,6 +46,19 @@ type WafDetection struct {
 	CreatedAt int64 `json:"created_at"`
 }
 
+type Signal struct {
+	ID                string                 `json:"id"`
+	RuleKey           *RuleKey               `json:"ruleKey"`
+	Scope             *map[string]Scope      `json:"scope"`
+	Severity          int                    `json:"severity"`           // 信号严重程度，原始信号不携带，从mysql-rules表填充
+	Tags              []string               `json:"tags"`               // 信号标签，一期只有系统生成，内容为规则标签，原始信号不携带，从mysql-rules表填充
+	EventIDs          []string               `json:"eventIDs,omitempty"` // 维护当前signal被哪些events关联，内容为event.id；原始信号不携带，palace daemon存储时进行初始化(空)或append
+	Context           map[string]interface{} `json:"context"`            // 检测端只传递原始的key-value，多语言交给后面处理；只用于展示，不用做筛选搜索
+	CreatedAt         int64                  `json:"createdAt"`
+	IsWhitelistFilter bool                   `json:"isWhitelistFilter"` // 是否被白名单过滤
+	WhitelistIDs      []int64                `json:"whitelistIDs"`      // 记录匹配的白名单策略id
+}
+
 type Event struct {
 	ID           string                 `json:"id"`
 	Type         string                 `json:"type"`         // 事件类型（关联类型）

internal/service/log_consumer.go

@@ -26,6 +26,7 @@ const (
 	EsIndexWafDetections      = "waf-detections*"
 	EsIndexWafDetectionsAlias = "waf-detections"
 	ESIndexEvents             = "events"
+	ESIndexSignals            = "signals"
 )
 
 var scramAlgo = map[string]scram.Algorithm{
@@ -147,6 +148,57 @@ func (s *LogConsumerService) genWafDetection(wafDetectionMessage model.WafDetect
 	return wafDetection, nil
 }
 
+func (s *LogConsumerService) genWafDetectionSignal(wafDetectionMessage model.WafDetectionMessage, attackedLog model.WafDetectionAttackedLog, eventID string) (model.Signal, error) {
+	signal := model.Signal{
+		ID:      id.Str(),
+		RuleKey: &model.RuleKey{Name: attackedLog.RuleName, Category: "WAF"},
+		Scope: &map[string]model.Scope{
+			"cluster": {
+				Kind: "cluster",
+				ID:   wafDetectionMessage.WafDetectionMessageBasic.ClusterKey,
+				Name: wafDetectionMessage.WafDetectionMessageBasic.ClusterKey,
+			},
+			"namespace": {
+				Kind: "namespace",
+				ID:   wafDetectionMessage.WafDetectionMessageBasic.Namespace,
+				Name: wafDetectionMessage.WafDetectionMessageBasic.Namespace,
+			},
+			"resource": {
+				Kind: "resource",
+				ID:   "",
+				Name: fmt.Sprintf("%s(%s)", wafDetectionMessage.WafDetectionMessageBasic.ResName, wafDetectionMessage.WafDetectionMessageBasic.ResKind),
+			},
+		},
+		Severity: 6,
+		Tags:     []string{"waf"},
+		EventIDs: []string{eventID},
+		Context: map[string]interface{}{
+			"attack_ip":   attackedLog.AttackIP,
+			"attack_time": attackedLog.AttackTime,
+			"attack_url":  attackedLog.AttackedURL,
+			"attack_app":  attackedLog.AttackedApp,
+			"attack_load": attackedLog.AttackLoad,
+			"rule_name":   attackedLog.RuleName,
+			"action":      attackedLog.Action,
+			"waf_body": map[string]interface{}{
+				"type": "code",
+				"request": map[string]interface{}{
+					"action":  attackedLog.Action,
+					"req_pkg": attackedLog.ReqPkg,
+				},
+				"response": map[string]interface{}{
+					"content_type": attackedLog.RspContentType,
+					"res_pkg":      attackedLog.RspPkg,
+				},
+			},
+		},
+		CreatedAt:         attackedLog.AttackTime,
+		IsWhitelistFilter: false,
+		WhitelistIDs:      []int64{},
+	}
+	return signal, nil
+}
+
 func (s *LogConsumerService) genWafDetectionEvent(wafDetectionMessage model.WafDetectionMessage, attackedLog model.WafDetectionAttackedLog) (model.Event, error) {
 	event := model.Event{
 		ID:          id.Str(),
@@ -228,6 +280,9 @@ func (s *LogConsumerService) genWafDetectionEvent(wafDetectionMessage model.WafD
 				},
 			},
 		},
+		SignalsCount: map[int]int{
+			6: 1,
+		},
 	}
 	return event, nil
 }
@@ -262,8 +317,8 @@ func (s *LogConsumerService) Handle(ctx context.Context, message []byte) error {
 			continue
 		}
 
-		bulkIndexSignal := es.NewBulkIndexRequest().Index(EsIndexWafDetectionsAlias)
-		bulkableRequests = append(bulkableRequests, bulkIndexSignal.Id(wafDetection.WafDetectionAttackedLog.ID).Doc(wafDetection))
+		bulkIndexWaflog := es.NewBulkIndexRequest().Index(EsIndexWafDetectionsAlias)
+		bulkableRequests = append(bulkableRequests, bulkIndexWaflog.Id(wafDetection.WafDetectionAttackedLog.ID).Doc(wafDetection))
 
 		event, err := s.genWafDetectionEvent(WafDetectionMessage, WafDetectionMessage.AttackedLog[i])
 		if err != nil {
@@ -273,6 +328,14 @@ func (s *LogConsumerService) Handle(ctx context.Context, message []byte) error {
 		log.Info().Msgf("waf event: %+v", event)
 		bulkIndexEvent := es.NewBulkIndexRequest().Index(ESIndexEvents)
 		bulkableRequests = append(bulkableRequests, bulkIndexEvent.Id(event.ID).Doc(event))
+
+		signal, err := s.genWafDetectionSignal(WafDetectionMessage, WafDetectionMessage.AttackedLog[i], event.ID)
+		if err != nil {
+			log.Err(err).Str("message.Value", string(message)).Msg("gen waf detection signal fails")
+			continue
+		}
+		bulkIndexSignal := es.NewBulkIndexRequest().Index(ESIndexSignals)
+		bulkableRequests = append(bulkableRequests, bulkIndexSignal.Id(signal.ID).Doc(signal))
 	}
 
 	s.esStore.Save(ctx, bulkableRequests)