diff --git a/bouncer.go b/bouncer.go index d362996..6b5cf93 100644 --- a/bouncer.go +++ b/bouncer.go @@ -392,6 +392,16 @@ func (bouncer *Bouncer) ServeHTTP(rw http.ResponseWriter, req *http.Request) { // Right here if we cannot join the stream we forbid the request to go on. if bouncer.crowdsecMode == configuration.StreamMode || bouncer.crowdsecMode == configuration.AloneMode { if isCrowdsecStreamHealthy { + cidrValue, cidrErr := bouncer.cacheClient.GetCIDR(remoteIP) + if cidrErr == nil { + bouncer.log.Debug(fmt.Sprintf("ServeHTTP ip:%s cidr:hit isBanned:%v", remoteIP, cidrValue)) + if cidrValue == cache.NoBannedValue { + bouncer.handleNextServeHTTP(rw, req, remoteIP) + } else { + bouncer.handleRemediationServeHTTP(rw, req, remoteIP, cidrValue) + } + return + } bouncer.handleNextServeHTTP(rw, req, remoteIP) } else { bouncer.log.Debug(fmt.Sprintf("ServeHTTP isCrowdsecStreamHealthy:false ip:%s updateFailure:%d", remoteIP, updateFailure)) @@ -669,11 +679,19 @@ func handleStreamCache(bouncer *Bouncer) error { default: bouncer.log.Info("handleStreamCache:unknownType " + decision.Type) } - bouncer.cacheClient.Set(decision.Value, value, int64(duration.Seconds())) + if strings.Contains(decision.Value, "/") { + bouncer.cacheClient.SetCIDR(decision.Value, value, int64(duration.Seconds())) + } else { + bouncer.cacheClient.Set(decision.Value, value, int64(duration.Seconds())) + } } } for _, decision := range stream.Deleted { - bouncer.cacheClient.Delete(decision.Value) + if strings.Contains(decision.Value, "/") { + bouncer.cacheClient.DeleteCIDR(decision.Value) + } else { + bouncer.cacheClient.Delete(decision.Value) + } } bouncer.log.Debug("handleStreamCache:updated") isCrowdsecStreamStartup = false diff --git a/pkg/cache/cache.go b/pkg/cache/cache.go index 3059f01..f223b37 100644 --- a/pkg/cache/cache.go +++ b/pkg/cache/cache.go @@ -10,6 +10,8 @@ import ( ttl_map "github.com/leprosus/golang-ttl-map" simpleredis "github.com/maxlerebourg/simpleredis" + + ip "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/ip" ) const ( @@ -25,6 +27,8 @@ const ( CacheMiss = "cache:miss" // CacheUnreachable error string when cache is unreachable. CacheUnreachable = "cache:unreachable" + // cidrPrefix store all cidr within the same key + cidrPrefix = "cidr:" ) //nolint:gochecknoglobals @@ -144,3 +148,35 @@ func (c *Client) Set(key string, value string, duration int64) { c.log.Debug(fmt.Sprintf("cache:Set key:%v value:%v duration:%vs", key, value, duration)) c.cache.set(key, value, duration) } + +func (c *Client) DeleteCIDR(cidr string) { + cidr = ip.NormalizeCIDR(cidr) + if cidr == "" { + return + } + c.cache.delete(cidrPrefix + cidr) + c.log.Debug(fmt.Sprintf("cache:DeleteCIDR cidr:%v", cidr)) +} + +func (c *Client) GetCIDR(ipStr string) (string, error) { + keys := ip.CIDRKeys(ipStr) + if keys == nil { + return "", errors.New(CacheMiss) + } + for _, key := range keys { + value, err := c.cache.get(cidrPrefix + key) + if err == nil { + return value, nil + } + } + return "", errors.New(CacheMiss) +} + +func (c *Client) SetCIDR(cidr, value string, duration int64) { + cidr = ip.NormalizeCIDR(cidr) + if cidr == "" { + return + } + c.cache.set(cidrPrefix+cidr, value, duration) + c.log.Debug(fmt.Sprintf("cache:SetCIDR cidr:%v value:%v duration:%vs", cidr, value, duration)) +} diff --git a/pkg/ip/cidr.go b/pkg/ip/cidr.go new file mode 100644 index 0000000..49a0129 --- /dev/null +++ b/pkg/ip/cidr.go @@ -0,0 +1,48 @@ +package ip + +import ( + "net" + "strconv" + "strings" +) + +// CIDRKeys returns all possible CIDR prefixes of an IP, from the most specific (/32 for IPv4, /128 for IPv6) to the least specific (/0). +func CIDRKeys(ipStr string) []string { + ip := net.ParseIP(ipStr) + if ip == nil { + return nil + } + ip4 := ip.To4() + if ip4 != nil { + keys := make([]string, 0, 33) + for bits := 32; bits >= 0; bits-- { + mask := net.CIDRMask(bits, 32) + n := make(net.IP, 4) + for i := 0; i < 4; i++ { + n[i] = ip4[i] & mask[i] + } + keys = append(keys, n.String()+"/"+strconv.Itoa(bits)) + } + return keys + } + ip16 := ip.To16() + keys := make([]string, 0, 129) + for bits := 128; bits >= 0; bits-- { + mask := net.CIDRMask(bits, 128) + n := make(net.IP, 16) + for i := 0; i < 16; i++ { + n[i] = ip16[i] & mask[i] + } + keys = append(keys, n.String()+"/"+strconv.Itoa(bits)) + } + return keys +} + +// NormalizeCIDR parses a CIDR string and returns its normalized form, or an empty string if invalid. +func NormalizeCIDR(cidrStr string) string { + _, ipNet, err := net.ParseCIDR(strings.TrimSpace(cidrStr)) + if err != nil { + return "" + } + return ipNet.String() +} diff --git a/pkg/ip/cidr_test.go b/pkg/ip/cidr_test.go new file mode 100644 index 0000000..5ede450 --- /dev/null +++ b/pkg/ip/cidr_test.go @@ -0,0 +1,126 @@ +package ip + +import ( + "strconv" + "strings" + "testing" +) + +func TestCIDRKeys(t *testing.T) { + tests := []struct { + ip string + wantKeys int + checks map[int]string + }{ + { + ip: "10.0.0.1", + wantKeys: 33, + checks: map[int]string{0: "10.0.0.1/32", 8: "10.0.0.0/24", 32: "0.0.0.0/0"}, + }, + { + ip: "2001:db8::1", + wantKeys: 129, + checks: map[int]string{0: "2001:db8::1/128", 32: "2001:db8::/96", 128: "::/0"}, + }, + } + for _, tt := range tests { + t.Run(tt.ip, func(t *testing.T) { + keys := CIDRKeys(tt.ip) + if keys == nil { + t.Fatal("CIDRKeys returned nil") + } + if len(keys) != tt.wantKeys { + t.Fatalf("expected %d keys, got %d", tt.wantKeys, len(keys)) + } + for idx, want := range tt.checks { + if keys[idx] != want { + t.Errorf("keys[%d] should be %s, got %s", idx, want, keys[idx]) + } + } + }) + } +} + +func TestCIDRKeys_MostToLeastSpecific(t *testing.T) { + ips := []string{"10.0.0.1", "2001:db8::1"} + for _, ip := range ips { + t.Run(ip, func(t *testing.T) { + keys := CIDRKeys(ip) + for i := 1; i < len(keys); i++ { + prevBits := strings.Split(keys[i-1], "/")[1] + curBits := strings.Split(keys[i], "/")[1] + prevN, _ := strconv.Atoi(prevBits) + curN, _ := strconv.Atoi(curBits) + if prevN <= curN { + t.Errorf("keys should go from most specific to least specific at index %d: /%d <= /%d", i, prevN, curN) + } + } + }) + } +} + +func TestCIDRKeys_IPVariants(t *testing.T) { + tests := []struct { + ip string + wantKeys int + }{ + {"0.0.0.0", 33}, + {"255.255.255.255", 33}, + {"1.2.3.4", 33}, + {"10.0.0.1", 33}, + {"192.168.1.1", 33}, + {"invalid", 0}, + {"", 0}, + {" ", 0}, + } + for _, tt := range tests { + t.Run(tt.ip, func(t *testing.T) { + keys := CIDRKeys(tt.ip) + if len(keys) != tt.wantKeys { + t.Errorf("CIDRKeys(%q) returned %d keys, want %d", tt.ip, len(keys), tt.wantKeys) + } + }) + } +} + +func TestCIDRKeys_VerifyNetworkAddress(t *testing.T) { + keys := CIDRKeys("10.1.2.3") + tests := []struct { + bits int + want string + }{ + {24, "10.1.2.0/24"}, + {16, "10.1.0.0/16"}, + {8, "10.0.0.0/8"}, + } + for _, tt := range tests { + if got := keys[32-tt.bits]; got != tt.want { + t.Errorf("/%d network should be %s, got %s", tt.bits, tt.want, got) + } + } +} + +func TestNormalizeCIDR(t *testing.T) { + tests := []struct { + input string + want string + }{ + {"10.0.0.0/8", "10.0.0.0/8"}, + {"10.0.0.0/16", "10.0.0.0/16"}, + {"192.168.1.0/24", "192.168.1.0/24"}, + {"2001:db8::/32", "2001:db8::/32"}, + {"0.0.0.0/0", "0.0.0.0/0"}, + {"::/0", "::/0"}, + {"invalid", ""}, + {"", ""}, + {" 10.0.0.0/8 ", "10.0.0.0/8"}, + } + for _, tt := range tests { + t.Run(tt.input, func(t *testing.T) { + got := NormalizeCIDR(tt.input) + if got != tt.want { + t.Errorf("NormalizeCIDR(%q) = %q, want %q", tt.input, got, tt.want) + } + }) + } +} diff --git a/tests/e2e/mock/lib/common.sh b/tests/e2e/mock/lib/common.sh index 3dc5f1d..7699c6d 100644 --- a/tests/e2e/mock/lib/common.sh +++ b/tests/e2e/mock/lib/common.sh @@ -78,11 +78,11 @@ ensure_mock() { # Poll a URL until it returns the expected status code, or fail. # Usage: wait_for_status URL CODE [TIMEOUT_SECONDS] [curl args...] wait_for_status() { - local url="$1" expected="$2" timeout="${3:-30}" + local url="$1" expected="$2" timeout="${3:-15}" shift 3 || true local elapsed=0 got="" while (( elapsed < timeout )); do - got=$(curl -s -o /dev/null -w '%{http_code}' "$@" "$url" || true) + got=$(curl -s -m 1 -o /dev/null -w '%{http_code}' "$@" "$url" || true) if [[ "$got" == "$expected" ]]; then return 0 fi @@ -98,11 +98,11 @@ wait_for_status() { # code alone can't tell the states apart (e.g. captcha page vs backend, both 200). # Usage: wait_for_body_contains URL NEEDLE [TIMEOUT_SECONDS] [curl args...] wait_for_body_contains() { - local url="$1" needle="$2" timeout="${3:-30}" + local url="$1" needle="$2" timeout="${3:-15}" shift 3 || true local elapsed=0 body="" while (( elapsed < timeout )); do - body=$(curl -s "$@" "$url" || true) + body=$(curl -s -m 1 "$@" "$url" || true) if grep -q "$needle" <<<"$body"; then return 0 fi @@ -119,7 +119,7 @@ assert_status() { local url="$1" expected="$2" shift 2 || true local got - got=$(curl -s -o /dev/null -w '%{http_code}' "$@" "$url") + got=$(curl -s -m 1 -o /dev/null -w '%{http_code}' "$@" "$url") if [[ "$got" != "$expected" ]]; then echo "assert_status: $url expected $expected, got $got" >&2 return 1 @@ -132,7 +132,7 @@ assert_header() { local url="$1" header="$2" expected="$3" shift 3 || true local got - got=$(curl -s -D - -o /dev/null "$@" "$url" | tr -d '\r' \ + got=$(curl -s -m 1 -D - -o /dev/null "$@" "$url" | tr -d '\r' \ | awk -v h="${header,,}" -F': ' 'tolower($1) == h { print $2; exit }') if [[ "$got" != "$expected" ]]; then echo "assert_header: $url header $header expected \"$expected\", got \"$got\"" >&2 @@ -146,7 +146,7 @@ assert_body_contains() { local url="$1" needle="$2" shift 2 || true local body - body=$(curl -s "$@" "$url") + body=$(curl -s -m 1 "$@" "$url") if ! grep -q "$needle" <<<"$body"; then echo "assert_body_contains: $url expected to contain \"$needle\", got:" >&2 echo "$body" >&2 @@ -158,12 +158,20 @@ assert_body_contains() { lapi_add_decision() { local ip="$1" type="${2:-ban}" duration="${3:-4h}" - curl -sS -X POST "http://127.0.0.1:${LAPI_PORT}/admin/decisions?ip=${ip}&type=${type}&duration=${duration}" >/dev/null + curl -sS -m 1 -X POST "http://127.0.0.1:${LAPI_PORT}/admin/decisions?ip=${ip}&type=${type}&duration=${duration}" >/dev/null } lapi_delete_decision() { local ip="$1" - curl -sS -X DELETE "http://127.0.0.1:${LAPI_PORT}/admin/decisions?ip=${ip}" >/dev/null + curl -sS -m 1 -X DELETE "http://127.0.0.1:${LAPI_PORT}/admin/decisions?ip=${ip}" >/dev/null +} + +lapi_set_stream_fail() { + curl -sS -m 1 -X POST "http://127.0.0.1:${LAPI_PORT}/admin/stream-fail" >/dev/null +} + +lapi_clear_stream_fail() { + curl -sS -m 1 -X DELETE "http://127.0.0.1:${LAPI_PORT}/admin/stream-fail" >/dev/null } # --- stack lifecycle --------------------------------------------------------- diff --git a/tests/e2e/mock/mocklapi/main.go b/tests/e2e/mock/mocklapi/main.go index 65f5ad6..d0a3e9e 100644 --- a/tests/e2e/mock/mocklapi/main.go +++ b/tests/e2e/mock/mocklapi/main.go @@ -21,6 +21,7 @@ import ( "net/http" "strings" "sync" + "sync/atomic" ) // Decision is the subset of a LAPI decision the plugin actually reads. @@ -34,6 +35,9 @@ var ( mu sync.Mutex active = map[string]Decision{} // ip -> decision currently in force deleted = map[string]Decision{} // ip -> decision to report in the stream "deleted" list + // streamFail makes /v1/decisions/stream return 500 when set, to exercise the + // bouncer's fail-closed behaviour on consecutive stream poll failures. + streamFail atomic.Bool ) func writeJSON(w http.ResponseWriter, v any) { @@ -173,6 +177,10 @@ func main() { // "deleted". Re-sending the same on every poll is harmless — the plugin just // re-adds to / re-deletes from its cache. mux.HandleFunc("/v1/decisions/stream", func(w http.ResponseWriter, _ *http.Request) { + if streamFail.Load() { + w.WriteHeader(http.StatusInternalServerError) + return + } mu.Lock() defer mu.Unlock() writeJSON(w, map[string][]Decision{"new": list(active), "deleted": list(deleted)}) @@ -183,6 +191,17 @@ func main() { w.WriteHeader(http.StatusCreated) }) + // Test control plane: make the stream endpoint fail (POST) or recover (DELETE). + mux.HandleFunc("/admin/stream-fail", func(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodPost: + streamFail.Store(true) + case http.MethodDelete: + streamFail.Store(false) + } + w.WriteHeader(http.StatusOK) + }) + // Test control plane: add / remove decisions instead of cscli. mux.HandleFunc("/admin/decisions", func(_ http.ResponseWriter, r *http.Request) { q := r.URL.Query() diff --git a/tests/e2e/mock/scenarios/none-mode/run.sh b/tests/e2e/mock/scenarios/none-mode/run.sh index 2afc06a..dc24c72 100755 --- a/tests/e2e/mock/scenarios/none-mode/run.sh +++ b/tests/e2e/mock/scenarios/none-mode/run.sh @@ -11,17 +11,25 @@ body() { echo "[$SCENARIO] no decision -> request passes (LAPI queried per request)" assert_status "http://127.0.0.1:${WEB_PORT}/foo" 200 -H "X-Forwarded-For: 1.2.3.4" - echo "[$SCENARIO] adding ban decision for 1.2.3.4" + echo "[$SCENARIO] adding ban decision for 1.2.3.4 and 2001:db8::1" lapi_add_decision 1.2.3.4 ban 5m + lapi_add_decision "2001:db8::1" ban 5m - echo "[$SCENARIO] none mode has no cache -> next request must be blocked immediately" + echo "[$SCENARIO] IP banned must be blocked (HTTP 403)" assert_status "http://127.0.0.1:${WEB_PORT}/foo" 403 -H "X-Forwarded-For: 1.2.3.4" + echo "[$SCENARIO] IPv6 banned must be blocked (HTTP 403)" + assert_status "http://127.0.0.1:${WEB_PORT}/foo" 403 -H "X-Forwarded-For: 2001:db8::1" + echo "[$SCENARIO] deleting decision" lapi_delete_decision 1.2.3.4 + lapi_delete_decision "2001:db8::1" - echo "[$SCENARIO] previously banned IP must pass again immediately" + echo "[$SCENARIO] previously banned IP must pass again" assert_status "http://127.0.0.1:${WEB_PORT}/foo" 200 -H "X-Forwarded-For: 1.2.3.4" + + echo "[$SCENARIO] previously banned IPv6 must pass again" + wait_for_status "http://127.0.0.1:${WEB_PORT}/foo" 200 15 -H "X-Forwarded-For: 2001:db8::1" } run_scenario "$SCENARIO" "$HERE" body diff --git a/tests/e2e/mock/scenarios/stream-mode/dynamic.yml b/tests/e2e/mock/scenarios/stream-mode/dynamic.yml index 302ea1b..8c9a30a 100644 --- a/tests/e2e/mock/scenarios/stream-mode/dynamic.yml +++ b/tests/e2e/mock/scenarios/stream-mode/dynamic.yml @@ -18,7 +18,8 @@ http: bouncer: enabled: "true" crowdsecMode: stream - updateIntervalSeconds: "2" + updateIntervalSeconds: "1" + updateMaxFailure: "2" crowdsecLapiScheme: http crowdsecLapiHost: "@@LAPI_HOST@@" crowdsecLapiKey: "@@APIKEY@@" diff --git a/tests/e2e/mock/scenarios/stream-mode/run.sh b/tests/e2e/mock/scenarios/stream-mode/run.sh index c584502..a47d63e 100755 --- a/tests/e2e/mock/scenarios/stream-mode/run.sh +++ b/tests/e2e/mock/scenarios/stream-mode/run.sh @@ -11,8 +11,9 @@ body() { echo "[$SCENARIO] no decision yet -> request allowed" assert_status "http://127.0.0.1:${WEB_PORT}/foo" 200 -H "X-Forwarded-For: 1.2.3.4" - echo "[$SCENARIO] adding ban decision for 1.2.3.4" + echo "[$SCENARIO] adding ban decision for 1.2.3.4 and 10.0.0.0/8" lapi_add_decision 1.2.3.4 ban 5m + lapi_add_decision 10.0.0.0/24 ban 5m echo "[$SCENARIO] banned IP must be blocked once the next stream poll lands (HTTP 403)" wait_for_status "http://127.0.0.1:${WEB_PORT}/foo" 403 15 -H "X-Forwarded-For: 1.2.3.4" @@ -20,11 +21,30 @@ body() { echo "[$SCENARIO] non-banned IP must still pass (HTTP 200)" assert_status "http://127.0.0.1:${WEB_PORT}/foo" 200 -H "X-Forwarded-For: 5.6.7.8" - echo "[$SCENARIO] deleting ban decision" + echo "[$SCENARIO] banned IP in CIDR must be blocked once polled (HTTP 403)" + wait_for_status "http://127.0.0.1:${WEB_PORT}/foo" 403 15 -H "X-Forwarded-For: 10.0.0.1" + + echo "[$SCENARIO] deleting ban decision for 1.2.3.4 and 10.0.0.0/8" lapi_delete_decision 1.2.3.4 + lapi_delete_decision 10.0.0.0/24 echo "[$SCENARIO] previously banned IP must pass again once the deletion is polled" wait_for_status "http://127.0.0.1:${WEB_PORT}/foo" 200 15 -H "X-Forwarded-For: 1.2.3.4" + + echo "[$SCENARIO] previously CIDR-banned IP must pass again once deletion is polled" + wait_for_status "http://127.0.0.1:${WEB_PORT}/foo" 200 15 -H "X-Forwarded-For: 10.0.0.1" + + echo "[$SCENARIO] making the stream endpoint fail -> bouncer must pass for one more cycle (updateMaxFailure: 2)" + lapi_set_stream_fail + sleep 2 # update cache is every 1 seconds then waiting for minimum 1 cycle + wait_for_status "http://127.0.0.1:${WEB_PORT}/foo" 200 15 -H "X-Forwarded-For: 8.8.8.8" + + echo "[$SCENARIO] bouncer must block everything (isStreamHealthy: false)" + wait_for_status "http://127.0.0.1:${WEB_PORT}/foo" 403 15 -H "X-Forwarded-For: 8.8.8.8" + + echo "[$SCENARIO] restoring the stream endpoint -> bouncer must recover and pass again" + lapi_clear_stream_fail + wait_for_status "http://127.0.0.1:${WEB_PORT}/foo" 200 15 -H "X-Forwarded-For: 8.8.8.8" } run_scenario "$SCENARIO" "$HERE" body