Clean docker-compose, update function name

✨ fix loop

.gitignore

@@ -1,2 +1,6 @@
 .idea/
 .DS_Store
+config
+db
+logs
+docker-compose.dev.yml

Makefile

@@ -18,3 +18,25 @@ vendor:
 
 clean:
 	rm -rf ./vendor
+
+run_dev:
+	docker-compose -f docker-compose.dev.yml up -d --remove-orphans
+
+run_local:
+	docker-compose -f docker-compose.local.yml up -d --remove-orphans
+
+run:
+	docker-compose -f docker-compose.yml up -d --remove-orphans
+
+restart_docker_dev:
+	docker-compose -f docker-compose.dev.yml restart
+
+restart_docker_local:
+	docker-compose -f docker-compose.local.yml restart
+
+restart_docker:
+	docker-compose -f docker-compose.yml restart
+
+show_metrics:
+	docker exec crowdsec cscli metrics
+

README.md

@@ -17,26 +17,35 @@ The crowdsec utility will provide the community blocklist which contains highly
 When used with crowdsec it will leverage the local API which will analyze traefik logs and take decisions on the requests made by users/bots. Malicious actors will be banned based on patterns against your website.
 
 There are 4 operating modes (CrowdsecMode) for this plugin:
-- none -> If the client IP is on ban list, it will get a http code 403 response.
-    Otherwise, request will continue as usual. All request call the Crowdsec LAPI
 
-- live -> If the client IP is on ban list, it will get a http code 403 response.
-    Otherwise, request will continue as usual.
-    The bouncer can leverage use of a local cache in order to reduce the number
-    of requests made to the Crowdsec LAPI. It will keep in cache the status for
-    each IP that makes queries.
-
-- stream -> Stream Streaming mode allows you to keep in the local cache only the Banned IPs,
-    every requests that does not hit the cache is authorized.
-    Every minute, the cache is updated with news from the Crowdsec LAPI.
-
-- alone -> Streaming mode but the blacklisted IPs are fetched on the CAPI.
-    Every 2 hours, the cache is updated with news from the Crowdsec CAPI.
+| Mode | Description |
+|------|------|
+| none | If the client IP is on ban list, it will get a http code 403 response. Otherwise, request will continue as usual. All request call the Crowdsec LAPI |
+| live | If the client IP is on ban list, it will get a http code 403 response. Otherwise, request will continue as usual.    The bouncer can leverage use of a local cache in order to reduce the number of requests made to the Crowdsec LAPI. It will keep in cache the status for  each IP that makes queries. |
+| stream | Stream Streaming mode allows you to keep in the local cache only the Banned IPs, every requests that does not hit the cache is authorized. Every minute, the cache is updated with news from the Crowdsec LAPI. |
+| alone | Standalone mode, similar to the streaming mode but the blacklisted IPs are fetched on the CAPI. Every 2 hours, the cache is updated with news from the Crowdsec CAPI. It does not include any localy banned IP, but can work without a crowdsec service|
 
 The recommanded mode for performance is the streaming mode, decisions are updated every 60 sec by default and that's the only communication between traefik and crowdsec. Every requests that happens hits the cache for quick decisions.
 
 ## Usage
 
+To get started, use the `docker-compose.yml` file.
+
+You can run it with:
+```bash
+make run
+```
+
+### Note
+Each middleware in traefik has it's own data and is instanciated by service.
+This means if there are 10 services protected by the bouncer in streaming alone or live mode, the cache will be duplicated to all 10 services.
+This is because traefik does not allow plugins to store data locally that can be consummed.
+
+The synchronisation with the crowdsec service will happen also 10 times in the period selected.
+It should be taken into account when fixing this period so each middleware has time to sync data from crowdsec.
+
+At each start of synchronisation, the middleware will wait a random number of seconds to avoid simultaneous calls to crowdsec.
+
 ### Variables
 - Enabled
   - bool
@@ -53,7 +62,7 @@ The recommanded mode for performance is the streaming mode, decisions are update
   - Crowdsec LAPI available on which host.
 - CrowdsecLapiKey
   - string
