edgeguard-native/internal/waf/spoe.go

package waf

import (
	"context"
	"log/slog"
	"net"
	"net/http"
	"strings"

	"github.com/corazawaf/coraza/v3/types"
	"github.com/dropmorepackets/haproxy-go/pkg/encoding"
	"github.com/dropmorepackets/haproxy-go/spop"
)

// SPOEAgent wraps the haproxy-go SPOE server and dispatches each
// inspected request to the appropriate per-domain Coraza engine.
type SPOEAgent struct {
	Manager     *Manager
	AlertWriter *AlertWriter
	Addr        string
}

// ListenAndServe starts the SPOE agent. Blocks until ctx is canceled.
func (a *SPOEAgent) ListenAndServe(ctx context.Context) error {
	agent := spop.Agent{
		Addr:        a.Addr,
		Handler:     spop.HandlerFunc(a.handle),
		BaseContext: ctx,
	}
	return agent.ListenAndServe()
}

// handle is called by the haproxy-go SPOE library for every NOTIFY
// frame HAProxy sends. It extracts the request data, runs Coraza,
// and optionally sets a txn.waf.status variable to trigger a deny ACL.
func (a *SPOEAgent) handle(ctx context.Context, w *encoding.ActionWriter, m *encoding.Message) {
	var (
		clientIP string
		method   string
		uri      string // full request URI (path + optional ?query)
		httpVer  string
		host     string
		rawHdrs  string
		body     []byte // gepufferter Request-Body (via HAProxy option http-buffer-request)
	)

	// Iterate over the key-value pairs HAProxy sent with this message.
	entry := encoding.AcquireKVEntry()
	defer encoding.ReleaseKVEntry(entry)
	for m.KV.Next(entry) {
		switch {
		case entry.NameEquals("src"):
			addr := entry.ValueAddr()
			if addr.IsValid() {
				clientIP = addr.String()
			}
		case entry.NameEquals("method"):
			method = string(entry.ValueBytes())
		case entry.NameEquals("uri"):
			uri = string(entry.ValueBytes())
		case entry.NameEquals("ver"):
			httpVer = string(entry.ValueBytes())
		case entry.NameEquals("host"):
			host = string(entry.ValueBytes())
		case entry.NameEquals("headers"):
			rawHdrs = string(entry.ValueBytes())
		case entry.NameEquals("body"):
			// Kopieren: entry wird nach Reset() wiederverwendet, der
			// zugrundeliegende Puffer darf nicht referenziert bleiben.
			if b := entry.ValueBytes(); len(b) > 0 {
				body = append([]byte(nil), b...)
			}
		}
		entry.Reset()
	}

	if host == "" {
		return
	}

	de, ok := a.Manager.GetForHost(host)
	if !ok {
		return // WAF not configured or disabled for this domain
	}

	// Trusted-Proxy-Handling: stammt die Verbindung von einem konfigurierten
	// Trusted-Proxy, ist die echte Client-IP das letzte X-Forwarded-For-Glied
	// (das der Proxy angehängt hat), nicht die Proxy-IP selbst.
	if clientIP != "" && len(de.TrustedProxies) > 0 && ipMatchesAny(clientIP, de.TrustedProxies) {
		if real := rightmostXFF(rawHdrs); real != "" {
			clientIP = real
		}
	}

	tx := de.WAF.NewTransaction()
	defer func() {
		tx.ProcessLogging()
		if err := tx.Close(); err != nil {
			slog.Warn("waf: tx.Close", "error", err)
		}
	}()

	// Feed connection metadata.
	if clientIP != "" {
		tx.ProcessConnection(clientIP, 0, "", 0)
	}

	if uri == "" {
		uri = "/"
	}
	if httpVer == "" {
		httpVer = "HTTP/1.1"
	}
	tx.ProcessURI(uri, method, httpVer)

	// Feed Host header first (required by many CRS rules).
	tx.AddRequestHeader("Host", host)

	// Parse and feed all raw headers.
	parseHeaders(rawHdrs, func(name, val string) {
		if !strings.EqualFold(name, "host") { // already added above
			tx.AddRequestHeader(name, val)
		}
	})

	// Evaluate request headers.
	interruption := tx.ProcessRequestHeaders()

	// Request-Body inspizieren (POST/PUT-Payloads: Form-SQLi, JSON-Injection,
	// Uploads). Nur wenn die Header-Phase noch nicht geblockt hat. HAProxy
	// liefert den Body via `option http-buffer-request` (bis tune.bufsize) —
	// größere Bodies werden zur Prüfung gekappt. Content-Type kam bereits
	// über die Header, sodass Coraza urlencoded/multipart/json korrekt parst.
	if interruption == nil && len(body) > 0 {
		if it, _, err := tx.WriteRequestBody(body); err != nil {
			slog.Warn("waf: WriteRequestBody", "error", err)
		} else if it != nil {
			interruption = it
		} else {
			it, err := tx.ProcessRequestBody()
			if err != nil {
				slog.Warn("waf: ProcessRequestBody", "error", err)
			} else if it != nil {
				interruption = it
			}
		}
	}

	// 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, "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.
// Control-flow rules (pass+nolog with empty message) are skipped —
// they are CRS paranoia-level skip-markers, not real detections.
func (a *SPOEAgent) sendAlert(host, clientIP, method, uri string, mr types.MatchedRule, blocked bool) {
	if a.AlertWriter == nil {
		return
	}
	ruleID := mr.Rule().ID()
	// Skip CRS setup/initialization rules (900xxx–909xxx) — they fire on
	// every request as part of CRS init and are not security events.
	// Real detection rules start at 910xxx (IP reputation) and above.
	if ruleID > 0 && ruleID < 910000 {
		return
	}
	// Skip control-flow rules with no message (PL-skip markers).
	if mr.Message() == "" {
		return
	}
	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,
	})
}

