🐛 Support range decision on stream mode

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,12 +679,20 @@ func handleStreamCache(bouncer *Bouncer) error {
 			default:
 				bouncer.log.Info("handleStreamCache:unknownType " + decision.Type)
 			}
+			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 {
+		if strings.Contains(decision.Value, "/") {
+			bouncer.cacheClient.DeleteCIDR(decision.Value)
+		} else {
 			bouncer.cacheClient.Delete(decision.Value)
 		}
+	}
 	bouncer.log.Debug("handleStreamCache:updated")
 	isCrowdsecStreamStartup = false
 	return nil

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))
+}

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()
+}

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)
+			}
+		})
+	}
+}

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 ---------------------------------------------------------

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()

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

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@@"

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