-  - Crowdsec LAPI generated key for the bouncer. 
+  - Crowdsec LAPI generated key for the bouncer : **must be unique by service**. 
 - CrowdsecCapiLogin
   - string
   - Used only in `alone` mode, login for Crowdsec CAPI
@@ -66,11 +75,11 @@ The recommanded mode for performance is the streaming mode, decisions are update
 - UpdateIntervalSeconds
   - int64
   - default: 60
-  - User only in `stream` mode, interval between fetching blacklisted IPs from LAPI
+  - Used only in `stream` mode, interval between fetching blacklisted IPs from LAPI
 - DefaultDecisionSeconds
   - int64
   - default: 60
-  - User only in `live` mode, decision duration of accepted IPs
+  - Used only in `live` mode, decision duration of accepted IPs
 
 ### Configuration
 

acquis.yaml

@@ -1,4 +1,4 @@
 filenames:
- - /var/log/traefik/*.log
+ - /var/log/traefik/access.log
 labels:
   type: traefik

bouncer.go

@@ -8,6 +8,7 @@ import (
 	"io"
 	"io/ioutil"
 	"log"
+	"math/rand"
 	"net"
 	"net/http"
 	"net/url"
@@ -116,14 +117,18 @@ func New(ctx context.Context, next http.Handler, config *Config, name string) (h
 		},
 		cache: ttl_map.New(),
 	}
-	// if we are on a stream mode, we fetch in a go routine every minute the new decisions.
-	if config.CrowdsecMode == streamMode {
-		go handleStreamCache(bouncer, true)
-	} else if config.CrowdsecMode == aloneMode {
-		getToken(bouncer)
-		time.AfterFunc(10*time.Second, func() {
-			handleStreamCache(bouncer, false)
-		})
+	if config.CrowdsecMode == streamMode || config.CrowdsecMode == aloneMode {
+		rand.Seed(time.Now().UnixNano())
+		if config.CrowdsecMode == aloneMode {
+			getToken(bouncer)
+		}
+		ticker := time.NewTicker(time.Duration(config.UpdateIntervalSeconds) * time.Second)
+		go func() {
+			go handleStreamCache(bouncer)
+			for range ticker.C {
+				go handleStreamCache(bouncer)
+			}
+		}()
 	}
 	return bouncer, nil
 }
@@ -138,13 +143,13 @@ func (a *Bouncer) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
 	// TODO Make sur remote address does not include the port.
 	remoteHost, _, err := net.SplitHostPort(req.RemoteAddr)
 	if err != nil {
-		log.Printf("failed to extract ip from remote address: %v", err)
+		logger(fmt.Sprintf("failed to extract ip from remote address: %v", err))
 		a.next.ServeHTTP(rw, req)
 		return
 	}
 
-	if a.crowdsecMode == streamMode || a.crowdsecMode == liveMode {
-		isBanned, err := getDecision(a.cache, remoteHost)
+	if a.crowdsecMode == streamMode || a.crowdsecMode == aloneMode || a.crowdsecMode == liveMode {
+		isBanned, err := getDecision(a, remoteHost)
 		if err == nil {
 			if isBanned {
 				rw.WriteHeader(http.StatusForbidden)
@@ -156,7 +161,7 @@ func (a *Bouncer) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
 	}
 
 	// Right here if we cannot join the stream we forbid the request to go on.
-	if a.crowdsecMode == streamMode {
+	if a.crowdsecMode == streamMode || a.crowdsecMode == aloneMode {
 		if a.crowdsecStreamHealthy {
 			a.next.ServeHTTP(rw, req)
 		} else {
@@ -195,6 +200,10 @@ type Login struct {
 	Expire string `json:"expire"`
 }
 
+func logger(str string) {
+	log.Printf("Crowdsec Bouncer Traefik Plugin - %s", str)
+}
+
 func contains(source []string, target string) bool {
 	for _, a := range source {
 		if a == target {
@@ -206,8 +215,8 @@ func contains(source []string, target string) bool {
 
 // Get Decision check in the cache if the IP has the banned / not banned value.
 // Otherwise return with an error to add the IP in cache if we are on.
-func getDecision(cache *ttl_map.Heap, clientIP string) (bool, error) {
-	banned, isCached := cache.Get(clientIP)
+func getDecision(a *Bouncer, clientIP string) (bool, error) {
+	banned, isCached := a.cache.Get(clientIP)
 	bannedString, isValid := banned.(string)
 	if isCached && isValid && len(bannedString) > 0 {
 		return bannedString == cacheBannedValue, nil
@@ -215,12 +224,15 @@ func getDecision(cache *ttl_map.Heap, clientIP string) (bool, error) {
 	return false, fmt.Errorf("no cache data")
 }
 
-func setDecision(cache *ttl_map.Heap, clientIP string, isBanned bool, duration int64) {
+func setDecision(a *Bouncer, clientIP string, isBanned bool, duration int64) {
+	if a.crowdsecMode == noneMode {
+		return
+	}
 	if isBanned {
-		log.Printf("%v banned", clientIP)
-		cache.Set(clientIP, cacheBannedValue, duration)
+		logger(fmt.Sprintf("%v banned", clientIP))
+		a.cache.Set(clientIP, cacheBannedValue, duration)
 	} else {
-		cache.Set(clientIP, cacheNoBannedValue, duration)
+		a.cache.Set(clientIP, cacheNoBannedValue, duration)
 	}
 }
 
@@ -235,9 +247,7 @@ func handleNoStreamCache(a *Bouncer, rw http.ResponseWriter, req *http.Request,
 	body := crowdsecQuery(a, routeURL.String(), false)
 
 	if bytes.Equal(body, []byte("null")) {
-		if a.crowdsecMode == liveMode {
-			setDecision(a.cache, remoteHost, false, a.defaultDecisionTimeout)
-		}
+		setDecision(a, remoteHost, false, a.defaultDecisionTimeout)
 		a.next.ServeHTTP(rw, req)
 		return
 	}
@@ -245,38 +255,45 @@ func handleNoStreamCache(a *Bouncer, rw http.ResponseWriter, req *http.Request,
 	var decisions []Decision
 	err := json.Unmarshal(body, &decisions)
 	if err != nil {
-		log.Printf("failed to parse body: %s", err)
+		logger(fmt.Sprintf("failed to parse body: %s", err))
 		rw.WriteHeader(http.StatusForbidden)
 		return
 	}
 	if len(decisions) == 0 {
-		if a.crowdsecMode == liveMode {
-			setDecision(a.cache, remoteHost, false, a.defaultDecisionTimeout)
-		}
+		setDecision(a, remoteHost, false, a.defaultDecisionTimeout)
 		a.next.ServeHTTP(rw, req)
 		return
 	}
 	rw.WriteHeader(http.StatusForbidden)
 	duration, err := time.ParseDuration(decisions[0].Duration)
 	if err != nil {
-		log.Printf("failed to parse duration: %s", err)
+		logger(fmt.Sprintf("failed to parse duration: %s", err))
 		return
 	}
-	setDecision(a.cache, remoteHost, true, int64(duration.Seconds()))
+	setDecision(a, remoteHost, true, int64(duration.Seconds()))
 }
 
-func handleStreamCache(a *Bouncer, initialized bool) {
+func waitStream() {
+	logger("Start Wait Stream")
+	sec := rand.Int63n(30)
+	logger(fmt.Sprintf("Sec=%v", sec))
+	time.Sleep(time.Duration(sec) * time.Second)
+	logger("End Wait Stream")
+}
+
+func handleStreamCache(a *Bouncer) {
+	logger(fmt.Sprintf("Start handleStreamCache with health=%v", a.crowdsecStreamHealthy))
 	// TODO clean properly on exit.
-	time.AfterFunc(time.Duration(a.updateInterval)*time.Second, func() {
-		handleStreamCache(a, false)
-	})
+
+	waitStream()
+	// wait random number of sec
 	var rawQuery string
 	var path string
 	if a.crowdsecMode == aloneMode {
 		rawQuery = ""
 		path = crowdsecCapiDecisions
 	} else {
-		rawQuery = fmt.Sprintf("startup=%t", initialized)
+		rawQuery = fmt.Sprintf("startup=%t", !a.crowdsecStreamHealthy)
 		path = crowdsecLapiStreamRoute
 	}
 	streamRouteURL := url.URL{
@@ -289,20 +306,21 @@ func handleStreamCache(a *Bouncer, initialized bool) {
 	var stream Stream
 	err := json.Unmarshal(body, &stream)
 	if err != nil {
-		log.Printf("error while parsing body: %s", err)
+		logger(fmt.Sprintf("error while parsing body: %s", err))
 		a.crowdsecStreamHealthy = false
 		return
 	}
 	for _, decision := range stream.New {
 		duration, err := time.ParseDuration(decision.Duration)
 		if err == nil {
-			setDecision(a.cache, decision.Value, true, int64(duration.Seconds()))
+			setDecision(a, decision.Value, true, int64(duration.Seconds()))
 		}
 	}
 	for _, decision := range stream.Deleted {
 		a.cache.Del(decision.Value)
 	}
 	a.crowdsecStreamHealthy = true
+	logger(fmt.Sprintf("End handleStreamCache with health=%v", a.crowdsecStreamHealthy))
 }
 
 func getToken(a *Bouncer) {
@@ -315,7 +333,7 @@ func getToken(a *Bouncer) {
 	var login Login
 	err := json.Unmarshal(body, &login)
 	if err != nil {
-		log.Printf("error while parsing body: %s", err)
+		logger(fmt.Sprintf("error while parsing body: %s", err))
 		a.crowdsecStreamHealthy = false
 		return
 	}
@@ -344,7 +362,7 @@ func crowdsecQuery(a *Bouncer, stringURL string, isPost bool) []byte {
 	}
 	res, err := a.client.Do(req)
 	if err != nil {
-		log.Printf("error while fetching %v: %s", stringURL, err)
+		logger(fmt.Sprintf("error while fetching %v: %s", stringURL, err))
 		a.crowdsecStreamHealthy = false
 		return nil
 	}
@@ -358,19 +376,19 @@ func crowdsecQuery(a *Bouncer, stringURL string, isPost bool) []byte {
 		return crowdsecQuery(a, stringURL, false)
 	}
 	if res.StatusCode != http.StatusOK {
-		log.Printf("error while fetching %v, status code: %d", stringURL, res.StatusCode)
+		logger(fmt.Sprintf("error while fetching %v, status code: %d", stringURL, res.StatusCode))
 		a.crowdsecStreamHealthy = false
 		return nil
 	}
 	defer func(body io.ReadCloser) {
 		err = body.Close()
 		if err != nil {
-			log.Printf("failed to close body reader: %s", err)
+			logger(fmt.Sprintf("failed to close body reader: %s", err))
 		}
 	}(res.Body)
 	body, err := ioutil.ReadAll(res.Body)
 	if err != nil {
-		log.Printf("error while reading body: %s", err)
+		logger(fmt.Sprintf("error while reading body: %s", err))
 		a.crowdsecStreamHealthy = false
 		return nil
 	}

docker-compose.local.yml

@@ -2,12 +2,12 @@ version: "3.8"
 
 services:
   traefik:
-    image: "traefik:v2.8.7"
+    image: "traefik:v2.8.8"
     container_name: "traefik"
     command:
       # - "--log.level=DEBUG"
       - "--accesslog"
-      - "--accesslog.filepath=/var/log/traefik/traefik.log"
+      - "--accesslog.filepath=/var/log/traefik/access.log"
       - "--api.insecure=true"
       - "--providers.docker=true"
       - "--providers.docker.exposedbydefault=false"
@@ -15,44 +15,69 @@ services:
 
       - "--experimental.localplugins.bouncer.modulename=github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin"
     volumes:
-      - "/var/run/docker.sock:/var/run/docker.sock:ro"
-      - "logs:/var/log/traefik"
+      - /var/run/docker.sock:/var/run/docker.sock:ro
+      - logs-local:/var/log/traefik
       - ./:/plugins-local/src/github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin
-    ports:
-      - 8000:80
-      - 8080:8080
-    depends_on:
-      - 'crowdsec'
-
-  whoami:
-    image: traefik/whoami
-    container_name: "simple-service"
     labels:
       - "traefik.enable=true"
-      - "traefik.http.routers.whoami.rule=Host(`localhost`)"
-      - "traefik.http.routers.whoami.entrypoints=web"
-      - "traefik.http.routers.whoami.middlewares=crowdsec@docker"
       - "traefik.http.middlewares.crowdsec.plugin.bouncer.enabled=true"
       - "traefik.http.middlewares.crowdsec.plugin.bouncer.crowdseclapikey=40796d93c2958f9e58345514e67740e5"
       - "traefik.http.middlewares.crowdsec.plugin.bouncer.crowdseccapilogin=3829a6c9870e4726a377d8951ebb64a1m8psGvMaq1ykJ3zX"
       - "traefik.http.middlewares.crowdsec.plugin.bouncer.crowdseccapipwd=Q5pgN8bRNInHGdx6QCksdPOJVLeLQ7ipJntSeuP3r8088zXzRVs4G8liXAKfI1k6"
       - "traefik.http.middlewares.crowdsec.plugin.bouncer.crowdseccapiscenarios=crowdsecurity/http-backdoors-attempts,baudneo/zoneminder-bf"
+    ports:
+      - 80:80
+      - 8080:8080
+    depends_on:
+      - crowdsec
+
+  whoami1:
+    image: traefik/whoami
+    container_name: "simple-service1"
+    labels:
+      - "traefik.enable=true"
+      # Definition of the router
+      - "traefik.http.routers.router1.rule=Host(`localhost`) && Path(`/foo`)"
+      - "traefik.http.routers.router1.entrypoints=web"
+      - "traefik.http.routers.router1.middlewares=crowdsec1@docker" 
+      # Definition of the service
+      - "traefik.http.services.service1.loadbalancer.server.port=80"
+      # Definitin of the middleware
+      - "traefik.http.middlewares.crowdsec1.plugin.bouncer.enabled=true"
+      # crowdseclapikey must be uniq to the middleware attached to the service
+      - "traefik.http.middlewares.crowdsec1.plugin.bouncer.crowdseclapikey=40796d93c2958f9e58345514e67740e5"
+  whoami2:
+    image: traefik/whoami
+    container_name: "simple-service2"
+    labels:
+      - "traefik.enable=true"
+      # Definition of the router
+      - "traefik.http.routers.router2.rule=Host(`localhost`) && Path(`/bar`)"
+      - "traefik.http.routers.router2.entrypoints=web"
+      - "traefik.http.routers.router2.middlewares=crowdsec2@docker"
+      # Definition of the service
+      - "traefik.http.services.service2.loadbalancer.server.port=80"
+      # Definitin of the middleware
+      - "traefik.http.middlewares.crowdsec2.plugin.bouncer.enabled=true"
+      # crowdseclapikey must be uniq to the middleware attached to the service
+      - "traefik.http.middlewares.crowdsec2.plugin.bouncer.crowdseclapikey=44c36dac5c4140af9f06f397508e82c7"
 
   crowdsec:
     image: crowdsecurity/crowdsec:v1.4.1
     container_name: "crowdsec"
-    command: rm -rf /etc/crowdsec/acquis.yaml
     environment:
       COLLECTIONS: crowdsecurity/traefik
       CUSTOM_HOSTNAME: crowdsec
-      BOUNCER_KEY_TRAEFIK: 40796d93c2958f9e58345514e67740e5
+      BOUNCER_KEY_TRAEFIK_1: 40796d93c2958f9e58345514e67740e5
+      BOUNCER_KEY_TRAEFIK_2: 44c36dac5c4140af9f06f397508e82c7
     volumes:
       - ./acquis.yaml:/etc/crowdsec/acquis.yaml:ro
-      - logs:/var/log/traefik:ro
-      - crowdsec-db:/var/lib/crowdsec/data/
-      - crowdsec-config:/etc/crowdsec/
-
+      - logs-local:/var/log/traefik:ro
+      - crowdsec-db-local:/var/lib/crowdsec/data/
+      - crowdsec-config-local:/etc/crowdsec/
+    labels:
+      - "traefik.enable=false"
 volumes:
-  logs:
-  crowdsec-db:
-  crowdsec-config:
+  logs-local:
+  crowdsec-db-local:
+  crowdsec-config-local:

docker-compose.yml

@@ -2,18 +2,19 @@ version: "3.8"
 
 services:
   traefik:
-    image: "traefik:v2.8.7"
+    image: "traefik:v2.8.8"
     container_name: "traefik"
     command:
       # - "--log.level=DEBUG"
       - "--accesslog"
-      - "--accesslog.filepath=/var/log/traefik/traefik.log"
+      - "--accesslog.filepath=/var/log/traefik/access.log"
       - "--api.insecure=true"
       - "--providers.docker=true"
       - "--providers.docker.exposedbydefault=false"
       - "--entrypoints.web.address=:80"
 
       - "--experimental.plugins.bouncer.modulename=github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin"
+      - "--experimental.plugins.bouncer.version=v1.0.6"
     volumes:
       - "/var/run/docker.sock:/var/run/docker.sock:ro"
       - "logs:/var/log/traefik"
@@ -23,30 +24,54 @@ services:
     depends_on:
       - 'crowdsec'
 
-  whoami:
+  whoami1:
     image: traefik/whoami
-    container_name: "simple-service"
+    container_name: "simple-service-1"
     labels:
       - "traefik.enable=true"
-      - "traefik.http.routers.whoami.rule=Host(`localhost`)"
-      - "traefik.http.routers.whoami.entrypoints=web"
-      - "traefik.http.routers.whoami.middlewares=crowdsec@docker"
-      - "traefik.http.middlewares.crowdsec.plugin.bouncer.enabled=true"
-      - "traefik.http.middlewares.crowdsec.plugin.bouncer.crowdseclapikey=FIXME-LAPI-KEY"
+      # Definition of the router
+      - "traefik.http.routers.router1.rule=Host(`localhost`) && Path(`/foo`)"
+      - "traefik.http.routers.router1.entrypoints=web"
+      - "traefik.http.routers.router1.middlewares=crowdsec2@docker"
+      # Definition of the service
+      - "traefik.http.services.service1.loadbalancer.server.port=80"
+      # Definitin of the middleware
+      - "traefik.http.middlewares.crowdsec1.plugin.bouncer.enabled=true"
+      # crowdseclapikey must be uniq to the middleware attached to the service
+      - "traefik.http.middlewares.crowdsec1.plugin.bouncer.crowdseclapikey=FIXME-LAPI-KEY-1"
+  
+  whoami2:
+    image: traefik/whoami
+    container_name: "simple-service-2"
+    labels:
+      - "traefik.enable=true"
+      # Definition of the router
+      - "traefik.http.routers.router2.rule=Host(`localhost`) && Path(`/bar`)"
+      - "traefik.http.routers.router2.entrypoints=web"
+      - "traefik.http.routers.router2.middlewares=crowdsec2@docker"
+      # Definition of the service
+      - "traefik.http.services.service2.loadbalancer.server.port=80"
+      # Definitin of the middleware
+      - "traefik.http.middlewares.crowdsec2.plugin.bouncer.enabled=true"
+      # crowdseclapikey must be uniq to the middleware attached to the service
+      - "traefik.http.middlewares.crowdsec2.plugin.bouncer.crowdseclapikey=FIXME-LAPI-KEY-2"
 
   crowdsec:
     image: crowdsecurity/crowdsec:v1.4.1
     container_name: "crowdsec"
-    command: rm -rf /etc/crowdsec/acquis.yaml
     environment:
       COLLECTIONS: crowdsecurity/traefik
       CUSTOM_HOSTNAME: crowdsec
-      BOUNCER_KEY_TRAEFIK: FIXME-LAPI-KEY
+      # We need to register one api key per service we will use
+      BOUNCER_KEY_TRAEFIK_1: FIXME-LAPI-KEY-1
+      BOUNCER_KEY_TRAEFIK_2: FIXME-LAPI-KEY-2
     volumes:
       - ./acquis.yaml:/etc/crowdsec/acquis.yaml:ro
       - logs:/var/log/traefik:ro
       - crowdsec-db:/var/lib/crowdsec/data/
       - crowdsec-config:/etc/crowdsec/
+    labels:
+      - "traefik.enable=false"
 
 volumes:
   logs: