siem-backend/internal/detector/detector.go

package detector

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"log"
	"strconv"
	"strings"
	"time"

	"example.com/siem-greenfield/internal/clickhouse"
	"example.com/siem-greenfield/internal/config"
	"example.com/siem-greenfield/internal/postgres"
	"example.com/siem-greenfield/internal/rules"
)

func Run(ctx context.Context, cfg config.Config) error {
	pg, e := postgres.Open(ctx, cfg.PostgresURL)
	if e != nil {
		return e
	}
	defer pg.Close()
	if e := syncBuiltins(ctx, cfg, pg); e != nil {
		return fmt.Errorf("sync built-in rules: %w", e)
	}
	if e := pg.EnsureCustomRuleSet(ctx, cfg.TenantID); e != nil {
		return e
	}
	ch := clickhouse.New(cfg)
	ticker := time.NewTicker(cfg.DetectorInterval)
	defer ticker.Stop()
	run := func() {
		if e := runAll(ctx, cfg, pg, ch); e != nil {
			log.Printf("detector cycle: %v", e)
		}
	}
	run()
	for {
		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
			run()
		}
	}
}

func syncBuiltins(ctx context.Context, cfg config.Config, pg *postgres.Store) error {
	sets, e := rules.LoadDir(cfg.RulesDir)
	if e != nil {
		return e
	}
	keep := make([]string, 0, len(sets))
	for _, rs := range sets {
		keep = append(keep, rs.ID)
		if e := pg.SyncRuleSet(ctx, cfg.TenantID, rs); e != nil {
			return fmt.Errorf("%s: %w", rs.ID, e)
		}
	}
	if e := pg.PruneBuiltinRuleSets(ctx, cfg.TenantID, keep); e != nil {
		return fmt.Errorf("prune built-in rule sets: %w", e)
	}
	log.Printf("rule engine: synchronized %d built-in rule sets", len(sets))
	return nil
}

func runAll(ctx context.Context, cfg config.Config, pg *postgres.Store, ch *clickhouse.Client) error {
	enabled, e := pg.ListRules(ctx, cfg.TenantID, true)
	if e != nil {
		return e
	}
	end := time.Now().UTC()
	for _, sr := range enabled {
		q, _, e := rules.Compile(sr.Rule, cfg.ClickHouseDB, cfg.TenantID, end)
		if e != nil {
			log.Printf("rule %s compile: %v", sr.ID, e)
			continue
		}
		qctx, cancel := context.WithTimeout(ctx, 20*time.Second)
		rows, e := ch.QueryJSON(qctx, q)
		cancel()
		if e != nil {
			log.Printf("rule %s query: %v", sr.ID, e)
			continue
		}
		for _, row := range rows {
			host := str(row["host_name"])
			user := str(row["user_name"])
			ip := str(row["source_ip"])
			workstation := str(row["workstation"])
			ws := timeVal(row["window_start"], end.Add(-time.Duration(sr.WindowSeconds)*time.Second))
			we := timeVal(row["window_end"], end)
			suppressed, e := pg.IsSuppressed(ctx, cfg.TenantID, sr.ID, host, user, ip, we)
			if e != nil {
				log.Printf("rule %s suppression: %v", sr.ID, e)
				continue
			}
			if suppressed {
				continue
			}
			bucket := sr.SuppressSeconds
			if bucket <= 0 {
				bucket = sr.WindowSeconds
			}
			if bucket < 60 {
				bucket = 60
			}
			fp := fingerprint(sr.ID, host, user, ip, workstation, strconv.FormatInt(ws.Unix()/int64(bucket), 10))
			count := int64(num(row["cnt"]))
			eventCode := uint32(num(row["event_code"]))
			d := postgres.Detection{Fingerprint: fp, RuleID: sr.ID, RuleSetID: sr.RuleSetID, RuleName: sr.Title, Severity: sr.Severity, Hostname: host, UserName: user, SourceIP: ip, Workstation: workstation, EventCode: eventCode, Score: sr.Score, WindowStart: ws, WindowEnd: we, Summary: rules.RenderSummary(sr.Summary, row), Count: max64(1, count), Tags: sr.Tags, MITRE: sr.MITRE}
			if e := pg.UpsertDetection(ctx, d, cfg.TenantID); e != nil {
				log.Printf("rule %s detection: %v", sr.ID, e)
			}
		}
	}
	return nil
}

func str(v any) string {
	if v == nil {
		return ""
	}
	return fmt.Sprint(v)
}
func num(v any) float64 {
	switch x := v.(type) {
	case float64:
		return x
	case jsonNumber:
		return x.Float()
	default:
		f, _ := strconv.ParseFloat(fmt.Sprint(v), 64)
		return f
	}
}

type jsonNumber string

func (n jsonNumber) Float() float64 { f, _ := strconv.ParseFloat(string(n), 64); return f }
func timeVal(v any, d time.Time) time.Time {
	s := str(v)
	for _, layout := range []string{"2006-01-02 15:04:05.999999", "2006-01-02 15:04:05", "2006-01-02T15:04:05Z07:00"} {
		if t, e := time.Parse(layout, s); e == nil {
			return t.UTC()
		}
	}
	return d
}
func fingerprint(v ...string) string {
	h := sha256.Sum256([]byte(strings.Join(v, "|")))
	return hex.EncodeToString(h[:])
}
func max64(a, b int64) int64 {
	if a > b {
		return a
	}
	return b
}