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 }