pretecno/crowdsec-bouncer-traefik-plugin
3
0
Fork 0
mirror of https://github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin.git synced 2026-07-20 11:09:00 +02:00
Code Issues Packages Releases Activity

🔥 e2e mock: simplify the Crowdsec LAPI mock

Browse Source 
Per review, trim the mock to the minimum the plugin actually exercises:

- Drop the stream delta bookkeeping (startup flag + "already streamed" set).
  The plugin re-Sets/Deletes its cache on every poll, so reporting the whole
  active set as "new" and removed ones as "deleted" is enough.
- Shrink the Decision struct to the three fields the plugin reads
  (value/type/duration); drop id/origin/scope/scenario and the id counter.
- Drop API-key auth and the /admin/reset endpoint — no scenario exercises
  either. Also drop the now-unused lapi_reset helper.
- Replace the store struct + methods with two package-level maps + a mutex.

mocklapi/main.go: 234 -> 118 lines. All six scenarios still pass.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
mhx
2026-06-06 12:03:19 +02:00
co-authored by Claude Opus 4.8
parent fcebbfe902
commit b67edfe308
3 changed files with 73 additions and 194 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/e2e/mock/README.md
+2 -2
View File
@@ -79,6 +79,6 @@ mock/
template).
2. In `run.sh`, define a `body` function with the assertions and call
`run_scenario "<name>" "$HERE" body`.
3. Drive decisions with `lapi_add_decision <ip> [type] [duration]`,
`lapi_delete_decision <ip>`, `lapi_reset`.
3. Drive decisions with `lapi_add_decision <ip> [type] [duration]` and
`lapi_delete_decision <ip>`.
4. Add `<name>` to `E2E_MOCK_SCENARIOS` in the `Makefile`.
tests/e2e/mock/lib/common.sh
+1 -6
View File
@@ -143,10 +143,6 @@ lapi_delete_decision() {
curl -sS -X DELETE "http://127.0.0.1:${LAPI_PORT}/admin/decisions?ip=${ip}" >/dev/null
}
lapi_reset() {
curl -sS -X POST "http://127.0.0.1:${LAPI_PORT}/admin/reset" >/dev/null
}
# --- stack lifecycle ---------------------------------------------------------
# start_stack SCENARIO_DIR
@@ -174,8 +170,7 @@ start_stack() {
"$mock_bin" \
--lapi-addr "127.0.0.1:${LAPI_PORT}" \
--backend-addr "127.0.0.1:${BACKEND_PORT}" \
--api-key "${LAPI_KEY}" >"$WORKDIR/mock.log" 2>&1 &
--backend-addr "127.0.0.1:${BACKEND_PORT}" >"$WORKDIR/mock.log" 2>&1 &
MOCK_PID=$!
( cd "$WORKDIR" && exec "$traefik_bin" --configfile=traefik.yml ) >"$WORKDIR/traefik.log" 2>&1 &
tests/e2e/mock/mocklapi/main.go
+70 -186
View File
@@ -1,179 +1,100 @@
// Command mocklapi is a minimal Crowdsec LAPI mock for the binary e2e suite.
// Command mocklapi is a minimal Crowdsec LAPI stand-in for the binary e2e
// suite. It answers only the few LAPI routes the plugin calls — live/none
// decision lookups, the stream poll and the usage-metrics push — and lets the
// test drive decisions through /admin instead of `cscli`. It also serves the
// stub upstream that Traefik proxies allowed requests to.
//
// It speaks just enough of the LAPI HTTP contract for the plugin to exercise
// its own logic — live/none queries, the stream delta protocol and the
// usage-metrics push — without running a real Crowdsec. Decisions are driven
// from the test via the /admin endpoints instead of `cscli`.
//
// This is deliberately NOT a Crowdsec/AppSec conformance harness: validating
// that Crowdsec or its AppSec engine behave correctly is the upstream
// maintainer's responsibility, not this plugin's. See the suite README.
// It is deliberately NOT a Crowdsec/AppSec conformance harness: Crowdsec's own
// correctness is the upstream maintainer's responsibility, not this plugin's.
// See the suite README.
package main
import (
"encoding/json"
"flag"
"fmt"
"log"
"net/http"
"sync"
)
const lapiKeyHeader = "X-Api-Key"
// Decision mirrors the subset of the LAPI decision object the plugin reads.
// Decision is the subset of a LAPI decision the plugin actually reads.
type Decision struct {
ID int `json:"id"`
Origin string `json:"origin"`
Type string `json:"type"`
Scope string `json:"scope"`
Value string `json:"value"`
Type string `json:"type"`
Duration string `json:"duration"`
Scenario string `json:"scenario"`
}
// store holds the active decisions and the stream delta bookkeeping.
type store struct {
mu sync.Mutex
decisions map[string]Decision
streamed map[string]struct{} // values already advertised to the stream consumer
nextID int
}
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
)
func newStore() *store {
return &store{
decisions: map[string]Decision{},
streamed: map[string]struct{}{},
nextID: 1,
}
}
func (s *store) add(ip, dtype, duration string) {
s.mu.Lock()
defer s.mu.Unlock()
s.decisions[ip] = Decision{
ID: s.nextID,
Origin: "cscli",
Type: dtype,
Scope: "Ip",
Value: ip,
Duration: duration,
Scenario: "e2e",
}
s.nextID++
}
func (s *store) delete(ip string) {
s.mu.Lock()
defer s.mu.Unlock()
delete(s.decisions, ip)
}
func (s *store) reset() {
s.mu.Lock()
defer s.mu.Unlock()
s.decisions = map[string]Decision{}
s.streamed = map[string]struct{}{}
}
func (s *store) get(ip string) (Decision, bool) {
s.mu.Lock()
defer s.mu.Unlock()
d, ok := s.decisions[ip]
return d, ok
}
// stream computes the new/deleted delta since the last poll. On startup the
// consumer expects the full current set as "new".
func (s *store) stream(startup bool) map[string][]Decision {
s.mu.Lock()
defer s.mu.Unlock()
newd := []Decision{}
deleted := []Decision{}
for ip, d := range s.decisions {
if startup {
newd = append(newd, d)
continue
}
if _, seen := s.streamed[ip]; !seen {
newd = append(newd, d)
}
}
if !startup {
for ip := range s.streamed {
if _, active := s.decisions[ip]; !active {
deleted = append(deleted, Decision{Value: ip, Scope: "Ip", Type: "ban"})
}
}
}
s.streamed = map[string]struct{}{}
for ip := range s.decisions {
s.streamed[ip] = struct{}{}
}
return map[string][]Decision{"new": newd, "deleted": deleted}
}
func writeJSON(w http.ResponseWriter, code int, payload any) {
body, _ := json.Marshal(payload)
func writeJSON(w http.ResponseWriter, v any) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(code)
_, _ = w.Write(body)
_ = json.NewEncoder(w).Encode(v)
}
func lapiHandler(s *store, apiKey string) http.Handler {
func list(m map[string]Decision) []Decision {
out := make([]Decision, 0, len(m))
for _, d := range m {
out = append(out, d)
}
return out
}
func main() {
lapiAddr := flag.String("lapi-addr", "127.0.0.1:8090", "address for the LAPI mock")
// The stub upstream Traefik proxies allowed requests to — the binary-suite
// equivalent of the traefik/whoami container. Not AppSec.
backendAddr := flag.String("backend-addr", "127.0.0.1:8091", "address for the stub upstream service")
flag.Parse()
go func() {
log.Fatal(http.ListenAndServe(*backendAddr, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
_, _ = w.Write([]byte("E2E_BACKEND_OK\n"))
})))
}()
mux := http.NewServeMux()
authorized := func(r *http.Request) bool {
return r.Header.Get(lapiKeyHeader) == apiKey
}
mux.HandleFunc("/health", func(w http.ResponseWriter, _ *http.Request) {
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
})
mux.HandleFunc("/v1/decisions/stream", func(w http.ResponseWriter, r *http.Request) {
if !authorized(r) {
writeJSON(w, http.StatusForbidden, map[string]string{"message": "access forbidden"})
return
}
startup := r.URL.Query().Get("startup") == "true"
writeJSON(w, http.StatusOK, s.stream(startup))
})
// Readiness probe for the test harness (empty body, 200).
mux.HandleFunc("/health", func(http.ResponseWriter, *http.Request) {})
// live / none mode: the plugin asks about one IP and expects a decision
// array, or the literal `null` when there is none.
mux.HandleFunc("/v1/decisions", func(w http.ResponseWriter, r *http.Request) {
if !authorized(r) {
writeJSON(w, http.StatusForbidden, map[string]string{"message": "access forbidden"})
mu.Lock()
defer mu.Unlock()
if d, ok := active[r.URL.Query().Get("ip")]; ok {
writeJSON(w, []Decision{d})
return
}
ip := r.URL.Query().Get("ip")
if d, ok := s.get(ip); ok {
writeJSON(w, http.StatusOK, []Decision{d})
return
}
// LAPI returns the literal `null` when no decision matches.
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte("null"))
})
mux.HandleFunc("/v1/usage-metrics", func(w http.ResponseWriter, r *http.Request) {
// Accept and ignore — we only assert the plugin can push without erroring.
if !authorized(r) {
writeJSON(w, http.StatusForbidden, map[string]string{"message": "access forbidden"})
return
}
writeJSON(w, http.StatusCreated, map[string]string{"message": "ok"})
// stream mode: report the whole active set as "new" and anything removed as
// "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) {
mu.Lock()
defer mu.Unlock()
writeJSON(w, map[string][]Decision{"new": list(active), "deleted": list(deleted)})
})
mux.HandleFunc("/admin/decisions", func(w http.ResponseWriter, r *http.Request) {
// usage-metrics push: accept and ignore.
mux.HandleFunc("/v1/usage-metrics", func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusCreated)
})
// Test control plane: add / remove decisions instead of cscli.
mux.HandleFunc("/admin/decisions", func(_ http.ResponseWriter, r *http.Request) {
q := r.URL.Query()
ip := q.Get("ip")
mu.Lock()
defer mu.Unlock()
switch r.Method {
case http.MethodPost:
if ip == "" {
writeJSON(w, http.StatusBadRequest, map[string]string{"message": "missing ip"})
return
}
dtype := q.Get("type")
if dtype == "" {
dtype = "ban"
@@ -182,53 +103,16 @@ func lapiHandler(s *store, apiKey string) http.Handler {
if duration == "" {
duration = "4h"
}
s.add(ip, dtype, duration)
writeJSON(w, http.StatusOK, map[string]string{"message": "added", "ip": ip, "type": dtype})
active[ip] = Decision{Value: ip, Type: dtype, Duration: duration}
delete(deleted, ip)
case http.MethodDelete:
s.delete(ip)
writeJSON(w, http.StatusOK, map[string]string{"message": "deleted", "ip": ip})
default:
writeJSON(w, http.StatusMethodNotAllowed, map[string]string{"message": "method not allowed"})
if d, ok := active[ip]; ok {
deleted[ip] = d
delete(active, ip)
}
}
})
mux.HandleFunc("/admin/reset", func(w http.ResponseWriter, _ *http.Request) {
s.reset()
writeJSON(w, http.StatusOK, map[string]string{"message": "reset"})
})
return mux
}
func backendHandler() http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/plain")
w.Header().Set("X-E2E-Backend", "ok")
w.WriteHeader(http.StatusOK)
if r.Method != http.MethodHead {
_, _ = w.Write([]byte("E2E_BACKEND_OK\n"))
}
})
}
func main() {
lapiAddr := flag.String("lapi-addr", "127.0.0.1:8090", "address for the LAPI mock")
// The stub upstream service Traefik proxies allowed requests to — the
// binary-suite equivalent of the traefik/whoami container. Not AppSec.
backendAddr := flag.String("backend-addr", "127.0.0.1:8091", "address of the stub upstream service Traefik proxies allowed requests to")
apiKey := flag.String("api-key", "e2e-mock-key", "expected X-Api-Key value")
flag.Parse()
s := newStore()
go func() {
if err := http.ListenAndServe(*backendAddr, backendHandler()); err != nil {
log.Fatalf("backend: %v", err)
}
}()
fmt.Printf("mocklapi: LAPI on %s, backend on %s\n", *lapiAddr, *backendAddr)
if err := http.ListenAndServe(*lapiAddr, lapiHandler(s, *apiKey)); err != nil {
log.Fatalf("lapi: %v", err)
}
log.Printf("mocklapi: LAPI on %s, backend on %s", *lapiAddr, *backendAddr)
log.Fatal(http.ListenAndServe(*lapiAddr, mux))
}