Bugfix 2.0

internal/api/server.go

@@ -82,7 +82,7 @@ func Run(ctx context.Context, cfg config.Config) error {
 func summary(w http.ResponseWriter, r *http.Request, cfg config.Config, pg *postgres.Store, ch *clickhouse.Client) {
 	ctx, cancel := context.WithTimeout(r.Context(), 8*time.Second)
 	defer cancel()
-	q := fmt.Sprintf(`SELECT uniqExactMerge(cnt_state) events_24h, uniqExact(host_name) active_hosts FROM siem.events_5m WHERE tenant_id=%s AND bucket>=now()-INTERVAL 24 HOUR`, clickhouse.Q(cfg.TenantID))
+	q := fmt.Sprintf(`SELECT uniqExactMerge(cnt_state) events_24h, uniqExact(host_name) active_hosts FROM %s.events_5m WHERE tenant_id=%s AND bucket>=now()-INTERVAL 24 HOUR`, clickhouse.Ident(cfg.ClickHouseDB), clickhouse.Q(cfg.TenantID))
 	rows, e := ch.QueryJSON(ctx, q)
 	if e != nil {
 		j(w, 500, map[string]string{"error": e.Error()})
@@ -118,7 +118,7 @@ func events(w http.ResponseWriter, r *http.Request, cfg config.Config, ch *click
 			where = append(where, "event_code="+v)
 		}
 	}
-	q := fmt.Sprintf(`SELECT event_uid,event_time,host_name,channel,event_code,category,action,outcome,severity,user_name,subject_user,target_user,source_ip,workstation,process_path,message,raw_object_key,raw_index FROM siem.events WHERE %s ORDER BY event_time DESC, ingest_time DESC LIMIT 1 BY event_uid LIMIT %d`, strings.Join(where, " AND "), limit)
+	q := fmt.Sprintf(`SELECT event_uid,event_time,host_name,channel,event_code,category,action,outcome,severity,user_name,subject_user,target_user,source_ip,workstation,process_path,message,raw_object_key,raw_index FROM %s.events WHERE %s ORDER BY event_time DESC, ingest_time DESC LIMIT 1 BY event_uid LIMIT %d`, clickhouse.Ident(cfg.ClickHouseDB), strings.Join(where, " AND "), limit)
 	ctx, cancel := context.WithTimeout(r.Context(), 12*time.Second)
 	defer cancel()
 	rows, e := ch.QueryJSON(ctx, q)

internal/clickhouse/client.go

@@ -17,12 +17,12 @@ import (
 )
 
 type Client struct {
-	base, user, pass string
-	hc               *http.Client
+	base, db, user, pass string
+	hc                   *http.Client
 }
 
 func New(cfg config.Config) *Client {
-	return &Client{base: cfg.ClickHouseURL, user: cfg.ClickHouseUser, pass: cfg.ClickHousePassword, hc: &http.Client{Timeout: 30 * time.Second}}
+	return &Client{base: cfg.ClickHouseURL, db: cfg.ClickHouseDB, user: cfg.ClickHouseUser, pass: cfg.ClickHousePassword, hc: &http.Client{Timeout: 30 * time.Second}}
 }
 func (c *Client) request(ctx context.Context, query string, body io.Reader) (*http.Response, error) {
 	u := c.base + "/?query=" + url.QueryEscape(query)
@@ -57,7 +57,7 @@ func (c *Client) InsertEvents(ctx context.Context, events []contracts.CanonicalE
 			return e
 		}
 	}
-	q := `INSERT INTO siem.events FORMAT JSONEachRow`
+	q := `INSERT INTO ` + Ident(c.db) + `.events FORMAT JSONEachRow`
 	r, e := c.request(ctx, q, &b)
 	if e != nil {
 		return e
@@ -95,4 +95,5 @@ func (c *Client) QueryJSON(ctx context.Context, q string) ([]map[string]any, err
 	}
 	return out, s.Err()
 }
-func Q(s string) string { return "'" + strings.ReplaceAll(s, "'", "''") + "'" }
+func Q(s string) string     { return "'" + strings.ReplaceAll(s, "'", "''") + "'" }
+func Ident(s string) string { return "`" + strings.ReplaceAll(s, "`", "``") + "`" }

internal/config/config.go

@@ -11,6 +11,7 @@ type Config struct {
 	ServiceAddr        string
 	PostgresURL        string
 	ClickHouseURL      string
+	ClickHouseDB       string
 	ClickHouseUser     string
 	ClickHousePassword string
 	KafkaBrokers       []string
@@ -23,7 +24,6 @@ type Config struct {
 	RawArchiveEnabled  bool
 	RawSpoolDir        string
 	RawRetention       time.Duration
-	EventRetention     time.Duration
 	DetectorInterval   time.Duration
 	DetectorLookback   time.Duration
 	UIQueryLimit       int
@@ -36,6 +36,7 @@ func Load() Config {
 		ServiceAddr:        env("SERVICE_ADDR", ":8080"),
 		PostgresURL:        env("POSTGRES_URL", "postgres://siem:siem@postgres:5432/siem?sslmode=disable"),
 		ClickHouseURL:      strings.TrimRight(env("CLICKHOUSE_URL", "http://clickhouse:8123"), "/"),
+		ClickHouseDB:       env("CLICKHOUSE_DB", "siem"),
 		ClickHouseUser:     env("CLICKHOUSE_USER", "siem"),
 		ClickHousePassword: env("CLICKHOUSE_PASSWORD", "siem"),
 		KafkaBrokers:       splitCSV(env("KAFKA_BROKERS", "redpanda:9092")),
@@ -48,7 +49,6 @@ func Load() Config {
 		RawArchiveEnabled:  envBool("RAW_ARCHIVE_ENABLED", true),
 		RawSpoolDir:        env("RAW_SPOOL_DIR", "/var/spool/siem-raw"),
 		RawRetention:       envDuration("RAW_RETENTION", 30*24*time.Hour),
-		EventRetention:     envDuration("EVENT_RETENTION", 90*24*time.Hour),
 		DetectorInterval:   envDuration("DETECTOR_INTERVAL", 30*time.Second),
 		DetectorLookback:   envDuration("DETECTOR_LOOKBACK", 20*time.Minute),
 		UIQueryLimit:       envInt("UI_QUERY_LIMIT", 500),

internal/detector/detector.go

@@ -50,7 +50,7 @@ func Run(ctx context.Context, cfg config.Config) error {
 func runAll(ctx context.Context, cfg config.Config, pg *postgres.Store, ch *clickhouse.Client) error {
 	end := time.Now().UTC()
 	start := end.Add(-cfg.DetectorLookback)
-	for _, r := range rules() {
+	for _, r := range rules(cfg.ClickHouseDB) {
 		q := r.query(start, end, cfg.TenantID)
 		rows, e := ch.QueryJSON(ctx, q)
 		if e != nil {
@@ -73,39 +73,40 @@ func runAll(ctx context.Context, cfg config.Config, pg *postgres.Store, ch *clic
 	}
 	return nil
 }
-func rules() []rule {
+func rules(db string) []rule {
+	table := clickhouse.Ident(db) + ".events"
 	return []rule{
-		{name: "audit_log_cleared", severity: "critical", eventCode: 1102, score: 9.8, query: simpleEvent(1102, 1), summary: func(m map[string]any) string {
+		{name: "audit_log_cleared", severity: "critical", eventCode: 1102, score: 9.8, query: simpleEvent(table, 1102, 1), summary: func(m map[string]any) string {
 			return fmt.Sprintf("Audit-Log auf %s wurde gelöscht", str(m["host_name"]))
 		}},
-		{name: "service_installed", severity: "high", eventCode: 7045, score: 8.0, query: simpleEvent(7045, 1), summary: func(m map[string]any) string {
+		{name: "service_installed", severity: "high", eventCode: 7045, score: 8.0, query: simpleEvent(table, 7045, 1), summary: func(m map[string]any) string {
 			return fmt.Sprintf("Neuer Dienst auf %s installiert", str(m["host_name"]))
 		}},
 		{name: "account_lockout", severity: "medium", eventCode: 4740, score: 5.5, query: func(s, e time.Time, t string) string {
-			return fmt.Sprintf(`SELECT host_name, target_user AS user_name, '' AS source_ip, workstation, uniqExact(event_uid) cnt, min(event_time) window_start, max(event_time) window_end FROM siem.events WHERE tenant_id=%s AND event_time>=%s AND event_time<%s AND event_code=4740 GROUP BY host_name,user_name,workstation HAVING cnt>=1`, clickhouse.Q(t), clickhouse.Q(ts(s)), clickhouse.Q(ts(e)))
+			return fmt.Sprintf(`SELECT host_name, target_user AS user_name, '' AS source_ip, workstation, uniqExact(event_uid) cnt, min(event_time) window_start, max(event_time) window_end FROM %s WHERE tenant_id=%s AND event_time>=%s AND event_time<%s AND event_code=4740 GROUP BY host_name,user_name,workstation HAVING cnt>=1`, table, clickhouse.Q(t), clickhouse.Q(ts(s)), clickhouse.Q(ts(e)))
 		}, summary: func(m map[string]any) string {
 			return fmt.Sprintf("Account-Lockout: %s; Caller %s; DC/Host %s (%d×)", str(m["user_name"]), fallback(str(m["workstation"]), "unbekannt"), str(m["host_name"]), int64(num(m["cnt"])))
 		}},
 		{name: "failed_logon_burst", severity: "high", eventCode: 4625, score: 7.5, query: func(s, e time.Time, t string) string {
-			return fmt.Sprintf(`SELECT host_name, target_user AS user_name, source_ip, workstation, uniqExact(event_uid) cnt, min(event_time) window_start, max(event_time) window_end FROM siem.events WHERE tenant_id=%s AND event_time>=%s AND event_time<%s AND event_code=4625 AND target_user!='' GROUP BY host_name,user_name,source_ip,workstation HAVING cnt>=20`, clickhouse.Q(t), clickhouse.Q(ts(s)), clickhouse.Q(ts(e)))
+			return fmt.Sprintf(`SELECT host_name, target_user AS user_name, source_ip, workstation, uniqExact(event_uid) cnt, min(event_time) window_start, max(event_time) window_end FROM %s WHERE tenant_id=%s AND event_time>=%s AND event_time<%s AND event_code=4625 AND target_user!='' GROUP BY host_name,user_name,source_ip,workstation HAVING cnt>=20`, table, clickhouse.Q(t), clickhouse.Q(ts(s)), clickhouse.Q(ts(e)))
 		}, summary: func(m map[string]any) string {
 			return fmt.Sprintf("%d fehlgeschlagene Logons für %s auf %s", int64(num(m["cnt"])), str(m["user_name"]), str(m["host_name"]))
 		}},
 		{name: "password_spray", severity: "high", eventCode: 4625, score: 8.5, query: func(s, e time.Time, t string) string {
-			return fmt.Sprintf(`SELECT '' AS host_name, '' AS user_name, source_ip, '' AS workstation, uniqExact(target_user) users, uniqExact(event_uid) cnt, min(event_time) window_start, max(event_time) window_end FROM siem.events WHERE tenant_id=%s AND event_time>=%s AND event_time<%s AND event_code=4625 AND source_ip!='' AND target_user!='' GROUP BY source_ip HAVING users>=10 AND cnt>=20`, clickhouse.Q(t), clickhouse.Q(ts(s)), clickhouse.Q(ts(e)))
+			return fmt.Sprintf(`SELECT '' AS host_name, '' AS user_name, source_ip, '' AS workstation, uniqExact(target_user) users, uniqExact(event_uid) cnt, min(event_time) window_start, max(event_time) window_end FROM %s WHERE tenant_id=%s AND event_time>=%s AND event_time<%s AND event_code=4625 AND source_ip!='' AND target_user!='' GROUP BY source_ip HAVING users>=10 AND cnt>=20`, table, clickhouse.Q(t), clickhouse.Q(ts(s)), clickhouse.Q(ts(e)))
 		}, summary: func(m map[string]any) string {
 			return fmt.Sprintf("Password-Spray von %s gegen %.0f Benutzer (%d Versuche)", str(m["source_ip"]), num(m["users"]), int64(num(m["cnt"])))
 		}},
 		{name: "privileged_group_change", severity: "critical", eventCode: 4728, score: 9.2, query: func(s, e time.Time, t string) string {
-			return fmt.Sprintf(`SELECT host_name, target_user AS user_name, '' AS source_ip, workstation, uniqExact(event_uid) cnt, min(event_time) window_start, max(event_time) window_end FROM siem.events WHERE tenant_id=%s AND event_time>=%s AND event_time<%s AND event_code IN (4728,4732,4756) GROUP BY host_name,user_name,workstation HAVING cnt>=1`, clickhouse.Q(t), clickhouse.Q(ts(s)), clickhouse.Q(ts(e)))
+			return fmt.Sprintf(`SELECT host_name, target_user AS user_name, '' AS source_ip, workstation, uniqExact(event_uid) cnt, min(event_time) window_start, max(event_time) window_end FROM %s WHERE tenant_id=%s AND event_time>=%s AND event_time<%s AND event_code IN (4728,4732,4756) GROUP BY host_name,user_name,workstation HAVING cnt>=1`, table, clickhouse.Q(t), clickhouse.Q(ts(s)), clickhouse.Q(ts(e)))
 		}, summary: func(m map[string]any) string {
 			return fmt.Sprintf("Privilegierte Gruppenmitgliedschaft geändert: %s auf %s", str(m["user_name"]), str(m["host_name"]))
 		}},
 	}
 }
-func simpleEvent(id uint32, min int) func(time.Time, time.Time, string) string {
+func simpleEvent(table string, id uint32, min int) func(time.Time, time.Time, string) string {
 	return func(s, e time.Time, t string) string {
-		return fmt.Sprintf(`SELECT host_name, '' AS user_name, '' AS source_ip, '' AS workstation, uniqExact(event_uid) cnt, min(event_time) window_start, max(event_time) window_end FROM siem.events WHERE tenant_id=%s AND event_time>=%s AND event_time<%s AND event_code=%d GROUP BY host_name HAVING cnt>=%d`, clickhouse.Q(t), clickhouse.Q(ts(s)), clickhouse.Q(ts(e)), id, min)
+		return fmt.Sprintf(`SELECT host_name, '' AS user_name, '' AS source_ip, '' AS workstation, uniqExact(event_uid) cnt, min(event_time) window_start, max(event_time) window_end FROM %s WHERE tenant_id=%s AND event_time>=%s AND event_time<%s AND event_code=%d GROUP BY host_name HAVING cnt>=%d`, table, clickhouse.Q(t), clickhouse.Q(ts(s)), clickhouse.Q(ts(e)), id, min)
 	}
 }
 func ts(t time.Time) string { return t.UTC().Format("2006-01-02 15:04:05.000") }
@@ -143,7 +144,12 @@ func fingerprint(v ...string) string {
 	h := sha256.Sum256([]byte(strings.Join(v, "|")))
 	return hex.EncodeToString(h[:])
 }
-func fallback(v, d string) string { if strings.TrimSpace(v) == "" { return d }; return v }
+func fallback(v, d string) string {
+	if strings.TrimSpace(v) == "" {
+		return d
+	}
+	return v
+}
 func max64(a, b int64) int64 {
 	if a > b {
 		return a