diff --git a/README.md b/README.md index cb7798f..d592982 100644 --- a/README.md +++ b/README.md @@ -74,6 +74,18 @@ Customize the maps under [`examples/haproxy/maps/`](examples/haproxy/maps/) and Map keys are lowercase literal substrings without whitespace. User-Agent headers are trivial to spoof, so this policy is traffic shaping only; it must not protect authenticated or otherwise sensitive endpoints. +### Optional IP reputation (CrowdSec) + +Berghain ships a reputation service that acts as a [CrowdSec](https://www.crowdsec.net/) bouncer and pushes +decisions (plus static feeds such as Tor exit nodes) into HAProxy stick-tables **live over the peers protocol** — +no reloads, and Berghain stays stateless. Bans are silent-dropped, captcha decisions raise the minimum challenge +level, and per-decision durations are honored via timed stick-table entries. + +It runs either embedded in the agent (a `reputation:` section in the spop config) or standalone +(`cmd/feedupdater`) for setups that scale the feed separately; the daemon behaves as a first-class peer in +meshes with multiple HAProxy instances. See [`examples/crowdsec/`](examples/crowdsec/) and +[`examples/haproxy/haproxy-reputation.cfg`](examples/haproxy/haproxy-reputation.cfg). + ## Running with Docker To run the project using Docker, follow these steps: diff --git a/cmd/feedupdater/main.go b/cmd/feedupdater/main.go new file mode 100644 index 0000000..b53543b --- /dev/null +++ b/cmd/feedupdater/main.go @@ -0,0 +1,58 @@ +// Command feedupdater runs the Berghain IP reputation service standalone. +// +// It is a thin wrapper around internal/reputation: it fetches CrowdSec +// decisions and static feeds and pushes individual-IP reputation into HAProxy +// stick-tables over the peers protocol. Simple deployments can skip this +// binary entirely and enable the same service inside cmd/spop via the +// `reputation:` config section; this command exists for setups that scale or +// isolate the feed daemon separately. +package main + +import ( + "context" + "flag" + "log/slog" + "os" + "os/signal" + + "github.com/DropMorePackets/berghain/internal/reputation" +) + +func main() { + var ( + cfg reputation.Config + logLevelArg string + ) + + flag.StringVar(&cfg.PeerListen, "peer-listen", "127.0.0.1:10001", "listen address for the HAProxy peers protocol") + flag.StringVar(&cfg.LocalPeer, "local-peer", "", "this peer's name in the HAProxy peers section (default berghain_feed)") + flag.StringVar(&cfg.TableV4, "table-v4", "", "IPv4 reputation stick-table name (default st_reputation_v4)") + flag.StringVar(&cfg.TableV6, "table-v6", "", "IPv6 reputation stick-table name (default st_reputation_v6)") + flag.DurationVar(&cfg.TableExpiry, "table-expiry", 0, "stick-table expire value, bounds entries without their own duration (default 24h)") + flag.StringVar(&cfg.MapsDir, "maps-dir", "", "directory to write CIDR map/ACL files into; empty disables CIDR feeds") + flag.DurationVar(&cfg.Interval, "interval", 0, "static feed refresh interval (default 6h)") + flag.StringVar(&cfg.Banlist, "banlist", "", "optional file of individual IPs to block (one per line)") + flag.StringVar(&cfg.CrowdSec.URL, "crowdsec-url", "", "CrowdSec LAPI URL (e.g. http://crowdsec:8080); empty disables the source") + flag.StringVar(&cfg.CrowdSec.APIKey, "crowdsec-api-key", "", "CrowdSec bouncer API key (or CROWDSEC_API_KEY env)") + flag.DurationVar(&cfg.CrowdSec.Interval, "crowdsec-interval", 0, "CrowdSec decision-stream poll interval (default 10s)") + torExits := flag.Bool("tor-exits", true, "challenge Tor exit nodes") + flag.StringVar(&logLevelArg, "loglevel", "info", "Logging level") + flag.Parse() + + cfg.TorExits = torExits + + var logLevel slog.Level + if err := logLevel.UnmarshalText([]byte(logLevelArg)); err != nil { + slog.Error("invalid log level, cannot proceed", "loglevel", logLevelArg) + os.Exit(1) + } + slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: logLevel}))) + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) + defer stop() + + if err := reputation.New(cfg).Run(ctx); err != nil { + slog.Error("reputation service failed", "error", err) + os.Exit(1) + } +} diff --git a/cmd/spop/config.go b/cmd/spop/config.go index e2e1c1b..634c14d 100644 --- a/cmd/spop/config.go +++ b/cmd/spop/config.go @@ -9,6 +9,7 @@ import ( "github.com/goccy/go-yaml" "github.com/DropMorePackets/berghain" + "github.com/DropMorePackets/berghain/internal/reputation" ) type Config struct { @@ -16,6 +17,12 @@ type Config struct { Listen string `yaml:"listen"` Default FrontendConfig `yaml:"default"` Frontend map[string]FrontendConfig `yaml:"frontend"` + + // Reputation optionally embeds the IP reputation service (CrowdSec + + // static feeds over the peers protocol) into this process, so a simple + // deployment needs no separate feed daemon. Enabled when peer_listen is + // set; cmd/feedupdater runs the same service standalone. + Reputation reputation.Config `yaml:"reputation"` } type Secret []byte diff --git a/cmd/spop/config.yaml b/cmd/spop/config.yaml index fc39457..efbbdf1 100644 --- a/cmd/spop/config.yaml +++ b/cmd/spop/config.yaml @@ -1,5 +1,15 @@ secret: JMal0XJRROOMsMdPqggG2tR56CTkpgN3r47GgUN/WSQ= +# Optional embedded IP reputation: serve CrowdSec decisions and static feeds +# to HAProxy over the peers protocol from this same process. See +# examples/haproxy/haproxy-reputation.cfg for the matching HAProxy side. +#reputation: +# peer_listen: 0.0.0.0:10001 # must match this peer's entry in the peers section +# local_peer: berghain_feed # our name in the peers section +# crowdsec: +# url: http://crowdsec:8080 # LAPI; api_key via CROWDSEC_API_KEY or api_key: +# maps_dir: examples/haproxy/maps # where CIDR feeds land; empty disables them + default: levels: - duration: 24h diff --git a/cmd/spop/config_test.go b/cmd/spop/config_test.go new file mode 100644 index 0000000..a90e5a3 --- /dev/null +++ b/cmd/spop/config_test.go @@ -0,0 +1,45 @@ +package main + +import ( + "testing" + "time" + + "github.com/goccy/go-yaml" +) + +func TestConfigReputationBlock(t *testing.T) { + var cfg Config + err := yaml.Unmarshal([]byte(` +secret: JMal0XJRROOMsMdPqggG2tR56CTkpgN3r47GgUN/WSQ= +reputation: + peer_listen: 0.0.0.0:10001 + crowdsec: + url: http://crowdsec:8080 + interval: 30s +`), &cfg) + if err != nil { + t.Fatal(err) + } + if !cfg.Reputation.Enabled() { + t.Error("reputation block with peer_listen must enable the service") + } + if cfg.Reputation.CrowdSec.URL != "http://crowdsec:8080" { + t.Errorf("crowdsec url = %q", cfg.Reputation.CrowdSec.URL) + } + if cfg.Reputation.CrowdSec.Interval != 30*time.Second { + t.Errorf("crowdsec interval = %v, want 30s", cfg.Reputation.CrowdSec.Interval) + } +} + +func TestConfigReputationAbsent(t *testing.T) { + var cfg Config + err := yaml.Unmarshal([]byte(` +secret: JMal0XJRROOMsMdPqggG2tR56CTkpgN3r47GgUN/WSQ= +`), &cfg) + if err != nil { + t.Fatal(err) + } + if cfg.Reputation.Enabled() { + t.Error("absent reputation block must leave the service disabled") + } +} diff --git a/cmd/spop/main.go b/cmd/spop/main.go index 770a7ac..78c6524 100644 --- a/cmd/spop/main.go +++ b/cmd/spop/main.go @@ -14,6 +14,8 @@ import ( "github.com/dropmorepackets/haproxy-go/pkg/encoding" "github.com/dropmorepackets/haproxy-go/spop" + + "github.com/DropMorePackets/berghain/internal/reputation" ) var ( @@ -58,12 +60,27 @@ func main() { ec := make(chan error) + cfg := loadConfig() + var wg sync.WaitGroup wg.Add(1) go func() { - ec <- runBerghain(&wg, ctx) + ec <- runBerghain(&wg, ctx, cfg) }() + // Embedded mode: run the IP reputation service (CrowdSec + static feeds + // over the peers protocol) in-process when configured, so simple setups + // need no separate feed daemon. + if cfg.Reputation.Enabled() { + wg.Add(1) + go func() { + defer wg.Done() + if err := reputation.New(cfg.Reputation).Run(ctx); err != nil { + ec <- err + } + }() + } + select { case <-c: cancelFunc() @@ -81,10 +98,9 @@ func main() { <-c } -func runBerghain(wg *sync.WaitGroup, ctx context.Context) error { +func runBerghain(wg *sync.WaitGroup, ctx context.Context, cfg Config) error { defer wg.Done() - cfg := loadConfig() if len(cfg.Secret) != 32 { Fatal("provided secret has invalid length", "have", len(cfg.Secret), "need", 32) } diff --git a/examples/crowdsec/README.md b/examples/crowdsec/README.md new file mode 100644 index 0000000..17a20fe --- /dev/null +++ b/examples/crowdsec/README.md @@ -0,0 +1,72 @@ +# CrowdSec IP reputation (optional) + +Berghain can act as a [CrowdSec](https://www.crowdsec.net/) *bouncer*: its +reputation service polls the CrowdSec local API decision stream and pushes +every decision into HAProxy stick-tables **live over the peers protocol** — no +map files, no reloads, and Berghain itself stays stateless. Decisions map to +the `gpt0` tag consumed by +[`examples/haproxy/haproxy-reputation.cfg`](../haproxy/haproxy-reputation.cfg): + +| CrowdSec decision | gpt0 | HAProxy behaviour | +|-------------------|------|--------------------------------------| +| `ban` | 1 | `silent-drop` | +| `captcha` | 3 | minimum Berghain challenge level 3 | +| anything else | 1 | fail closed to a ban | + +Per-decision durations are honored: entries are pushed as timed stick-table +updates and expire in HAProxy exactly when the decision does, even if the +service is down at that moment. The same service also challenges Tor exit +nodes (static feed) unless `tor_exits: false` is set. + +## Running it + +The service runs **embedded** in the Berghain agent (`reputation:` section in +the spop config), so the docker setup stays two containers: the existing +haproxy+berghain container and CrowdSec. + +1. Start CrowdSec once so you can register the bouncer: + + ```sh + docker compose -f docker-compose.yml -f examples/crowdsec/docker-compose.crowdsec.yml up -d crowdsec + docker compose -f docker-compose.yml -f examples/crowdsec/docker-compose.crowdsec.yml \ + exec crowdsec cscli bouncers add berghain + ``` + +2. Export the printed key and start the rest: + + ```sh + export CROWDSEC_API_KEY= + docker compose -f docker-compose.yml -f examples/crowdsec/docker-compose.crowdsec.yml up + ``` + +3. Try it — add a decision and watch the stick-table: + + ```sh + docker compose -f docker-compose.yml -f examples/crowdsec/docker-compose.crowdsec.yml \ + exec crowdsec cscli decisions add --ip 203.0.113.7 --type ban --duration 5m + ``` + + Within the poll interval (10s by default) requests from that address are + silent-dropped; `cscli decisions delete --ip 203.0.113.7` lifts it again + within one poll. + +CrowdSec only produces decisions when it can *see* traffic (or when another +machine in your CrowdSec network reports it): feed it your HAProxy logs via an +acquisition file, or rely on the community blocklist that comes with console +enrollment. Both are standard CrowdSec configuration — see their +[HAProxy collection](https://app.crowdsec.net/hub/author/crowdsecurity/collections/haproxy). + +## Standalone mode + +Deployments that scale HAProxy and the feed separately can run the exact same +service as its own daemon instead of embedding it: + +```sh +go run ./cmd/feedupdater \ + -peer-listen 0.0.0.0:10001 \ + -crowdsec-url http://crowdsec:8080 # key via CROWDSEC_API_KEY +``` + +Every HAProxy in the peers mesh then lists `berghain_feed` once and they all +learn the same tables; the daemon copes fine with being one peer among many +(it validates handshakes and acknowledges the updates the other peers teach). diff --git a/examples/crowdsec/config.yaml b/examples/crowdsec/config.yaml new file mode 100644 index 0000000..0310445 --- /dev/null +++ b/examples/crowdsec/config.yaml @@ -0,0 +1,17 @@ +# Berghain config for the CrowdSec example: the default challenge levels plus +# the embedded IP reputation service. The API key comes from the +# CROWDSEC_API_KEY environment variable (see examples/crowdsec/README.md). +secret: JMal0XJRROOMsMdPqggG2tR56CTkpgN3r47GgUN/WSQ= + +reputation: + peer_listen: 127.0.0.1:10001 + local_peer: berghain_feed + crowdsec: + url: http://crowdsec:8080 + +default: + levels: + - duration: 24h + type: none + - duration: 30m + type: pow diff --git a/examples/crowdsec/docker-compose.crowdsec.yml b/examples/crowdsec/docker-compose.crowdsec.yml new file mode 100644 index 0000000..b902cf3 --- /dev/null +++ b/examples/crowdsec/docker-compose.crowdsec.yml @@ -0,0 +1,39 @@ +# CrowdSec IP reputation overlay (issue #82). +# +# Runs CrowdSec next to the regular single-container Berghain setup and turns +# on Berghain's embedded reputation service: it polls the CrowdSec LAPI +# decision stream as a bouncer and pushes decisions into HAProxy stick-tables +# over the peers protocol. See examples/crowdsec/README.md for the key setup. +# +# docker compose -f docker-compose.yml -f examples/crowdsec/docker-compose.crowdsec.yml up + +services: + berghain-haproxy: + volumes: + - ./examples/haproxy/haproxy-reputation.cfg:/app/haproxy.cfg + - ./examples/haproxy/berghain.cfg:/app/examples/haproxy/berghain.cfg + - ./examples/crowdsec/config.yaml:/app/config.yaml + # -L names this HAProxy's own entry in the peers section. + command: sh -c "haproxy -L haproxy_local -f haproxy.cfg & ./berghain -config config.yaml" + environment: + # Register with: docker compose exec crowdsec cscli bouncers add berghain + - CROWDSEC_API_KEY=${CROWDSEC_API_KEY:?run cscli bouncers add berghain and export the key} + depends_on: + - crowdsec + + # Prefer the standalone daemon instead of embedded mode? Disable the + # reputation section in config.yaml and run cmd/feedupdater as its own + # service with the same flags-equivalent settings. + + crowdsec: + image: crowdsecurity/crowdsec:latest + environment: + - COLLECTIONS=crowdsecurity/base-http-scenarios + volumes: + - crowdsec-db:/var/lib/crowdsec/data + - crowdsec-config:/etc/crowdsec + restart: unless-stopped + +volumes: + crowdsec-db: + crowdsec-config: diff --git a/examples/haproxy/haproxy-reputation.cfg b/examples/haproxy/haproxy-reputation.cfg new file mode 100644 index 0000000..2523f58 --- /dev/null +++ b/examples/haproxy/haproxy-reputation.cfg @@ -0,0 +1,126 @@ +# Optional live IP-reputation policy for Berghain (issue #82). +# +# This is a complete alternative to examples/haproxy/haproxy.cfg. Berghain's +# reputation service (embedded in cmd/spop via the `reputation:` config +# section, or standalone as cmd/feedupdater) joins the peers mesh below as +# "berghain_feed" and pushes per-IP reputation into the st_reputation +# stick-tables live — no map files, no reloads. The gpt0 tag encodes the +# action: 1 = block (silent-drop), >= 2 = minimum challenge level. +# +# HAProxy must know which peer entry is itself: run it with +# haproxy -L haproxy_local -f examples/haproxy/haproxy-reputation.cfg +# +# The mesh survives more peers: add further HAProxy instances to the peers +# section and they all learn the same reputation tables. +# +# Validate from the repository root with: +# haproxy -c -L haproxy_local -f examples/haproxy/haproxy-reputation.cfg + +global + log stdout format raw local0 + +defaults + mode http + log global + timeout client 5s + timeout server 5s + timeout connect 5s + option httplog + +listen stats + bind 127.0.0.1:8000 + stats enable + stats uri / + stats refresh 15s + +# The reputation service listens as the "berghain_feed" peer. HAProxy dials it +# on startup to resync and keeps the session for live pushes. +peers berghain + peer haproxy_local 127.0.0.1:10000 + peer berghain_feed 127.0.0.1:10001 + +frontend test + bind *:8080 + log-format "%ci:%cp\ [%t]\ %ft\ %b/%s\ %Th/%Ti/%TR/%Tq/%Tw/%Tc/%Tr/%Tt\ %ST\ %B\ %CC\ %CS\ %tsc\ %ac/%fc/%bc/%sc/%rc\ %sq/%bq\ %hr\ %hs\ %{+Q}r\ %ID spoa-error:\ %[var(txn.berghain.error)]" + + acl berghain_path path /cdn-cgi/challenge-platform/challenge + + # HAProxy issues the initial support ID; continuation requests carry it in their body. + http-request set-var-fmt(txn.berghain.session) "bh@%[uuid()]" if berghain_path METH_GET + + # Individual-IP reputation is pushed live into the st_reputation tables by + # the Berghain reputation service over the peers protocol. Stick-tables + # key on exact addresses, so CIDR feeds (e.g. the generated + # maps/cloudflare-ips.lst) would be matched with `src -f ` instead. + acl src_is_v4 src -m ip 0.0.0.0/0 + http-request set-var(req.rep) src,table_gpt0(st_reputation_v4) if src_is_v4 + http-request set-var(req.rep) src,table_gpt0(st_reputation_v6) if !src_is_v4 + + # Reputation tag 1 (e.g. a CrowdSec ban): drop without an answer. Keep the + # Berghain endpoint out so a ban lifted mid-challenge cannot wedge clients. + http-request silent-drop if { var(req.rep) -m int eq 1 } + + http-request track-sc1 src table st_src + + filter spoe engine berghain config examples/haproxy/berghain.cfg + + # Base challenge level from request rate; rules are increasing so the + # highest match wins. + http-request set-var(req.berghain.level) int(1) if { sc1_http_req_rate gt 5 } + http-request set-var(req.berghain.level) int(2) if { sc1_http_req_rate gt 10 } + http-request set-var(req.berghain.level) int(3) if { sc1_http_req_rate gt 15 } + + # Flagged sources (reputation tag >= 2, e.g. a CrowdSec captcha decision or + # a Tor exit) raise the MINIMUM level — only ever upward, via the ge guard. + http-request set-var(req.berghain.level) int(3) if { var(req.rep) -m int ge 2 } !{ var(req.berghain.level) -m int ge 3 } + + acl berghain_active var(req.berghain.level) -m found + + http-request send-spoe-group berghain validate if !berghain_path berghain_active + http-request return status 501 if { var(txn.berghain.error) -m found } + + acl berghain_valid var(txn.berghain.valid) -m bool + acl is_ssl ssl_fc + + http-request return status 403 content-type "text/html" file "web/dist/default/index.html" if !berghain_valid !berghain_path berghain_active !is_ssl + http-request return status 403 content-type "text/html" file "web/dist/native-crypto/index.html" if !berghain_valid !berghain_path berghain_active is_ssl + http-request wait-for-body time 5s if berghain_path METH_POST + use_backend berghain_http if berghain_path + + default_backend app_backend + +backend st_src + stick-table type ipv6 size 1m expire 15m store http_req_rate(10s) + +# The reputation tables. Names, key types and the expire value must match the +# reputation service configuration (table_v4/table_v6/table_expiry). +backend st_reputation_v4 + stick-table type ip size 1m expire 24h store gpt0 peers berghain + +backend st_reputation_v6 + stick-table type ipv6 size 1m expire 24h store gpt0 peers berghain + +backend app_backend + mode http + http-request return status 200 content-type "text/plain" string "Hello World!" + +backend berghain_http + mode http + filter spoe engine berghain_challenge config examples/haproxy/berghain.cfg + + acl is_challenge_path path /cdn-cgi/challenge-platform/challenge + + http-request send-spoe-group berghain_challenge challenge if is_challenge_path + http-request return status 501 if { var(txn.berghain.error) -m found } + + acl has_token var(txn.berghain.token) -m found + + http-after-response add-header set-cookie "berghain=%[var(txn.berghain.token)]; %[var(txn.berghain.domain)] path=/;" if has_token + http-request return status 200 content-type "application/json" lf-string "%[var(txn.berghain.response)]" if is_challenge_path + + http-request return status 404 + +backend berghain_spop + mode tcp + option spop-check + server localhost unix@./spop.sock check diff --git a/go.mod b/go.mod index daeb91c..2d69c29 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,6 @@ go 1.21.0 toolchain go1.23.2 require ( - github.com/dropmorepackets/haproxy-go v0.0.7 + github.com/dropmorepackets/haproxy-go v0.1.1 github.com/goccy/go-yaml v1.18.0 ) diff --git a/go.sum b/go.sum index 30f1980..1b56302 100644 --- a/go.sum +++ b/go.sum @@ -1,6 +1,4 @@ -github.com/dropmorepackets/haproxy-go v0.0.6 h1:0u0u4MLS+mbIrYCQrIkHq8PQvt6ePJgF6ogTIFZQzx8= -github.com/dropmorepackets/haproxy-go v0.0.6/go.mod h1:4a2AmmVjvg2zPNdizGZrMN8ZSUpj90U43VlcdbOIBnU= -github.com/dropmorepackets/haproxy-go v0.0.7 h1:atXkB0MSRBZrAgpq+Vj/E4KysQ4CiI0O5QGUr+HvfTw= -github.com/dropmorepackets/haproxy-go v0.0.7/go.mod h1:4a2AmmVjvg2zPNdizGZrMN8ZSUpj90U43VlcdbOIBnU= +github.com/dropmorepackets/haproxy-go v0.1.1 h1:qYovzYpGHanQCBQW5k92uJqIdjcwSJQHZ0VFRiI3FJ0= +github.com/dropmorepackets/haproxy-go v0.1.1/go.mod h1:4a2AmmVjvg2zPNdizGZrMN8ZSUpj90U43VlcdbOIBnU= github.com/goccy/go-yaml v1.18.0 h1:8W7wMFS12Pcas7KU+VVkaiCng+kG8QiFeFwzFb+rwuw= github.com/goccy/go-yaml v1.18.0/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA= diff --git a/internal/peerserver/peerserver.go b/internal/peerserver/peerserver.go new file mode 100644 index 0000000..38b4ad5 --- /dev/null +++ b/internal/peerserver/peerserver.go @@ -0,0 +1,513 @@ +// Package peerserver implements the *sending* side of the HAProxy peers +// protocol: it lets an external process push stick-table entries into a running +// HAProxy without a reload. haproxy-go's peers package only implements the +// receiving side, so the framing/handshake here is built on its exported +// sticktable encoders and message constants. +// +// The server is designed to live in a peers mesh where it is not the only +// peer: it validates handshakes, acknowledges entry updates pushed by the +// other peers (so they consider us synced instead of re-teaching forever), +// and answers resync requests from any number of connections. +// +// It is used to feed individual-IP reputation (bans, Tor exit nodes, ...) into +// HAProxy stick-tables live. Stick-tables key on exact IPs, so CIDR/ASN feeds +// still belong in map files — see internal/reputation. +package peerserver + +import ( + "bufio" + "encoding/binary" + "fmt" + "io" + "log/slog" + "net" + "net/netip" + "sort" + "strings" + "sync" + "time" + + "github.com/dropmorepackets/haproxy-go/peers" + "github.com/dropmorepackets/haproxy-go/peers/sticktable" + "github.com/dropmorepackets/haproxy-go/pkg/encoding" +) + +// Entry is the desired state for one address. +type Entry struct { + // Value is the gpt0 tag (e.g. 1=block, 3=challenge); 0 clears the entry. + Value uint32 + // ExpiresAt bounds the entry's lifetime in HAProxy (sent as a timed + // update). The zero value means the table's default expiry applies. + ExpiresAt time.Time +} + +// entryState is an Entry plus the update ID it was taught under. Entries keep +// their ID for their whole life so resyncs replay the same IDs that already +// went out and acknowledgements stay unambiguous. +type entryState struct { + value uint32 + expireAt time.Time + id uint32 +} + +func (st entryState) expired(now time.Time) bool { + return !st.expireAt.IsZero() && now.After(st.expireAt) +} + +// table holds the entries for one HAProxy stick-table (one address family). +type table struct { + def *sticktable.Definition + keyType sticktable.KeyType + + mu sync.Mutex + entries map[netip.Addr]entryState + updateID uint32 +} + +// Server serves one or more reputation stick-tables to connected HAProxy peers. +// The zero value is not usable; use New. +type Server struct { + localPeer string + expiry time.Duration + + v4 *table + v6 *table + + mu sync.Mutex + conns map[*conn]struct{} +} + +// New creates a server exposing an IPv4 table and an IPv6 table with the given +// names (which must match the HAProxy `backend` stick-table names) and default +// expiry. +func New(localPeer, v4Name, v6Name string, expiry time.Duration) *Server { + exp := uint64(expiry.Milliseconds()) + mkTable := func(id uint64, name string, kt sticktable.KeyType, kl uint64) *table { + return &table{ + keyType: kt, + entries: make(map[netip.Addr]entryState), + def: &sticktable.Definition{ + StickTableID: id, + Name: name, + KeyType: kt, + KeyLength: kl, + DataTypes: []sticktable.DataTypeDefinition{{DataType: sticktable.DataTypeGPT0}}, + Expiry: exp, + }, + } + } + return &Server{ + localPeer: localPeer, + expiry: expiry, + v4: mkTable(1, v4Name, sticktable.KeyTypeIPv4Address, 4), + v6: mkTable(2, v6Name, sticktable.KeyTypeIPv6Address, 16), + conns: make(map[*conn]struct{}), + } +} + +func (s *Server) tableFor(a netip.Addr) *table { + if a.Is4() { + return s.v4 + } + return s.v6 +} + +// Set adds or updates a reputation entry and pushes it to all connected peers. +func (s *Server) Set(a netip.Addr, e Entry) { + a = a.Unmap() + t := s.tableFor(a) + + t.mu.Lock() + if cur, ok := t.entries[a]; ok && cur.value == e.Value && cur.expireAt.Equal(e.ExpiresAt) { + t.mu.Unlock() + return + } + t.updateID++ + st := entryState{value: e.Value, expireAt: e.ExpiresAt, id: t.updateID} + t.entries[a] = st + t.mu.Unlock() + + s.broadcast(func(c *conn) { _ = c.sendEntry(t, a, st) }) +} + +// Delete clears an entry. HAProxy has no peer "delete"; we teach value 0 with +// a bounded expiry so every peer — including ones that reconnect later and +// resync — converges on the entry being gone. +func (s *Server) Delete(a netip.Addr) { + s.Set(a, Entry{Value: 0, ExpiresAt: time.Now().Add(s.expiry)}) +} + +// ReplaceAll moves the full entry set to the given desired state (used after a +// feed refresh): new/changed entries are pushed, dropped entries are cleared, +// and locally expired ones are pruned. +func (s *Server) ReplaceAll(values map[netip.Addr]Entry) { + seen := make(map[netip.Addr]struct{}, len(values)) + for a, e := range values { + seen[a.Unmap()] = struct{}{} + s.Set(a, e) + } + + now := time.Now() + for _, t := range []*table{s.v4, s.v6} { + var stale []netip.Addr + t.mu.Lock() + for a, st := range t.entries { + if st.expired(now) { + delete(t.entries, a) + continue + } + if _, ok := seen[a]; !ok && st.value != 0 { + stale = append(stale, a) + } + } + t.mu.Unlock() + for _, a := range stale { + s.Delete(a) + } + } +} + +// Get returns the live entry for an address, if any. +func (s *Server) Get(a netip.Addr) (Entry, bool) { + a = a.Unmap() + t := s.tableFor(a) + t.mu.Lock() + defer t.mu.Unlock() + st, ok := t.entries[a] + if !ok || st.value == 0 || st.expired(time.Now()) { + return Entry{}, false + } + return Entry{Value: st.value, ExpiresAt: st.expireAt}, true +} + +// Len returns the number of live non-zero entries across both tables. +func (s *Server) Len() int { + now := time.Now() + n := 0 + for _, t := range []*table{s.v4, s.v6} { + t.mu.Lock() + for _, st := range t.entries { + if st.value != 0 && !st.expired(now) { + n++ + } + } + t.mu.Unlock() + } + return n +} + +func (s *Server) broadcast(fn func(*conn)) { + s.mu.Lock() + conns := make([]*conn, 0, len(s.conns)) + for c := range s.conns { + conns = append(conns, c) + } + s.mu.Unlock() + for _, c := range conns { + fn(c) + } +} + +// Serve accepts HAProxy peer connections until the listener is closed. +func (s *Server) Serve(l net.Listener) error { + for { + nc, err := l.Accept() + if err != nil { + return err + } + c := &conn{nc: nc, br: bufio.NewReader(nc), srv: s} + s.mu.Lock() + s.conns[c] = struct{}{} + s.mu.Unlock() + go func() { + if err := c.serve(); err != nil { + slog.Debug("peer connection closed", "error", err) + } + s.mu.Lock() + delete(s.conns, c) + s.mu.Unlock() + _ = nc.Close() + }() + } +} + +type conn struct { + nc net.Conn + br *bufio.Reader + srv *Server + + wmu sync.Mutex // serialises all writes on this connection + + // State of the remote peer's teach stream, needed to acknowledge its + // updates: entry updates apply to the last announced table definition. + remoteDef *sticktable.Definition + remoteUpdateID uint32 +} + +func (c *conn) serve() error { + var h peers.Handshake + if _, err := h.ReadFrom(c.br); err != nil { + return fmt.Errorf("handshake: %w", err) + } + + status := peers.HandshakeStatusHandshakeSucceeded + switch { + case h.ProtocolIdentifier != "HAProxyS": + status = peers.HandshakeStatusProtocolError + case !strings.HasPrefix(h.Version, "2."): + status = peers.HandshakeStatusBadVersion + case h.RemotePeer != c.srv.localPeer: + // The remote addressed a peer name that is not ours. + status = peers.HandshakeStatusLocalPeerIdentifierMismatch + } + if _, err := c.nc.Write([]byte(fmt.Sprintf("%d\n", status))); err != nil { + return fmt.Errorf("handshake reply: %w", err) + } + if status != peers.HandshakeStatusHandshakeSucceeded { + return fmt.Errorf("handshake rejected with %d: proto=%q version=%q target=%q from=%q", + status, h.ProtocolIdentifier, h.Version, h.RemotePeer, h.LocalPeerIdentifier) + } + slog.Debug("peer connected", "remote", h.LocalPeerIdentifier) + + // Proactively announce tables and push the current state. + if err := c.fullSync(); err != nil { + return err + } + + // Heartbeat so HAProxy does not consider us dead (5s timeout). + stop := make(chan struct{}) + defer close(stop) + go func() { + t := time.NewTicker(3 * time.Second) + defer t.Stop() + for { + select { + case <-stop: + return + case <-t.C: + c.wmu.Lock() + _, err := c.nc.Write([]byte{byte(peers.MessageClassControl), byte(peers.ControlMessageHeartbeat)}) + c.wmu.Unlock() + if err != nil { + return + } + } + } + }() + + return c.readLoop() +} + +// readLoop consumes messages from the remote peer. In a mesh the other peers +// teach us *their* stick-table state; we acknowledge those updates so they +// consider this peer synced. Resync requests replay our full state. +func (c *conn) readLoop() error { + for { + class, err := c.br.ReadByte() + if err != nil { + return err + } + typ, err := c.br.ReadByte() + if err != nil { + return err + } + + // Messages with type >= 128 carry a varint-length-prefixed payload. + var payload []byte + if typ >= 0x80 { + n, err := encoding.ReadVarint(c.br) + if err != nil { + return err + } + payload = make([]byte, n) + if _, err := io.ReadFull(c.br, payload); err != nil { + return err + } + } + + switch peers.MessageClass(class) { + case peers.MessageClassControl: + if peers.ControlMessageType(typ) == peers.ControlMessageSyncRequest { + if err := c.fullSync(); err != nil { + return err + } + } + case peers.MessageClassStickTableUpdates: + if err := c.handleStickTableMessage(peers.StickTableUpdateMessageType(typ), payload); err != nil { + return err + } + } + } +} + +// handleStickTableMessage tracks the remote teach stream and acknowledges its +// entry updates. Tables whose definition we cannot decode are skipped (their +// updates are consumed without an ack), which keeps unknown data types from +// killing the connection. +func (c *conn) handleStickTableMessage(typ peers.StickTableUpdateMessageType, payload []byte) error { + switch typ { + case peers.StickTableUpdateMessageTypeStickTableDefinition: + var def sticktable.Definition + if _, err := def.Unmarshal(payload); err != nil { + slog.Debug("undecodable remote table definition", "error", err) + c.remoteDef = nil + return nil + } + c.remoteDef = &def + return nil + case peers.StickTableUpdateMessageTypeStickTableSwitch: + // Switches reference an earlier definition by ID; we only track the + // last one, so treat the stream as unknown until the next definition. + c.remoteDef = nil + return nil + case peers.StickTableUpdateMessageTypeUpdateAcknowledge: + // Acks for our own updates; resyncs replay stable IDs, so there is + // nothing to track. + return nil + case peers.StickTableUpdateMessageTypeEntryUpdate, + peers.StickTableUpdateMessageTypeUpdateTimed, + peers.StickTableUpdateMessageTypeIncrementalEntryUpdate, + peers.StickTableUpdateMessageTypeIncrementalEntryUpdateTimed: + if c.remoteDef == nil { + return nil + } + + e := sticktable.EntryUpdate{ + StickTable: c.remoteDef, + LocalUpdateID: c.remoteUpdateID + 1, // incremental updates imply last+1 + } + switch typ { + case peers.StickTableUpdateMessageTypeEntryUpdate: + e.WithLocalUpdateID = true + case peers.StickTableUpdateMessageTypeUpdateTimed: + e.WithLocalUpdateID = true + e.WithExpiry = true + case peers.StickTableUpdateMessageTypeIncrementalEntryUpdateTimed: + e.WithExpiry = true + } + if _, err := e.Unmarshal(payload); err != nil { + slog.Debug("undecodable remote entry update", "error", err) + return nil + } + c.remoteUpdateID = e.LocalUpdateID + + return c.sendAck(c.remoteDef.StickTableID, e.LocalUpdateID) + default: + return nil + } +} + +// sendAck acknowledges the remote peer's update so it considers us in sync. +func (c *conn) sendAck(tableID uint64, updateID uint32) error { + var payload [14]byte + n, err := encoding.PutVarint(payload[:], tableID) + if err != nil { + return err + } + binary.BigEndian.PutUint32(payload[n:], updateID) + return c.sendMessage(peers.MessageClassStickTableUpdates, + byte(peers.StickTableUpdateMessageTypeUpdateAcknowledge), payload[:n+4]) +} + +// fullSync sends every table definition and all current entries (in teach +// order), followed by a resync-finished control message. +func (c *conn) fullSync() error { + now := time.Now() + for _, t := range []*table{c.srv.v4, c.srv.v6} { + if err := c.sendDefinition(t); err != nil { + return err + } + + type teachEntry struct { + addr netip.Addr + st entryState + } + t.mu.Lock() + snapshot := make([]teachEntry, 0, len(t.entries)) + for a, st := range t.entries { + if st.expired(now) { + continue + } + snapshot = append(snapshot, teachEntry{addr: a, st: st}) + } + t.mu.Unlock() + + // Replay in the order the entries were originally taught so the + // remote's last-seen update ID moves monotonically. + sort.Slice(snapshot, func(i, j int) bool { return snapshot[i].st.id < snapshot[j].st.id }) + for _, te := range snapshot { + if err := c.sendEntry(t, te.addr, te.st); err != nil { + return err + } + } + } + c.wmu.Lock() + _, err := c.nc.Write([]byte{byte(peers.MessageClassControl), byte(peers.ControlMessageSyncFinished)}) + c.wmu.Unlock() + return err +} + +func (c *conn) sendMessage(class peers.MessageClass, typ byte, payload []byte) error { + var lenbuf [10]byte + ln, err := encoding.PutVarint(lenbuf[:], uint64(len(payload))) + if err != nil { + return err + } + c.wmu.Lock() + defer c.wmu.Unlock() + if _, err := c.nc.Write([]byte{byte(class), typ}); err != nil { + return err + } + if _, err := c.nc.Write(lenbuf[:ln]); err != nil { + return err + } + _, err = c.nc.Write(payload) + return err +} + +func (c *conn) sendDefinition(t *table) error { + buf := make([]byte, 256) + n, err := t.def.Marshal(buf) + if err != nil { + return err + } + return c.sendMessage(peers.MessageClassStickTableUpdates, + byte(peers.StickTableUpdateMessageTypeStickTableDefinition), buf[:n]) +} + +// sendEntry marshals and sends an entry update (update-id + key + gpt0), +// using the timed variant when the entry carries its own expiry. +func (c *conn) sendEntry(t *table, a netip.Addr, st entryState) error { + var key sticktable.MapKey + if t.keyType == sticktable.KeyTypeIPv4Address { + k := sticktable.IPv4AddressKey(a) + key = &k + } else { + k := sticktable.IPv6AddressKey(a) + key = &k + } + + d := sticktable.UnsignedIntegerData(st.value) + e := sticktable.EntryUpdate{ + StickTable: t.def, + WithLocalUpdateID: true, + LocalUpdateID: st.id, + Key: key, + Data: []sticktable.MapData{&d}, + } + + typ := peers.StickTableUpdateMessageTypeEntryUpdate + if !st.expireAt.IsZero() { + remaining := max(time.Until(st.expireAt), 0) + e.WithExpiry = true + e.Expiry = uint32(remaining.Milliseconds()) + typ = peers.StickTableUpdateMessageTypeUpdateTimed + } + + buf := make([]byte, 64) + n, err := e.Marshal(buf) + if err != nil { + return err + } + return c.sendMessage(peers.MessageClassStickTableUpdates, byte(typ), buf[:n]) +} diff --git a/internal/peerserver/peerserver_test.go b/internal/peerserver/peerserver_test.go new file mode 100644 index 0000000..8b42fc7 --- /dev/null +++ b/internal/peerserver/peerserver_test.go @@ -0,0 +1,304 @@ +package peerserver + +import ( + "bufio" + "encoding/binary" + "io" + "net" + "net/netip" + "testing" + "time" + + "github.com/dropmorepackets/haproxy-go/peers" + "github.com/dropmorepackets/haproxy-go/peers/sticktable" + "github.com/dropmorepackets/haproxy-go/pkg/encoding" +) + +func startServer(t *testing.T, srv *Server) net.Addr { + t.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { l.Close() }) + go srv.Serve(l) + return l.Addr() +} + +func dialPeer(t *testing.T, addr net.Addr, targetPeer string) (net.Conn, *bufio.Reader) { + t.Helper() + c, err := net.Dial("tcp", addr.String()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { c.Close() }) + _ = c.SetDeadline(time.Now().Add(5 * time.Second)) + + if _, err := peers.NewHandshake(targetPeer).WriteTo(c); err != nil { + t.Fatal(err) + } + return c, bufio.NewReader(c) +} + +type frame struct { + class byte + typ byte + payload []byte +} + +func readFrame(t *testing.T, br *bufio.Reader) frame { + t.Helper() + class, err := br.ReadByte() + if err != nil { + t.Fatalf("read class: %v", err) + } + typ, err := br.ReadByte() + if err != nil { + t.Fatalf("read type: %v", err) + } + var payload []byte + if typ >= 0x80 { + n, err := encoding.ReadVarint(br) + if err != nil { + t.Fatal(err) + } + payload = make([]byte, n) + if _, err := io.ReadFull(br, payload); err != nil { + t.Fatal(err) + } + } + return frame{class: class, typ: typ, payload: payload} +} + +// TestServerPushesEntry connects to the server the way HAProxy does (handshake, +// then read messages) and verifies our hand-marshalled definition + entry decode +// back correctly via the library's receive-side Unmarshal. +func TestServerPushesEntry(t *testing.T) { + srv := New("berghain_feed", "st_reputation_v4", "st_reputation_v6", time.Hour) + srv.Set(netip.MustParseAddr("1.2.3.4"), Entry{Value: 1}) + + addr := startServer(t, srv) + _, br := dialPeer(t, addr, "berghain_feed") + + if status, err := br.ReadString('\n'); err != nil || status != "200\n" { + t.Fatalf("handshake status = %q, err = %v", status, err) + } + + var def *sticktable.Definition + for { + f := readFrame(t, br) + if peers.MessageClass(f.class) != peers.MessageClassStickTableUpdates { + continue + } + + switch peers.StickTableUpdateMessageType(f.typ) { + case peers.StickTableUpdateMessageTypeStickTableDefinition: + var d sticktable.Definition + if _, err := d.Unmarshal(f.payload); err != nil { + t.Fatalf("definition unmarshal: %v", err) + } + def = &d + case peers.StickTableUpdateMessageTypeEntryUpdate: + if def == nil { + t.Fatal("entry update before a definition") + } + if def.Name != "st_reputation_v4" { + continue + } + e := sticktable.EntryUpdate{StickTable: def, WithLocalUpdateID: true} + if _, err := e.Unmarshal(f.payload); err != nil { + t.Fatalf("entry unmarshal: %v", err) + } + if e.Key.String() != "1.2.3.4" { + continue + } + if len(e.Data) != 1 || e.Data[0].String() != "1" { + t.Fatalf("entry data = %v, want gpt0=1", e.Data) + } + return // success + } + } +} + +// TestServerPushesTimedEntry verifies entries with their own expiry go out as +// timed updates carrying the remaining lifetime. +func TestServerPushesTimedEntry(t *testing.T) { + srv := New("berghain_feed", "st_reputation_v4", "st_reputation_v6", time.Hour) + srv.Set(netip.MustParseAddr("1.2.3.4"), Entry{Value: 3, ExpiresAt: time.Now().Add(4 * time.Hour)}) + + addr := startServer(t, srv) + _, br := dialPeer(t, addr, "berghain_feed") + + if status, err := br.ReadString('\n'); err != nil || status != "200\n" { + t.Fatalf("handshake status = %q, err = %v", status, err) + } + + var def *sticktable.Definition + for { + f := readFrame(t, br) + + switch peers.StickTableUpdateMessageType(f.typ) { + case peers.StickTableUpdateMessageTypeStickTableDefinition: + var d sticktable.Definition + if _, err := d.Unmarshal(f.payload); err != nil { + t.Fatalf("definition unmarshal: %v", err) + } + def = &d + case peers.StickTableUpdateMessageTypeEntryUpdate: + t.Fatal("expected a timed update, got a plain entry update") + case peers.StickTableUpdateMessageTypeUpdateTimed: + if def == nil || def.Name != "st_reputation_v4" { + continue + } + e := sticktable.EntryUpdate{StickTable: def, WithLocalUpdateID: true, WithExpiry: true} + if _, err := e.Unmarshal(f.payload); err != nil { + t.Fatalf("timed entry unmarshal: %v", err) + } + remaining := time.Duration(e.Expiry) * time.Millisecond + if remaining <= 3*time.Hour || remaining > 4*time.Hour { + t.Fatalf("timed entry expiry = %v, want just under 4h", remaining) + } + return // success + } + } +} + +// TestHandshakeRejectsWrongPeer verifies that a handshake addressed to a +// different local peer name is answered with 503 and the connection dropped. +func TestHandshakeRejectsWrongPeer(t *testing.T) { + srv := New("berghain_feed", "st_reputation_v4", "st_reputation_v6", time.Hour) + addr := startServer(t, srv) + _, br := dialPeer(t, addr, "some_other_peer") + + if status, err := br.ReadString('\n'); err != nil || status != "503\n" { + t.Fatalf("handshake status = %q, err = %v, want 503", status, err) + } + if _, err := br.ReadByte(); err != io.EOF { + t.Fatalf("expected connection close after reject, got err = %v", err) + } +} + +// TestServerAcksRemoteUpdates plays the other-peer role: teach the server a +// table and an entry update, and expect an UpdateAcknowledge with our table ID +// and update ID back. This is what keeps a multi-peer mesh from re-teaching us +// on every reconnect. +func TestServerAcksRemoteUpdates(t *testing.T) { + srv := New("berghain_feed", "st_reputation_v4", "st_reputation_v6", time.Hour) + addr := startServer(t, srv) + c, br := dialPeer(t, addr, "berghain_feed") + + if status, err := br.ReadString('\n'); err != nil || status != "200\n" { + t.Fatalf("handshake status = %q, err = %v", status, err) + } + + send := func(typ peers.StickTableUpdateMessageType, payload []byte) { + t.Helper() + var lenbuf [10]byte + n, err := encoding.PutVarint(lenbuf[:], uint64(len(payload))) + if err != nil { + t.Fatal(err) + } + msg := append([]byte{byte(peers.MessageClassStickTableUpdates), byte(typ)}, lenbuf[:n]...) + if _, err := c.Write(append(msg, payload...)); err != nil { + t.Fatal(err) + } + } + + def := &sticktable.Definition{ + StickTableID: 7, + Name: "st_src", + KeyType: sticktable.KeyTypeIPv4Address, + KeyLength: 4, + DataTypes: []sticktable.DataTypeDefinition{{DataType: sticktable.DataTypeGPT0}}, + Expiry: 60000, + } + buf := make([]byte, 256) + n, err := def.Marshal(buf) + if err != nil { + t.Fatal(err) + } + send(peers.StickTableUpdateMessageTypeStickTableDefinition, buf[:n]) + + key := sticktable.IPv4AddressKey(netip.MustParseAddr("9.9.9.9")) + data := sticktable.UnsignedIntegerData(1) + e := sticktable.EntryUpdate{ + StickTable: def, + WithLocalUpdateID: true, + LocalUpdateID: 42, + Key: &key, + Data: []sticktable.MapData{&data}, + } + n, err = e.Marshal(buf) + if err != nil { + t.Fatal(err) + } + send(peers.StickTableUpdateMessageTypeEntryUpdate, buf[:n]) + + for { + f := readFrame(t, br) + if peers.MessageClass(f.class) != peers.MessageClassStickTableUpdates || + peers.StickTableUpdateMessageType(f.typ) != peers.StickTableUpdateMessageTypeUpdateAcknowledge { + continue + } + tableID, n, err := encoding.Varint(f.payload) + if err != nil { + t.Fatal(err) + } + if tableID != 7 { + t.Fatalf("acked table ID = %d, want 7", tableID) + } + if got := binary.BigEndian.Uint32(f.payload[n:]); got != 42 { + t.Fatalf("acked update ID = %d, want 42", got) + } + return // success + } +} + +// TestDeleteConvergesOnResync verifies a deleted entry is taught as a zeroed, +// expiring entry to fresh connections, so peers that missed the live delete +// still converge. +func TestDeleteConvergesOnResync(t *testing.T) { + srv := New("berghain_feed", "st_reputation_v4", "st_reputation_v6", time.Hour) + srv.Set(netip.MustParseAddr("1.2.3.4"), Entry{Value: 1}) + srv.Delete(netip.MustParseAddr("1.2.3.4")) + + if srv.Len() != 0 { + t.Fatalf("Len() = %d after delete, want 0", srv.Len()) + } + + addr := startServer(t, srv) + _, br := dialPeer(t, addr, "berghain_feed") + + if status, err := br.ReadString('\n'); err != nil || status != "200\n" { + t.Fatalf("handshake status = %q, err = %v", status, err) + } + + var def *sticktable.Definition + for { + f := readFrame(t, br) + + switch peers.StickTableUpdateMessageType(f.typ) { + case peers.StickTableUpdateMessageTypeStickTableDefinition: + var d sticktable.Definition + if _, err := d.Unmarshal(f.payload); err != nil { + t.Fatal(err) + } + def = &d + case peers.StickTableUpdateMessageTypeUpdateTimed: + if def == nil || def.Name != "st_reputation_v4" { + continue + } + e := sticktable.EntryUpdate{StickTable: def, WithLocalUpdateID: true, WithExpiry: true} + if _, err := e.Unmarshal(f.payload); err != nil { + t.Fatal(err) + } + if e.Key.String() != "1.2.3.4" { + continue + } + if e.Data[0].String() != "0" { + t.Fatalf("resynced deleted entry has gpt0=%s, want 0", e.Data[0]) + } + return // success + } + } +} diff --git a/internal/reputation/crowdsec.go b/internal/reputation/crowdsec.go new file mode 100644 index 0000000..f53cd39 --- /dev/null +++ b/internal/reputation/crowdsec.go @@ -0,0 +1,168 @@ +package reputation + +import ( + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "net/netip" + "net/url" + "strings" + "time" + + "github.com/DropMorePackets/berghain/internal/peerserver" +) + +// The decision stream is a small JSON document; limit reads defensively. +const crowdsecMaxStreamLength = 64 << 20 + +type csDecision struct { + Scope string `json:"scope"` + Value string `json:"value"` + Type string `json:"type"` + Duration string `json:"duration"` +} + +type csStream struct { + New []csDecision `json:"new"` + Deleted []csDecision `json:"deleted"` +} + +// crowdsecLoop implements a CrowdSec bouncer: it polls the LAPI decision +// stream and maintains the "crowdsec" reputation source from it. The first +// poll (and the first after any error) passes startup=true so the LAPI +// replays the full active decision set. +func (s *Service) crowdsecLoop(ctx context.Context) { + slog.InfoContext(ctx, "polling the CrowdSec decision stream", + "url", s.cfg.CrowdSec.URL, "interval", s.cfg.CrowdSec.Interval.String()) + + t := time.NewTicker(s.cfg.CrowdSec.Interval) + defer t.Stop() + + startup := true + poll := func() { + if err := s.pollCrowdSec(ctx, startup); err != nil { + if ctx.Err() == nil { + slog.ErrorContext(ctx, "crowdsec decision stream poll failed", "error", err) + } + // Resync from scratch once the LAPI is reachable again: deletions + // that happened while we could not poll are gone from the stream. + startup = true + return + } + startup = false + } + + poll() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + poll() + } + } +} + +func (s *Service) pollCrowdSec(ctx context.Context, startup bool) error { + u, err := url.Parse(s.cfg.CrowdSec.URL) + if err != nil { + return err + } + u = u.JoinPath("/v1/decisions/stream") + q := u.Query() + q.Set("startup", fmt.Sprintf("%t", startup)) + u.RawQuery = q.Encode() + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil) + if err != nil { + return err + } + req.Header.Set("X-Api-Key", s.cfg.CrowdSec.APIKey) + + resp, err := s.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("unexpected status %d", resp.StatusCode) + } + + var stream csStream + if err := json.NewDecoder(io.LimitReader(resp.Body, crowdsecMaxStreamLength)).Decode(&stream); err != nil { + return err + } + + // On startup the stream is the full state; otherwise apply the delta on + // top of what we already know. + var state map[netip.Addr]peerserver.Entry + if startup { + state = make(map[netip.Addr]peerserver.Entry) + } else { + state = s.snapshotSource("crowdsec") + if state == nil { + state = make(map[netip.Addr]peerserver.Entry) + } + } + + now := time.Now() + var skippedScopes, applied, deleted int + for _, d := range stream.New { + a, ok := decisionAddr(d) + if !ok { + skippedScopes++ + continue + } + e := peerserver.Entry{Value: decisionAction(d.Type)} + if dur, err := time.ParseDuration(d.Duration); err == nil && dur > 0 { + e.ExpiresAt = now.Add(dur) + } + state[a] = e + applied++ + } + for _, d := range stream.Deleted { + a, ok := decisionAddr(d) + if !ok { + continue + } + delete(state, a) + deleted++ + } + + if startup || applied > 0 || deleted > 0 { + s.setSource("crowdsec", state) + } + if skippedScopes > 0 { + slog.WarnContext(ctx, "skipped non-ip crowdsec decisions", + "count", skippedScopes, "reason", "stick-tables cannot prefix-match; range decisions need map files") + } + return nil +} + +func decisionAddr(d csDecision) (netip.Addr, bool) { + if !strings.EqualFold(d.Scope, "ip") { + return netip.Addr{}, false + } + a, err := netip.ParseAddr(d.Value) + if err != nil { + return netip.Addr{}, false + } + return a.Unmap(), true +} + +// decisionAction maps a CrowdSec remediation type onto our gpt0 actions. +// Unknown remediation types fail closed to a block, matching how CrowdSec +// bouncers treat custom decision types they do not implement. +func decisionAction(typ string) uint32 { + switch strings.ToLower(typ) { + case "captcha": + return actionChallenge + case "ban": + return actionBlock + default: + return actionBlock + } +} diff --git a/internal/reputation/feeds.go b/internal/reputation/feeds.go new file mode 100644 index 0000000..73c5c29 --- /dev/null +++ b/internal/reputation/feeds.go @@ -0,0 +1,203 @@ +package reputation + +import ( + "bufio" + "context" + "fmt" + "io" + "net/http" + "net/netip" + "os" + "path/filepath" + "strings" + "time" + + "log/slog" + + "github.com/DropMorePackets/berghain/internal/peerserver" +) + +// cidrSource is a feed of CIDRs written verbatim to an ACL/map file. +// Stick-tables key on exact addresses, so prefix feeds must stay files. +type cidrSource struct { + name string + urls []string + outFile string +} + +var cidrSources = []cidrSource{ + { + name: "cloudflare", + urls: []string{"https://www.cloudflare.com/ips-v4", "https://www.cloudflare.com/ips-v6"}, + outFile: "cloudflare-ips.lst", + }, +} + +// ipSource is a feed of individual IPs served over the peers protocol. +type ipSource struct { + name string + urls []string + action uint32 +} + +var torExitSource = ipSource{ + name: "tor_exit", + urls: []string{"https://check.torproject.org/torbulkexitlist"}, + action: actionChallenge, +} + +func (s *Service) fetch(ctx context.Context, url string) (string, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return "", err + } + resp, err := s.client.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return "", fmt.Errorf("unexpected status %d", resp.StatusCode) + } + b, err := io.ReadAll(io.LimitReader(resp.Body, 64<<20)) + if err != nil { + return "", err + } + return string(b), nil +} + +// cidrOnly keeps non-comment, non-empty lines verbatim (they are already CIDRs). +func cidrOnly(line string) string { + line = strings.TrimSpace(line) + if line == "" || strings.HasPrefix(line, "#") { + return "" + } + return line +} + +func parseLines(body string, parse func(string) string) []string { + var out []string + sc := bufio.NewScanner(strings.NewReader(body)) + sc.Buffer(make([]byte, 0, 64*1024), 1<<20) + for sc.Scan() { + if line := parse(sc.Text()); line != "" { + out = append(out, line) + } + } + return out +} + +// parseIPs parses one IP per line, skipping comments and blanks. +func parseIPs(body string) []netip.Addr { + var out []netip.Addr + sc := bufio.NewScanner(strings.NewReader(body)) + sc.Buffer(make([]byte, 0, 64*1024), 1<<20) + for sc.Scan() { + s := strings.TrimSpace(sc.Text()) + if s == "" || strings.HasPrefix(s, "#") { + continue + } + if a, err := netip.ParseAddr(s); err == nil { + out = append(out, a) + } + } + return out +} + +// writeAtomic writes lines to path via a temp file + rename. +func writeAtomic(path string, lines []string) error { + dir := filepath.Dir(path) + tmp, err := os.CreateTemp(dir, ".feedupdater-*") + if err != nil { + return err + } + tmpName := tmp.Name() + defer os.Remove(tmpName) + + w := bufio.NewWriter(tmp) + fmt.Fprintln(w, "# Generated by berghain reputation feeds — do not edit by hand.") + for _, l := range lines { + fmt.Fprintln(w, l) + } + if err := w.Flush(); err != nil { + tmp.Close() + return err + } + if err := tmp.Close(); err != nil { + return err + } + return os.Rename(tmpName, path) +} + +func (s *Service) updateCIDR(ctx context.Context, src cidrSource) error { + var lines []string + for _, u := range src.urls { + body, err := s.fetch(ctx, u) + if err != nil { + return fmt.Errorf("fetch %s: %w", u, err) + } + lines = append(lines, parseLines(body, cidrOnly)...) + } + if len(lines) == 0 { + return fmt.Errorf("no entries parsed") + } + return writeAtomic(filepath.Join(s.cfg.MapsDir, src.outFile), lines) +} + +// refreshStatic fetches the URL feeds and the optional banlist into the +// "static" source and rewrites the CIDR map files. +func (s *Service) refreshStatic(ctx context.Context) { + if s.cfg.MapsDir != "" { + for _, src := range cidrSources { + if err := s.updateCIDR(ctx, src); err != nil { + slog.ErrorContext(ctx, "cidr feed update failed", "source", src.name, "error", err) + continue + } + slog.InfoContext(ctx, "cidr feed updated", "source", src.name, + "file", filepath.Join(s.cfg.MapsDir, src.outFile)) + } + } + + entries := make(map[netip.Addr]peerserver.Entry) + + if s.cfg.TorExits == nil || *s.cfg.TorExits { + for _, u := range torExitSource.urls { + body, err := s.fetch(ctx, u) + if err != nil { + slog.ErrorContext(ctx, "feed fetch failed", "source", torExitSource.name, "url", u, "error", err) + continue + } + for _, a := range parseIPs(body) { + entries[a.Unmap()] = peerserver.Entry{Value: torExitSource.action} + } + } + } + + if s.cfg.Banlist != "" { + b, err := os.ReadFile(s.cfg.Banlist) + if err != nil { + slog.ErrorContext(ctx, "reading banlist", "path", s.cfg.Banlist, "error", err) + } else { + for _, a := range parseIPs(string(b)) { + entries[a.Unmap()] = peerserver.Entry{Value: actionBlock} // blocks override challenges + } + } + } + + s.setSource("static", entries) +} + +func (s *Service) staticFeedLoop(ctx context.Context) { + t := time.NewTicker(s.cfg.Interval) + defer t.Stop() + + s.refreshStatic(ctx) + for { + select { + case <-ctx.Done(): + return + case <-t.C: + s.refreshStatic(ctx) + } + } +} diff --git a/internal/reputation/reputation_test.go b/internal/reputation/reputation_test.go new file mode 100644 index 0000000..e5dea9d --- /dev/null +++ b/internal/reputation/reputation_test.go @@ -0,0 +1,198 @@ +package reputation + +import ( + "context" + "net/http" + "net/http/httptest" + "net/netip" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/DropMorePackets/berghain/internal/peerserver" +) + +func newTestService(t *testing.T) *Service { + t.Helper() + return New(Config{PeerListen: "127.0.0.1:0"}) +} + +func Test_stronger(t *testing.T) { + later := time.Now().Add(4 * time.Hour) + earlier := time.Now().Add(time.Hour) + + block := peerserver.Entry{Value: actionBlock, ExpiresAt: earlier} + challenge := peerserver.Entry{Value: actionChallenge, ExpiresAt: later} + unlimited := peerserver.Entry{Value: actionChallenge} + + if got := stronger(challenge, block); got != block { + t.Errorf("stronger(challenge, block) = %+v, want the block", got) + } + if got := stronger(block, challenge); got != block { + t.Errorf("stronger(block, challenge) = %+v, want the block", got) + } + if got := stronger(challenge, unlimited); got != unlimited { + t.Errorf("stronger(challenge, unlimited) = %+v, want the unlimited entry", got) + } + shorter := peerserver.Entry{Value: actionChallenge, ExpiresAt: earlier} + if got := stronger(shorter, challenge); got != challenge { + t.Errorf("stronger(shorter, longer) = %+v, want the longer entry", got) + } +} + +func TestSourceMergeDoesNotClobber(t *testing.T) { + s := newTestService(t) + tor := netip.MustParseAddr("192.0.2.1") + banned := netip.MustParseAddr("192.0.2.2") + both := netip.MustParseAddr("192.0.2.3") + + s.setSource("static", map[netip.Addr]peerserver.Entry{ + tor: {Value: actionChallenge}, + both: {Value: actionChallenge}, + }) + s.setSource("crowdsec", map[netip.Addr]peerserver.Entry{ + banned: {Value: actionBlock}, + both: {Value: actionBlock}, + }) + + // A crowdsec update must not clear the static entries and vice versa. + if e, ok := s.srv.Get(tor); !ok || e.Value != actionChallenge { + t.Errorf("tor entry = %+v, %v; want challenge", e, ok) + } + if e, ok := s.srv.Get(banned); !ok || e.Value != actionBlock { + t.Errorf("banned entry = %+v, %v; want block", e, ok) + } + if e, ok := s.srv.Get(both); !ok || e.Value != actionBlock { + t.Errorf("contested entry = %+v, %v; want block to win", e, ok) + } + + // Dropping the crowdsec decisions clears only crowdsec-owned entries. + s.setSource("crowdsec", map[netip.Addr]peerserver.Entry{}) + if _, ok := s.srv.Get(banned); ok { + t.Error("banned entry survived crowdsec clearing its source") + } + if e, ok := s.srv.Get(both); !ok || e.Value != actionChallenge { + t.Errorf("contested entry after crowdsec clear = %+v, %v; want static challenge", e, ok) + } + if _, ok := s.srv.Get(tor); !ok { + t.Error("static tor entry lost after crowdsec update") + } +} + +func TestPollCrowdSec(t *testing.T) { + var startups []string + lapi := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/decisions/stream" { + http.NotFound(w, r) + return + } + if r.Header.Get("X-Api-Key") != "test-key" { + w.WriteHeader(http.StatusForbidden) + return + } + startup := r.URL.Query().Get("startup") + startups = append(startups, startup) + w.Header().Set("Content-Type", "application/json") + + if startup == "true" { + w.Write([]byte(`{"new": [ + {"scope": "Ip", "value": "198.51.100.1", "type": "ban", "duration": "3h59m57s"}, + {"scope": "Ip", "value": "198.51.100.2", "type": "captcha", "duration": "1h"}, + {"scope": "Range", "value": "203.0.113.0/24", "type": "ban", "duration": "4h"} + ], "deleted": []}`)) + return + } + w.Write([]byte(`{"new": [ + {"scope": "Ip", "value": "198.51.100.3", "type": "unknown-custom", "duration": "1h"} + ], "deleted": [ + {"scope": "Ip", "value": "198.51.100.1", "type": "ban", "duration": "-1s"} + ]}`)) + })) + defer lapi.Close() + + s := New(Config{ + PeerListen: "127.0.0.1:0", + CrowdSec: CrowdSecConfig{URL: lapi.URL, APIKey: "test-key"}, + }) + ctx := context.Background() + + if err := s.pollCrowdSec(ctx, true); err != nil { + t.Fatal(err) + } + + if e, ok := s.srv.Get(netip.MustParseAddr("198.51.100.1")); !ok || e.Value != actionBlock { + t.Errorf("ban decision = %+v, %v; want block", e, ok) + } else if until := time.Until(e.ExpiresAt); until <= 3*time.Hour || until > 4*time.Hour { + t.Errorf("ban decision expiry in %v, want just under 4h", until) + } + if e, ok := s.srv.Get(netip.MustParseAddr("198.51.100.2")); !ok || e.Value != actionChallenge { + t.Errorf("captcha decision = %+v, %v; want challenge", e, ok) + } + if _, ok := s.srv.Get(netip.MustParseAddr("203.0.113.0")); ok { + t.Error("range decision must be skipped, not applied to its base address") + } + + if err := s.pollCrowdSec(ctx, false); err != nil { + t.Fatal(err) + } + + if _, ok := s.srv.Get(netip.MustParseAddr("198.51.100.1")); ok { + t.Error("deleted decision still present after delta poll") + } + if e, ok := s.srv.Get(netip.MustParseAddr("198.51.100.3")); !ok || e.Value != actionBlock { + t.Errorf("unknown remediation type = %+v, %v; want fail-closed block", e, ok) + } + if e, ok := s.srv.Get(netip.MustParseAddr("198.51.100.2")); !ok || e.Value != actionChallenge { + t.Errorf("untouched decision after delta = %+v, %v; want unchanged challenge", e, ok) + } + + if len(startups) != 2 || startups[0] != "true" || startups[1] != "false" { + t.Errorf("startup params = %v, want [true false]", startups) + } +} + +func TestRefreshStatic(t *testing.T) { + feed := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/torbulkexitlist": + w.Write([]byte("# comment\n192.0.2.10\n192.0.2.11\n")) + case "/ips-v4": + w.Write([]byte("203.0.113.0/24\n")) + default: + http.NotFound(w, r) + } + })) + defer feed.Close() + + origTor, origCIDR := torExitSource, cidrSources + t.Cleanup(func() { torExitSource, cidrSources = origTor, origCIDR }) + torExitSource.urls = []string{feed.URL + "/torbulkexitlist"} + cidrSources = []cidrSource{{name: "cloudflare", urls: []string{feed.URL + "/ips-v4"}, outFile: "cloudflare-ips.lst"}} + + mapsDir := t.TempDir() + banlist := filepath.Join(t.TempDir(), "banlist.txt") + if err := os.WriteFile(banlist, []byte("192.0.2.11\n"), 0o644); err != nil { + t.Fatal(err) + } + + s := New(Config{PeerListen: "127.0.0.1:0", MapsDir: mapsDir, Banlist: banlist}) + s.refreshStatic(context.Background()) + + if e, ok := s.srv.Get(netip.MustParseAddr("192.0.2.10")); !ok || e.Value != actionChallenge { + t.Errorf("tor exit = %+v, %v; want challenge", e, ok) + } + // The banlist entry overlaps the tor list; the block must win. + if e, ok := s.srv.Get(netip.MustParseAddr("192.0.2.11")); !ok || e.Value != actionBlock { + t.Errorf("banlisted tor exit = %+v, %v; want block", e, ok) + } + + b, err := os.ReadFile(filepath.Join(mapsDir, "cloudflare-ips.lst")) + if err != nil { + t.Fatal(err) + } + if want := "203.0.113.0/24\n"; !strings.Contains(string(b), want) { + t.Errorf("cidr map file %q does not contain %q", b, want) + } +} diff --git a/internal/reputation/service.go b/internal/reputation/service.go new file mode 100644 index 0000000..7f48b18 --- /dev/null +++ b/internal/reputation/service.go @@ -0,0 +1,218 @@ +// Package reputation feeds IP reputation into HAProxy. +// +// It combines multiple sources — the CrowdSec LAPI decision stream and static +// URL feeds — into one desired state and pushes individual-IP entries into +// HAProxy stick-tables live over the peers protocol (internal/peerserver). +// CIDR feeds are written to map files instead, because stick-tables cannot do +// longest-prefix matching. +// +// The same service powers both deployment modes: embedded in the SPOP agent +// (cmd/spop, `reputation:` config section) for the simple single-container +// setup, and standalone (cmd/feedupdater) where scaling or isolation calls +// for it. +package reputation + +import ( + "context" + "log/slog" + "maps" + "net" + "net/http" + "net/netip" + "os" + "sync" + "time" + + "github.com/DropMorePackets/berghain/internal/peerserver" +) + +// Reputation actions, encoded as the gpt0 tag value in the stick-tables. +const ( + actionNone uint32 = 0 // no entry / cleared + actionBlock uint32 = 1 // silent-drop + actionChallenge uint32 = 3 // raise to the highest challenge level +) + +type CrowdSecConfig struct { + // URL of the local API, e.g. http://crowdsec:8080. Empty disables the source. + URL string `yaml:"url"` + // APIKey is a bouncer key (cscli bouncers add berghain). Falls back to the + // CROWDSEC_API_KEY environment variable so the secret can stay out of files. + APIKey string `yaml:"api_key"` + // Interval between decision-stream polls. Default 10s. + Interval time.Duration `yaml:"interval"` +} + +type Config struct { + // PeerListen is the peers-protocol listen address (e.g. 0.0.0.0:10001). + // It must match this peer's entry in the HAProxy `peers` section. + PeerListen string `yaml:"peer_listen"` + // LocalPeer is our name in the HAProxy peers section. Default berghain_feed. + LocalPeer string `yaml:"local_peer"` + // TableV4/TableV6 name the reputation stick-tables. Defaults + // st_reputation_v4 / st_reputation_v6. + TableV4 string `yaml:"table_v4"` + TableV6 string `yaml:"table_v6"` + // TableExpiry mirrors the `expire` of the HAProxy stick-tables and bounds + // entries that carry no per-decision duration. Default 24h. + TableExpiry time.Duration `yaml:"table_expiry"` + + // MapsDir receives the CIDR feed files. Empty disables CIDR feeds. + MapsDir string `yaml:"maps_dir"` + // Interval between static feed refreshes. Default 6h. + Interval time.Duration `yaml:"interval"` + // Banlist optionally names a local file of individual IPs to block. + Banlist string `yaml:"banlist"` + // TorExits toggles the Tor exit-node feed. Default true. + TorExits *bool `yaml:"tor_exits"` + + CrowdSec CrowdSecConfig `yaml:"crowdsec"` +} + +// Enabled reports whether the config activates the service at all. +func (c Config) Enabled() bool { + return c.PeerListen != "" +} + +func (c Config) withDefaults() Config { + if c.LocalPeer == "" { + c.LocalPeer = "berghain_feed" + } + if c.TableV4 == "" { + c.TableV4 = "st_reputation_v4" + } + if c.TableV6 == "" { + c.TableV6 = "st_reputation_v6" + } + if c.TableExpiry == 0 { + c.TableExpiry = 24 * time.Hour + } + if c.Interval == 0 { + c.Interval = 6 * time.Hour + } + if c.CrowdSec.Interval == 0 { + c.CrowdSec.Interval = 10 * time.Second + } + if c.CrowdSec.APIKey == "" { + c.CrowdSec.APIKey = os.Getenv("CROWDSEC_API_KEY") + } + return c +} + +// Service runs the peers listener and the feed loops. +type Service struct { + cfg Config + srv *peerserver.Server + client *http.Client + + mu sync.Mutex + sources map[string]map[netip.Addr]peerserver.Entry +} + +func New(cfg Config) *Service { + cfg = cfg.withDefaults() + return &Service{ + cfg: cfg, + srv: peerserver.New(cfg.LocalPeer, cfg.TableV4, cfg.TableV6, cfg.TableExpiry), + client: &http.Client{Timeout: 30 * time.Second}, + sources: make(map[string]map[netip.Addr]peerserver.Entry), + } +} + +// Run serves the peers protocol and refreshes all feeds until ctx is done. +func (s *Service) Run(ctx context.Context) error { + l, err := net.Listen("tcp", s.cfg.PeerListen) + if err != nil { + return err + } + slog.InfoContext(ctx, "serving reputation over the peers protocol", + "listen", s.cfg.PeerListen, "peer", s.cfg.LocalPeer) + + var wg sync.WaitGroup + + wg.Add(1) + go func() { + defer wg.Done() + <-ctx.Done() + _ = l.Close() + }() + + wg.Add(1) + go func() { + defer wg.Done() + s.staticFeedLoop(ctx) + }() + + if s.cfg.CrowdSec.URL != "" { + wg.Add(1) + go func() { + defer wg.Done() + s.crowdsecLoop(ctx) + }() + } + + err = s.srv.Serve(l) + wg.Wait() + if ctx.Err() != nil { + return nil // closed by shutdown + } + return err +} + +// setSource replaces one source's desired state and pushes the merged state of +// all sources to the connected peers. Merging (instead of letting each source +// write directly) prevents a static-feed refresh from clearing CrowdSec +// decisions and vice versa. +func (s *Service) setSource(name string, entries map[netip.Addr]peerserver.Entry) { + s.mu.Lock() + s.sources[name] = entries + + merged := make(map[netip.Addr]peerserver.Entry) + for _, src := range s.sources { + for a, e := range src { + if cur, ok := merged[a]; ok { + e = stronger(cur, e) + } + merged[a] = e + } + } + s.mu.Unlock() + + s.srv.ReplaceAll(merged) + slog.Info("reputation updated", "source", name, "entries", s.srv.Len()) +} + +// snapshotSource returns a copy of one source's current desired state. +func (s *Service) snapshotSource(name string) map[netip.Addr]peerserver.Entry { + s.mu.Lock() + defer s.mu.Unlock() + return maps.Clone(s.sources[name]) +} + +// stronger picks the entry that must win when two sources disagree about one +// address: a block always beats a challenge, higher challenge levels beat +// lower ones, and for equal actions the longer-lived entry wins. +func stronger(a, b peerserver.Entry) peerserver.Entry { + if a.Value != b.Value { + switch { + case a.Value == actionBlock: + return a + case b.Value == actionBlock: + return b + case a.Value > b.Value: + return a + default: + return b + } + } + switch { + case a.ExpiresAt.IsZero(): + return a + case b.ExpiresAt.IsZero(): + return b + case a.ExpiresAt.After(b.ExpiresAt): + return a + default: + return b + } +} diff --git a/test/e2e/peers_test.go b/test/e2e/peers_test.go new file mode 100644 index 0000000..63a8429 --- /dev/null +++ b/test/e2e/peers_test.go @@ -0,0 +1,261 @@ +//go:build e2e + +package e2e + +import ( + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "sync" + "testing" + "time" +) + +// TestReputationPeersMesh runs the reputation service (cmd/feedupdater) as one +// peer in a mesh with TWO HAProxy instances and asserts, via the admin +// sockets, that: +// +// - banlist and CrowdSec decisions appear in the reputation stick-tables of +// BOTH instances (live push over the peers protocol), +// - the CrowdSec decision's duration is honored (timed entry update), +// - a deleted decision is zeroed everywhere, +// - the HAProxies still replicate their own tables through the mesh while +// the daemon is a member (it acknowledges their updates), and +// - a restarted HAProxy resyncs back to the full state. +func TestReputationPeersMesh(t *testing.T) { + if _, err := exec.LookPath("haproxy"); err != nil { + t.Skip("haproxy binary not available") + } + + const ( + peerA = "127.0.0.1:19100" + peerB = "127.0.0.1:19101" + peerFeed = "127.0.0.1:19102" + feA = "127.0.0.1:19110" + feB = "127.0.0.1:19111" + + banlistIP = "198.51.100.10" + crowdsecIP = "198.51.100.20" + ) + + dir := t.TempDir() + + // --- fake CrowdSec LAPI ------------------------------------------------ + var ( + lapiMu sync.Mutex + startupBody = `{"new":[{"scope":"Ip","value":"` + crowdsecIP + `","type":"ban","duration":"4h"}],"deleted":[]}` + deltas []string + ) + lapi := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("X-Api-Key") != "e2e-key" { + w.WriteHeader(http.StatusForbidden) + return + } + lapiMu.Lock() + defer lapiMu.Unlock() + w.Header().Set("Content-Type", "application/json") + if r.URL.Query().Get("startup") == "true" { + io.WriteString(w, startupBody) + return + } + if len(deltas) > 0 { + io.WriteString(w, deltas[0]) + deltas = deltas[1:] + return + } + io.WriteString(w, `{"new":[],"deleted":[]}`) + })) + defer lapi.Close() + + // --- reputation daemon -------------------------------------------------- + banlist := filepath.Join(dir, "banlist.txt") + if err := os.WriteFile(banlist, []byte(banlistIP+"\n"), 0o644); err != nil { + t.Fatal(err) + } + + feedBin := filepath.Join(dir, "feedupdater") + build := exec.Command("go", "-C", "../..", "build", "-o", feedBin, "./cmd/feedupdater") + if out, err := build.CombinedOutput(); err != nil { + t.Fatalf("building feedupdater: %v\n%s", err, out) + } + + feed := exec.Command(feedBin, + "-peer-listen", peerFeed, + "-banlist", banlist, + "-tor-exits=false", + "-interval", "2s", + "-crowdsec-url", lapi.URL, + "-crowdsec-interval", "1s", + ) + feed.Env = append(os.Environ(), "CROWDSEC_API_KEY=e2e-key") + feedLog := &strings.Builder{} + feed.Stdout, feed.Stderr = feedLog, feedLog + if err := feed.Start(); err != nil { + t.Fatal(err) + } + defer func() { + _ = feed.Process.Kill() + _, _ = feed.Process.Wait() + if t.Failed() { + t.Logf("feedupdater log:\n%s", feedLog.String()) + } + }() + + // --- two HAProxy instances ---------------------------------------------- + haproxyCfg := func(sock, fe string) string { + return ` +global + stats socket ` + sock + ` level admin + log stdout format raw local0 + +defaults + mode http + timeout client 5s + timeout server 5s + timeout connect 5s + +peers test_peers + peer haproxy_a ` + peerA + ` + peer haproxy_b ` + peerB + ` + peer berghain_feed ` + peerFeed + ` + +frontend fe + bind ` + fe + ` + http-request track-sc1 src table st_visits + http-request return status 200 content-type "text/plain" string "ok" + +backend st_visits + stick-table type ip size 1m expire 10m store http_req_cnt peers test_peers + +backend st_reputation_v4 + stick-table type ip size 1m expire 24h store gpt0 peers test_peers + +backend st_reputation_v6 + stick-table type ipv6 size 1m expire 24h store gpt0 peers test_peers +` + } + + sockA := filepath.Join(dir, "a.sock") + sockB := filepath.Join(dir, "b.sock") + cfgA := filepath.Join(dir, "a.cfg") + cfgB := filepath.Join(dir, "b.cfg") + if err := os.WriteFile(cfgA, []byte(haproxyCfg(sockA, feA)), 0o644); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(cfgB, []byte(haproxyCfg(sockB, feB)), 0o644); err != nil { + t.Fatal(err) + } + + startHAProxy := func(localPeer, cfg string) *exec.Cmd { + t.Helper() + cmd := exec.Command("haproxy", "-db", "-L", localPeer, "-f", cfg) + log := &strings.Builder{} + cmd.Stdout, cmd.Stderr = log, log + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _ = cmd.Process.Kill() + _, _ = cmd.Process.Wait() + if t.Failed() { + t.Logf("haproxy %s log:\n%s", localPeer, log.String()) + } + }) + return cmd + } + haproxyA := startHAProxy("haproxy_a", cfgA) + startHAProxy("haproxy_b", cfgB) + + // --- helpers -------------------------------------------------------------- + showTable := func(sock, table string) string { + c, err := net.Dial("unix", sock) + if err != nil { + return "" + } + defer c.Close() + _ = c.SetDeadline(time.Now().Add(2 * time.Second)) + fmt.Fprintf(c, "show table %s\n", table) + b, _ := io.ReadAll(c) + return string(b) + } + + // entryLine returns the table line for a key, if present. + entryLine := func(sock, table, key string) (string, bool) { + for _, line := range strings.Split(showTable(sock, table), "\n") { + if strings.Contains(line, "key="+key+" ") { + return line, true + } + } + return "", false + } + + waitFor := func(what string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(30 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(250 * time.Millisecond) + } + t.Fatalf("timed out waiting for %s", what) + } + + hasValue := func(sock, key, gpt0 string) bool { + line, ok := entryLine(sock, "st_reputation_v4", key) + return ok && strings.Contains(line, "gpt0="+gpt0) + } + + // --- assertions ----------------------------------------------------------- + + // 1. Both feeds reach both HAProxy instances. + for _, sock := range []string{sockA, sockB} { + waitFor("banlist entry on "+sock, func() bool { return hasValue(sock, banlistIP, "1") }) + waitFor("crowdsec entry on "+sock, func() bool { return hasValue(sock, crowdsecIP, "1") }) + } + + // 2. The CrowdSec decision's 4h duration is honored via a timed update: + // its expiry must sit well below the 24h table default. + line, _ := entryLine(sockA, "st_reputation_v4", crowdsecIP) + exp := 0 + for _, f := range strings.Fields(line) { + if v, ok := strings.CutPrefix(f, "exp="); ok { + exp, _ = strconv.Atoi(v) + } + } + if exp <= 0 || exp > int((4*time.Hour+time.Minute).Milliseconds()) { + t.Fatalf("crowdsec entry expiry = %dms, want ~4h (timed update): %q", exp, line) + } + + // 3. The HAProxies still replicate their own tables through the mesh while + // the daemon is connected as a peer. + if _, err := http.Get("http://" + feA + "/"); err != nil { + t.Fatal(err) + } + waitFor("st_visits replication a->b", func() bool { + _, ok := entryLine(sockB, "st_visits", "127.0.0.1") + return ok + }) + + // 4. A deleted decision is zeroed on both instances. + lapiMu.Lock() + startupBody = `{"new":[],"deleted":[]}` + deltas = append(deltas, `{"new":[],"deleted":[{"scope":"Ip","value":"`+crowdsecIP+`","type":"ban","duration":"-1s"}]}`) + lapiMu.Unlock() + for _, sock := range []string{sockA, sockB} { + waitFor("crowdsec delete on "+sock, func() bool { return hasValue(sock, crowdsecIP, "0") }) + } + + // 5. A restarted HAProxy resyncs the reputation state from the mesh. + _ = haproxyA.Process.Kill() + _, _ = haproxyA.Process.Wait() + startHAProxy("haproxy_a", cfgA) + waitFor("banlist entry after restart", func() bool { return hasValue(sockA, banlistIP, "1") }) +}