package waf import ( "context" "log/slog" "sync" "sync/atomic" "time" "github.com/jackc/pgx/v5/pgxpool" ) // Alert represents a single WAF rule match that was logged. type Alert struct { ID int64 `json:"id"` DomainID *int64 `json:"domain_id,omitempty"` Hostname string `json:"hostname"` ClientIP string `json:"client_ip"` Method string `json:"method"` URI string `json:"uri"` RuleID int `json:"rule_id"` RuleMsg string `json:"rule_msg"` Severity string `json:"severity"` Action string `json:"action"` // "detected" | "blocked" CreatedAt time.Time `json:"created_at"` } // AlertWriter accepts Alert values via a buffered channel and writes // them to PostgreSQL asynchronously so SPOE handling stays low-latency. type AlertWriter struct { pool *pgxpool.Pool ch chan Alert stop chan struct{} done chan struct{} closeOnce sync.Once closed atomic.Bool } // NewAlertWriter creates an AlertWriter and starts its background goroutine. // bufSize is the number of unwritten alerts that can queue before drops. func NewAlertWriter(pool *pgxpool.Pool, bufSize int) *AlertWriter { aw := &AlertWriter{ pool: pool, ch: make(chan Alert, bufSize), stop: make(chan struct{}), done: make(chan struct{}), } go aw.run() return aw } // Send enqueues an alert. Drops silently if the channel is full (or the // writer is closing) to avoid slowing down / panicking SPOE handling. func (aw *AlertWriter) Send(a Alert) { if aw.closed.Load() { return } select { case aw.ch <- a: default: slog.Warn("waf: alert channel full — dropping alert", "host", a.Hostname, "rule", a.RuleID) } } // Close stops the writer and flushes buffered alerts (best-effort). // Safe to call multiple times. The channel is never closed → Send never // panics even if it races with Close. func (aw *AlertWriter) Close() { aw.closeOnce.Do(func() { aw.closed.Store(true) close(aw.stop) }) <-aw.done } func (aw *AlertWriter) run() { defer close(aw.done) for { select { case a := <-aw.ch: aw.write(a) case <-aw.stop: // Restliche gepufferte Alerts noch wegschreiben, dann Ende. for { select { case a := <-aw.ch: aw.write(a) default: return } } } } } func (aw *AlertWriter) write(a Alert) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() // Resolve domain_id from hostname (best-effort). var domainID *int64 var id int64 if err := aw.pool.QueryRow(ctx, `SELECT id FROM domains WHERE name = $1 AND active = true LIMIT 1`, a.Hostname, ).Scan(&id); err == nil { domainID = &id } if _, err := aw.pool.Exec(ctx, ` INSERT INTO waf_alerts (domain_id, hostname, client_ip, method, uri, rule_id, rule_msg, severity, action) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9) `, domainID, a.Hostname, a.ClientIP, a.Method, a.URI, a.RuleID, a.RuleMsg, a.Severity, a.Action, ); err != nil { slog.Warn("waf: write alert to db failed", "error", err) } }