crowdsec-bouncer-traefik-plugin/tests/e2e/mock/mocklapi/main.go

// 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, and a hardcoded Redis
// stand-in for exercising the redis cache path.
//
// It is NOT a Crowdsec/AppSec conformance harness — the real WAF engine (OWASP
// CRS, virtual patching) is out of scope. The AppSec endpoint here emulates a
// single deterministic rule so the suite can exercise the plugin's AppSec
// wiring (header forwarding, allow/block handling) end to end. See the README.
package main

import (
	"bufio"
	"encoding/json"
	"flag"
	"io"
	"log"
	"net"
	"net/http"
	"strings"
	"sync"
	"sync/atomic"
)

// Decision is the subset of a LAPI decision the plugin actually reads.
type Decision struct {
	Value    string `json:"value"`
	Type     string `json:"type"`
	Duration string `json:"duration"`
}

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) {
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(v)
}

func list(m map[string]Decision) []Decision {
	out := make([]Decision, 0, len(m))
	for _, d := range m {
		out = append(out, d)
	}
	return out
}

// --- Redis mock (inline-command wire format, as spoken by simpleredis) ---

// serveRedis is a hardcoded stand-in. When verdicts is true it plays a replica
// that holds decisions: every line is scanned for known IPs, 1.2.3.4 → "f"
// (clean), 1.2.3.5 → "t" (banned); any other GET is a miss ($-1). When verdicts
// is false it plays the primary and answers every GET with a miss, so a
// scenario can prove reads are served from the replica and not the primary.
// SET, DEL, AUTH, SELECT get +OK (they don't read the response anyway).
func serveRedis(addr string, verdicts bool) {
	ln, err := net.Listen("tcp", addr)
	if err != nil {
		log.Fatal(err)
	}
	defer ln.Close()

	for {
		conn, err := ln.Accept()
		if err != nil {
			continue
		}
		go func(conn net.Conn) {
			defer conn.Close()
			rd := bufio.NewReader(conn)
			for {
				line, _, err := rd.ReadLine()
				if err != nil {
					return
				}
				s := string(line)
				switch {
				case verdicts && strings.Contains(s, "1.2.3.4"):
					conn.Write([]byte("$1\r\nf\r\n"))
				case verdicts && strings.Contains(s, "1.2.3.5"):
					conn.Write([]byte("$1\r\nt\r\n"))
				case strings.HasPrefix(strings.ToUpper(s), "GET "):
					conn.Write([]byte("$-1\r\n"))
				default:
					conn.Write([]byte("+OK\r\n"))
				}
			}
		}(conn)
	}
}

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")
	// AppSec WAF stand-in (the real engine listens on :7422). Not a CRS engine.
	appsecAddr := flag.String("appsec-addr", "127.0.0.1:8092", "address for the AppSec mock")
	// Redis stand-ins on plain TCP ports, enough to exercise the plugin's redis
	// cache path. The primary answers every GET with a miss; the replica serves
	// the hardcoded verdicts, so a scenario pointing redisCacheReadHosts at the
	// replica proves reads are offloaded to replicas.
	redisAddr := flag.String("redis-addr", "127.0.0.1:8093", "address for the Redis primary mock (writes; GET always misses)")
	redisReadAddr := flag.String("redis-read-addr", "127.0.0.1:8094", "address for the Redis replica mock (serves cached verdicts)")
	// Optional TLS for the LAPI: when both are set the LAPI is served over HTTPS
	// (cert signed by the scenario's throwaway CA) so the suite can exercise the
	// bouncer's system-trust-store path. Backend and AppSec stay plaintext.
	lapiTLSCert := flag.String("lapi-tls-cert", "", "PEM cert to serve the LAPI over HTTPS (optional)")
	lapiTLSKey := flag.String("lapi-tls-key", "", "PEM key for --lapi-tls-cert")
	flag.Parse()

	go func() {
		log.Fatal(http.ListenAndServe(*backendAddr, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
			_, _ = w.Write([]byte("E2E_BACKEND_OK\n"))
		})))
	}()

	// AppSec mock: the plugin forwards the request metadata in X-Crowdsec-Appsec-*
	// headers and reads our status — 200 allows, 403 blocks. We emulate one
	// deterministic virtual-patching rule (block any URI containing "rpc2", the
	// exact probe from examples/appsec-enabled) so the plugin's AppSec path is
	// exercised without standing up the real WAF.
	go func() {
		log.Fatal(http.ListenAndServe(*appsecAddr, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			if strings.Contains(r.Header.Get("X-Crowdsec-Appsec-Uri"), "403") {
				w.WriteHeader(http.StatusForbidden)
			}
			if strings.Contains(r.Header.Get("X-Crowdsec-Appsec-Uri"), "500") {
				w.WriteHeader(http.StatusInternalServerError)
			}
			if strings.Contains(r.Header.Get("X-Crowdsec-Appsec-Uri"), "502") {
				w.WriteHeader(http.StatusBadGateway)
			}
			// Read body
			body, err := io.ReadAll(r.Body)
			if err != nil {
				w.WriteHeader(http.StatusInternalServerError)
				return
			}
			defer r.Body.Close()
			if strings.Contains(string(body), "a=0") {
				w.WriteHeader(http.StatusForbidden)
				return
			}
		})))
	}()

	go serveRedis(*redisAddr, false)
	go serveRedis(*redisReadAddr, true)

	mux := http.NewServeMux()

	// 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) {
		mu.Lock()
		defer mu.Unlock()
		if d, ok := active[r.URL.Query().Get("ip")]; ok {
			writeJSON(w, []Decision{d})
			return
		}
		_, _ = w.Write([]byte("null"))
	})

	// 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) {
		if streamFail.Load() {
			w.WriteHeader(http.StatusInternalServerError)
			return
		}
		mu.Lock()
		defer mu.Unlock()
		writeJSON(w, map[string][]Decision{"new": list(active), "deleted": list(deleted)})
	})

	// usage-metrics push: accept and ignore.
	mux.HandleFunc("/v1/usage-metrics", func(w http.ResponseWriter, _ *http.Request) {
		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()
		ip := q.Get("ip")
		mu.Lock()
		defer mu.Unlock()
		switch r.Method {
		case http.MethodPost:
			dtype := q.Get("type")
			if dtype == "" {
				dtype = "ban"
			}
			duration := q.Get("duration")
			if duration == "" {
				duration = "4h"
			}
			active[ip] = Decision{Value: ip, Type: dtype, Duration: duration}
			delete(deleted, ip)
		case http.MethodDelete:
			if d, ok := active[ip]; ok {
				deleted[ip] = d
				delete(active, ip)
			}
		}
	})

	if *lapiTLSCert != "" && *lapiTLSKey != "" {
		log.Printf("mocklapi: LAPI on %s (TLS), backend on %s, appsec on %s, redis on %s (read %s)", *lapiAddr, *backendAddr, *appsecAddr, *redisAddr, *redisReadAddr)
		log.Fatal(http.ListenAndServeTLS(*lapiAddr, *lapiTLSCert, *lapiTLSKey, mux))
	}
	log.Printf("mocklapi: LAPI on %s, backend on %s, appsec on %s, redis on %s (read %s)", *lapiAddr, *backendAddr, *appsecAddr, *redisAddr, *redisReadAddr)
	log.Fatal(http.ListenAndServe(*lapiAddr, mux))
}