发布 0.1.0:通知渠道、告警规则、令牌版本与安全加固

2026-07-10 17:38:34 +08:00
parent 4af6a0ca92
commit dbba1f4905
78 changed files with 6898 additions and 551 deletions
@@ -57,6 +57,9 @@ type LogEventService struct {

 	relayPollTick    time.Duration // 订阅确认轮询间隔,零值用默认
 	relayPollTimeout time.Duration // 订阅确认轮询上限,零值用默认
+
+	alertMu     sync.Mutex         // 保护告警规则冷却表
+	alertSentAt map[uint]time.Time // 规则 ID → 上次告警时刻(阈值型规则冷却)
 }

 // NewLogEventService 组装依赖;调用 StartParser / StartCleanup 后台协程后生效。
@@ -80,37 +83,62 @@ type LogWebhookInfo struct {

 // EnsureSecret 为配置生成(或幂等返回)回传 secret。
 func (s *LogEventService) EnsureSecret(ctx context.Context, cfgID uint) (LogWebhookInfo, error) {
-	if err := s.requireConfig(ctx, cfgID); err != nil {
+	secret, err := generateWebhookSecret()
+	if err != nil {
 		return LogWebhookInfo{}, err
 	}
-	if info, ok, err := s.SecretInfo(ctx, cfgID); err != nil || ok {
-		return info, err
+	var info LogWebhookInfo
+	err = s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		var err error
+		info, err = ensureSecretTx(tx, cfgID, secret)
+		return err
+	})
+	if err != nil {
+		return LogWebhookInfo{}, fmt.Errorf("ensure webhook secret: %w", err)
 	}
+	return info, nil
+}
+
+func generateWebhookSecret() (string, error) {
 	buf := make([]byte, 32)
 	if _, err := rand.Read(buf); err != nil {
-		return LogWebhookInfo{}, fmt.Errorf("generate webhook secret: %w", err)
+		return "", fmt.Errorf("generate webhook secret: %w", err)
+	}
+	return hex.EncodeToString(buf), nil
+}
+
+func ensureSecretTx(tx *gorm.DB, cfgID uint, secret string) (LogWebhookInfo, error) {
+	if err := lockOciConfig(tx, cfgID); err != nil {
+		return LogWebhookInfo{}, err
+	}
+	if info, ok, err := secretInfoTx(tx, cfgID); err != nil || ok {
+		return info, err
 	}
-	secret := hex.EncodeToString(buf)
 	st := model.Setting{Key: secretKey(cfgID), Value: secret, UpdatedAt: time.Now()}
-	if err := s.db.WithContext(ctx).Save(&st).Error; err != nil {
+	if err := tx.Create(&st).Error; err != nil {
 		return LogWebhookInfo{}, fmt.Errorf("save webhook secret: %w", err)
 	}
 	return webhookInfo(secret, st.UpdatedAt), nil
 }

-// requireConfig 校验配置存在,不存在透传 gorm.ErrRecordNotFound(api 层映射 404)。
-func (s *LogEventService) requireConfig(ctx context.Context, cfgID uint) error {
-	var cfg model.OciConfig
-	if err := s.db.WithContext(ctx).Select("id").First(&cfg, cfgID).Error; err != nil {
-		return fmt.Errorf("find oci config %d: %w", cfgID, err)
-	}
-	return nil
-}
-
 // SecretInfo 查询配置是否已生成 secret;未生成时 ok 为 false。
 func (s *LogEventService) SecretInfo(ctx context.Context, cfgID uint) (LogWebhookInfo, bool, error) {
+	var info LogWebhookInfo
+	var ok bool
+	err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		if err := lockOciConfig(tx, cfgID); err != nil {
+			return err
+		}
+		var err error
+		info, ok, err = secretInfoTx(tx, cfgID)
+		return err
+	})
+	return info, ok, err
+}
+
+func secretInfoTx(tx *gorm.DB, cfgID uint) (LogWebhookInfo, bool, error) {
 	var st model.Setting
-	err := s.db.WithContext(ctx).First(&st, "key = ?", secretKey(cfgID)).Error
+	err := tx.First(&st, "key = ?", secretKey(cfgID)).Error
 	if err != nil {
 		if errors.Is(err, gorm.ErrRecordNotFound) {
 			return LogWebhookInfo{}, false, nil
@@ -154,19 +182,61 @@ func (s *LogEventService) ResolveSecret(ctx context.Context, secret string) (uin
 		return 0, false
 	}
 	for _, row := range rows {
-		if subtle.ConstantTimeCompare([]byte(row.Value), []byte(secret)) == 1 {
-			id, err := strconv.ParseUint(strings.TrimPrefix(row.Key, logWebhookSecretPrefix), 10, 64)
-			if err != nil {
-				return 0, false
-			}
-			return uint(id), true
+		if subtle.ConstantTimeCompare([]byte(row.Value), []byte(secret)) != 1 {
+			continue
 		}
+		id, ok := webhookConfigID(row.Key)
+		if !ok || !s.configExists(ctx, id) {
+			return 0, false
+		}
+		return id, true
 	}
 	return 0, false
 }

+// webhookConfigID 从 secret Setting 键解析租户配置 ID。
+func webhookConfigID(key string) (uint, bool) {
+	id, err := strconv.ParseUint(strings.TrimPrefix(key, logWebhookSecretPrefix), 10, 64)
+	return uint(id), err == nil && id > 0
+}
+
+// configExists 确认 secret 对应租户仍存在;数据库异常按认证失败处理。
+func (s *LogEventService) configExists(ctx context.Context, id uint) bool {
+	var cfg model.OciConfig
+	err := s.db.WithContext(ctx).Select("id").First(&cfg, id).Error
+	if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
+		log.Printf("resolve webhook config: %v", err)
+	}
+	return err == nil
+}
+
 // Ingest 落一条回传事件;MessageID 唯一索引冲突即静默忽略(at-least-once 幂等)。
 func (s *LogEventService) Ingest(ctx context.Context, cfgID uint, messageID string, payload []byte, truncated bool) error {
+	err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		if err := lockOciConfig(tx, cfgID); err != nil {
+			return err
+		}
+		return createLogEvent(tx, cfgID, messageID, payload, truncated)
+	})
+	if err != nil {
+		return fmt.Errorf("ingest log event: %w", err)
+	}
+	return nil
+}
+
+// lockOciConfig 与租户删除共用行锁,避免并发清理后写入孤儿数据。
+func lockOciConfig(tx *gorm.DB, cfgID uint) error {
+	var cfg model.OciConfig
+	err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
+		Select("id").First(&cfg, cfgID).Error
+	if err != nil {
+		return fmt.Errorf("lock oci config %d: %w", cfgID, err)
+	}
+	return nil
+}
+
+// createLogEvent 在已锁定租户的事务内幂等写入事件。
+func createLogEvent(tx *gorm.DB, cfgID uint, messageID string, payload []byte, truncated bool) error {
 	event := model.LogEvent{
 		OciConfigID: cfgID,
 		MessageID:   messageID,
@@ -174,13 +244,9 @@ func (s *LogEventService) Ingest(ctx context.Context, cfgID uint, messageID stri
 		Truncated:   truncated,
 		ReceivedAt:  time.Now(),
 	}
-	err := s.db.WithContext(ctx).
-		Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "message_id"}}, DoNothing: true}).
-		Create(&event).Error
-	if err != nil {
-		return fmt.Errorf("ingest log event: %w", err)
-	}
-	return nil
+	return tx.Clauses(clause.OnConflict{
+		Columns: []clause.Column{{Name: "message_id"}}, DoNothing: true,
+	}).Create(&event).Error
 }

 // LogEventQuery 是回传事件查询参数;CfgID 为 0 表示全部租户。