// rightmostXFF gibt den letzten (vom nächstgelegenen Proxy angehängten)
// X-Forwarded-For-Eintrag zurück, sofern es eine gültige IP ist.
func rightmostXFF(rawHdrs string) string {
	var val string
	for _, line := range strings.Split(rawHdrs, "\n") {
		line = strings.TrimRight(line, "\r")
		idx := strings.IndexByte(line, ':')
		if idx <= 0 {
			continue
		}
		if strings.EqualFold(strings.TrimSpace(line[:idx]), "x-forwarded-for") {
			val = strings.TrimSpace(line[idx+1:]) // letzter XFF-Header gewinnt
		}
	}
	if val == "" {
		return ""
	}
	parts := strings.Split(val, ",")
	cand := strings.TrimSpace(parts[len(parts)-1])
	if net.ParseIP(cand) == nil {
		return ""
	}
	return cand
}

// ipMatchesAny prüft, ob ip exakt einer IP oder einem CIDR aus list entspricht.
func ipMatchesAny(ip string, list []string) bool {
	parsed := net.ParseIP(ip)
	if parsed == nil {
		return false
	}
	for _, e := range list {
		e = strings.TrimSpace(e)
		if e == "" {
			continue
		}
		if strings.Contains(e, "/") {
			if _, n, err := net.ParseCIDR(e); err == nil && n.Contains(parsed) {
				return true
			}
		} else if pe := net.ParseIP(e); pe != nil && pe.Equal(parsed) {
			return true
		}
	}
	return false
}

// parseHeaders splits HAProxy raw headers ("Name: value\r\n…") and
// calls fn for each valid header line.
func parseHeaders(raw string, fn func(name, val string)) {
	for _, line := range strings.Split(raw, "\n") {
		line = strings.TrimRight(line, "\r")
		if line == "" {
			continue
		}
		idx := strings.IndexByte(line, ':')
		if idx <= 0 {
			continue
		}
		name := strings.TrimSpace(line[:idx])
		val := strings.TrimSpace(line[idx+1:])
		if name != "" {
			fn(name, val)
		}
	}
}