feat(waf): Alerts — Regelübereinstimmungen in DB + UI — v1.2.77
- Migration 0038: waf_alerts-Tabelle - AlertWriter (Buffered-Channel → async DB-Write) - SPOE: MatchedRules → sendAlert() nach ProcessRequestHeaders() - API: GET /waf/alerts + DELETE /waf/alerts - WAF-Page: Tabs Domains | Alarme; Alarme-Tabelle mit Rule-ID, Severity, Aktion (Detected/Blocked), URI, Client-IP + Purge-Button Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
20
internal/database/migrations/0038_waf_alerts.sql
Normal file
20
internal/database/migrations/0038_waf_alerts.sql
Normal file
@@ -0,0 +1,20 @@
|
||||
-- +goose Up
|
||||
CREATE TABLE IF NOT EXISTS waf_alerts (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
domain_id BIGINT REFERENCES domains(id) ON DELETE CASCADE,
|
||||
hostname TEXT NOT NULL,
|
||||
client_ip TEXT NOT NULL,
|
||||
method TEXT NOT NULL,
|
||||
uri TEXT NOT NULL,
|
||||
rule_id INT NOT NULL DEFAULT 0,
|
||||
rule_msg TEXT NOT NULL DEFAULT '',
|
||||
severity TEXT NOT NULL DEFAULT '',
|
||||
action TEXT NOT NULL, -- 'detected' | 'blocked'
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS waf_alerts_domain_created ON waf_alerts(domain_id, created_at DESC);
|
||||
CREATE INDEX IF NOT EXISTS waf_alerts_created ON waf_alerts(created_at DESC);
|
||||
|
||||
-- +goose Down
|
||||
DROP TABLE IF EXISTS waf_alerts;
|
||||
@@ -36,6 +36,8 @@ func (h *WafHandler) Register(rg *gin.RouterGroup) {
|
||||
g.GET("/configs", h.List)
|
||||
g.GET("/configs/:domain_id", h.Get)
|
||||
g.PUT("/configs/:domain_id", h.Upsert)
|
||||
g.GET("/alerts", h.ListAlerts)
|
||||
g.DELETE("/alerts", h.PurgeAlerts)
|
||||
}
|
||||
|
||||
// List returns all WAF configs.
|
||||
@@ -134,6 +136,48 @@ func (h *WafHandler) Upsert(c *gin.Context) {
|
||||
c.JSON(http.StatusOK, gin.H{"config": result})
|
||||
}
|
||||
|
||||
// ListAlerts returns recent WAF alerts. Optional: ?domain_id=X&limit=N
|
||||
func (h *WafHandler) ListAlerts(c *gin.Context) {
|
||||
var domainID *int64
|
||||
if v := c.Query("domain_id"); v != "" {
|
||||
id, err := strconv.ParseInt(v, 10, 64)
|
||||
if err != nil {
|
||||
response.BadRequest(c, errors.New("invalid domain_id"))
|
||||
return
|
||||
}
|
||||
domainID = &id
|
||||
}
|
||||
limit := 200
|
||||
if v := c.Query("limit"); v != "" {
|
||||
if n, err := strconv.Atoi(v); err == nil && n > 0 {
|
||||
limit = n
|
||||
}
|
||||
}
|
||||
alerts, err := h.Repo.ListAlerts(c.Request.Context(), domainID, limit)
|
||||
if err != nil {
|
||||
response.Internal(c, err)
|
||||
return
|
||||
}
|
||||
response.OK(c, gin.H{"alerts": alerts})
|
||||
}
|
||||
|
||||
// PurgeAlerts deletes old WAF alerts. Optional: ?days=N (default 30)
|
||||
func (h *WafHandler) PurgeAlerts(c *gin.Context) {
|
||||
days := 30
|
||||
if v := c.Query("days"); v != "" {
|
||||
if n, err := strconv.Atoi(v); err == nil && n > 0 {
|
||||
days = n
|
||||
}
|
||||
}
|
||||
if err := h.Repo.PurgeAlerts(c.Request.Context(), days); err != nil {
|
||||
response.Internal(c, err)
|
||||
return
|
||||
}
|
||||
_ = h.Audit.Log(c.Request.Context(), actorOf(c), "waf.alerts.purge",
|
||||
"", gin.H{"days": days}, h.NodeID)
|
||||
response.OK(c, gin.H{"ok": true, "days": days})
|
||||
}
|
||||
|
||||
// defaultConfig returns a sensible disabled default for a domain
|
||||
// that has no WAF config row yet.
|
||||
func defaultConfig(domainID int64) models.WafConfig {
|
||||
|
||||
@@ -113,6 +113,73 @@ func (r *Repo) ListEnabled(ctx context.Context) ([]models.WafConfig, error) {
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// WafAlert mirrors the waf_alerts DB row.
|
||||
type WafAlert 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"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
}
|
||||
|
||||
// ListAlerts returns recent WAF alerts, optionally filtered by domain_id.
|
||||
func (r *Repo) ListAlerts(ctx context.Context, domainID *int64, limit int) ([]WafAlert, error) {
|
||||
if limit <= 0 || limit > 1000 {
|
||||
limit = 200
|
||||
}
|
||||
var rows interface{ Next() bool; Scan(...any) error; Close(); Err() error }
|
||||
var err error
|
||||
if domainID != nil {
|
||||
rows2, e := r.Pool.Query(ctx, `
|
||||
SELECT id, domain_id, hostname, client_ip, method, uri,
|
||||
rule_id, rule_msg, severity, action, created_at
|
||||
FROM waf_alerts
|
||||
WHERE domain_id = $1
|
||||
ORDER BY created_at DESC LIMIT $2
|
||||
`, *domainID, limit)
|
||||
rows, err = rows2, e
|
||||
} else {
|
||||
rows2, e := r.Pool.Query(ctx, `
|
||||
SELECT id, domain_id, hostname, client_ip, method, uri,
|
||||
rule_id, rule_msg, severity, action, created_at
|
||||
FROM waf_alerts
|
||||
ORDER BY created_at DESC LIMIT $1
|
||||
`, limit)
|
||||
rows, err = rows2, e
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := make([]WafAlert, 0, limit)
|
||||
for rows.Next() {
|
||||
var a WafAlert
|
||||
if err := rows.Scan(
|
||||
&a.ID, &a.DomainID, &a.Hostname, &a.ClientIP, &a.Method, &a.URI,
|
||||
&a.RuleID, &a.RuleMsg, &a.Severity, &a.Action, &a.CreatedAt,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, a)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// PurgeAlerts removes alerts older than the given number of days.
|
||||
func (r *Repo) PurgeAlerts(ctx context.Context, olderThanDays int) error {
|
||||
_, err := r.Pool.Exec(ctx,
|
||||
`DELETE FROM waf_alerts WHERE created_at < NOW() - ($1 || ' days')::interval`,
|
||||
olderThanDays,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
// DomainConfigPair combines a domain hostname with its WAF config.
|
||||
type DomainConfigPair struct {
|
||||
Hostname string
|
||||
|
||||
84
internal/waf/alerts.go
Normal file
84
internal/waf/alerts.go
Normal file
@@ -0,0 +1,84 @@
|
||||
package waf
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"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
|
||||
}
|
||||
|
||||
// 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),
|
||||
}
|
||||
go aw.run()
|
||||
return aw
|
||||
}
|
||||
|
||||
// Send enqueues an alert. Drops silently if the channel is full to
|
||||
// avoid slowing down SPOE request handling.
|
||||
func (aw *AlertWriter) Send(a Alert) {
|
||||
select {
|
||||
case aw.ch <- a:
|
||||
default:
|
||||
slog.Warn("waf: alert channel full — dropping alert", "host", a.Hostname, "rule", a.RuleID)
|
||||
}
|
||||
}
|
||||
|
||||
func (aw *AlertWriter) run() {
|
||||
for a := range aw.ch {
|
||||
aw.write(a)
|
||||
}
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"github.com/corazawaf/coraza/v3/types"
|
||||
"github.com/dropmorepackets/haproxy-go/pkg/encoding"
|
||||
"github.com/dropmorepackets/haproxy-go/spop"
|
||||
)
|
||||
@@ -13,8 +14,9 @@ import (
|
||||
// SPOEAgent wraps the haproxy-go SPOE server and dispatches each
|
||||
// inspected request to the appropriate per-domain Coraza engine.
|
||||
type SPOEAgent struct {
|
||||
Manager *Manager
|
||||
Addr string
|
||||
Manager *Manager
|
||||
AlertWriter *AlertWriter
|
||||
Addr string
|
||||
}
|
||||
|
||||
// ListenAndServe starts the SPOE agent. Blocks until ctx is cancelled.
|
||||
@@ -106,33 +108,48 @@ func (a *SPOEAgent) handle(ctx context.Context, w *encoding.ActionWriter, m *enc
|
||||
|
||||
// Evaluate request headers.
|
||||
interruption := tx.ProcessRequestHeaders()
|
||||
|
||||
// Log all matched rules (detection + blocking).
|
||||
for _, mr := range tx.MatchedRules() {
|
||||
a.sendAlert(host, clientIP, method, uri, mr, interruption != nil)
|
||||
}
|
||||
|
||||
if interruption != nil {
|
||||
status := interruption.Status
|
||||
if status == 0 {
|
||||
status = http.StatusForbidden
|
||||
}
|
||||
slog.Info("waf: request blocked",
|
||||
"host", host,
|
||||
"method", method,
|
||||
"uri", uri,
|
||||
"client", clientIP,
|
||||
"status", status,
|
||||
"mode", de.Mode,
|
||||
"host", host, "method", method, "uri", uri,
|
||||
"client", clientIP, "status", status, "rule", interruption.RuleID,
|
||||
)
|
||||
if de.Mode == "blocking" {
|
||||
if err := w.SetInt64(encoding.VarScopeTransaction, "status", int64(status)); err != nil {
|
||||
slog.Warn("waf: SetInt64 status", "error", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// sendAlert enqueues a WAF alert for async DB write.
|
||||
func (a *SPOEAgent) sendAlert(host, clientIP, method, uri string, mr types.MatchedRule, blocked bool) {
|
||||
if a.AlertWriter == nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Alert-only log for detection mode.
|
||||
if tx.IsInterrupted() && de.Mode != "blocking" {
|
||||
slog.Info("waf: request flagged (detection)",
|
||||
"host", host, "method", method, "uri", uri, "client", clientIP,
|
||||
)
|
||||
action := "detected"
|
||||
if blocked && mr.Disruptive() {
|
||||
action = "blocked"
|
||||
}
|
||||
a.AlertWriter.Send(Alert{
|
||||
Hostname: host,
|
||||
ClientIP: clientIP,
|
||||
Method: method,
|
||||
URI: uri,
|
||||
RuleID: mr.Rule().ID(),
|
||||
RuleMsg: mr.Message(),
|
||||
Severity: mr.Rule().Severity().String(),
|
||||
Action: action,
|
||||
})
|
||||
}
|
||||
|
||||
// parseHeaders splits HAProxy raw headers ("Name: value\r\n…") and
|
||||
|
||||
Reference in New Issue
Block a user