@@ -304,18 +370,44 @@ func (s *LogEventService) parseOnce(ctx context.Context) {
 		log.Printf("log event parse load: %v", err)
 		return
 	}
-	for i := range events {
-		e := &events[i]
-		parsed := parseLogEvent([]byte(e.Payload))
-		e.EventType, e.Source, e.SourceIP, e.EventTime =
-			parsed.EventType, parsed.Source, parsed.SourceIP, parsed.EventTime
-		e.Processed = true
-		if err := s.db.WithContext(ctx).Save(e).Error; err != nil {
-			log.Printf("log event parse save %d: %v", e.ID, err)
-			return
-		}
-		s.notifyCritical(ctx, e, parsed)
+	if len(events) == 0 {
+		return
 	}
+	rules := s.loadEnabledAlertRules(ctx)
+	for i := range events {
+		s.processLogEvent(ctx, &events[i], rules)
+	}
+}
+
+// processLogEvent 仅在条件更新命中原行后触发通知,删除并发胜出时静默跳过。
+func (s *LogEventService) processLogEvent(ctx context.Context, event *model.LogEvent, rules []model.AlertRule) {
+	parsed := parseLogEvent([]byte(event.Payload))
+	if !s.updateParsedEvent(ctx, event, parsed) {
+		return
+	}
+	s.notifyCritical(ctx, event, parsed)
+	s.matchAlertRules(ctx, rules, event, parsed)
+}
+
+// updateParsedEvent 用条件 UPDATE 禁止 Save 在删除后隐式重建事件。
+func (s *LogEventService) updateParsedEvent(ctx context.Context, event *model.LogEvent, parsed parsedEvent) bool {
+	updates := map[string]any{
+		"event_type": parsed.EventType, "source": parsed.Source,
+		"source_ip": parsed.SourceIP, "event_time": parsed.EventTime, "processed": true,
+	}
+	res := s.db.WithContext(ctx).Model(&model.LogEvent{}).
+		Where("id = ? AND processed = ?", event.ID, false).Updates(updates)
+	if res.Error != nil {
+		log.Printf("log event parse update %d: %v", event.ID, res.Error)
+		return false
+	}
+	if res.RowsAffected != 1 {
+		return false
+	}
+	event.EventType, event.Source, event.SourceIP, event.EventTime =
+		parsed.EventType, parsed.Source, parsed.SourceIP, parsed.EventTime
+	event.Processed = true
+	return true
 }

 // onsEnvelope 覆盖 ONS 消息与 CloudEvents 审计事件的常见字段;
@@ -443,6 +535,7 @@ func (s *LogEventService) cleanupOnce(ctx context.Context) {
 	if err := s.cleanup(ctx, logEventRetention, logEventMaxRows); err != nil {
 		log.Printf("log event cleanup: %v", err)
 	}
+	s.cleanupAlertHits(ctx)
 }

 // cleanup 先删过期记录,再对超量部分删最旧;阈值参数化便于测试。