diff --git a/Makefile b/Makefile index ac90851..b506c01 100644 --- a/Makefile +++ b/Makefile @@ -31,15 +31,24 @@ run_behindproxy: run: docker-compose -f docker-compose.yml up -d --remove-orphans -restart_docker_dev: +restart_dev: docker-compose -f docker-compose.dev.yml restart -restart_docker_local: +restart_local: docker-compose -f docker-compose.local.yml restart -restart_docker: +restart: docker-compose -f docker-compose.yml restart +show_logs: + docker-compose -f docker-compose.yml restart + +show_local_logs: + docker-compose -f docker-compose.local.yml logs -f + +show_dev_logs: + docker-compose -f docker-compose.dev.yml logs -f + clean_all_docker: docker-compose -f exemples/behind-proxy/docker-compose.cloudflare.yml down --remove-orphans docker-compose -f docker-compose.local.yml down --remove-orphans diff --git a/README.md b/README.md index 7f282e8..c07db75 100644 --- a/README.md +++ b/README.md @@ -49,6 +49,9 @@ At each start of synchronisation, the middleware will wait a random number of se - Enabled - bool - enable the plugin +- LogLevel + - string + - default: `INFO`, expected value are: `INFO`, `DEBUG` - CrowdsecMode - string - default: `live`, expected value are: `none`, `live`, `stream` @@ -120,7 +123,7 @@ http: enabled: false updateIntervalSeconds: 60 defaultDecisionSeconds: 60 - crowdsecMode: stream + crowdsecMode: live crowdsecLapiKey: privateKey crowdsecLapiHost: crowdsec:8080 crowdsecLapiScheme: http diff --git a/bouncer.go b/bouncer.go index 3b0c181..aa5ec3e 100644 --- a/bouncer.go +++ b/bouncer.go @@ -7,16 +7,14 @@ import ( "fmt" "io" "io/ioutil" - "log" - "math/rand" - "net" "net/http" "net/url" "text/template" "time" - ttl_map "github.com/leprosus/golang-ttl-map" - ip "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/ip" + cache "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/cache" + ip "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/ip" + logger "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/logger" ) const ( @@ -26,28 +24,33 @@ const ( crowdsecLapiHeader = "X-Api-Key" crowdsecLapiRoute = "v1/decisions" crowdsecLapiStreamRoute = "v1/decisions/stream" - cacheBannedValue = "t" - cacheNoBannedValue = "f" - xForwardedFor = "X-Forwarded-For" + cacheTimeoutKey = "updated" +) + +var ( + crowdsecStreamHealthy = false + ticker chan bool ) // Config the plugin configuration. type Config struct { Enabled bool `json:"enabled,omitempty"` + LogLevel string `json:"logLevel,omitempty"` CrowdsecMode string `json:"crowdsecMode,omitempty"` CrowdsecLapiScheme string `json:"crowdsecLapiScheme,omitempty"` CrowdsecLapiHost string `json:"crowdsecLapiHost,omitempty"` CrowdsecLapiKey string `json:"crowdsecLapiKey,omitempty"` + ForwardedHeadersCustomName string `json:"forwardedheaderscustomheader,omitempty"` UpdateIntervalSeconds int64 `json:"updateIntervalSeconds,omitempty"` DefaultDecisionSeconds int64 `json:"defaultDecisionSeconds,omitempty"` ForwardedHeadersTrustedIPs []string `json:"forwardedheaderstrustedips,omitempty"` - ForwardedHeadersCustomName string `json:"forwardedheaderscustomheader,omitempty"` } // CreateConfig creates the default plugin configuration. func CreateConfig() *Config { return &Config{ Enabled: false, + LogLevel: "INFO", CrowdsecMode: liveMode, CrowdsecLapiScheme: "http", CrowdsecLapiHost: "crowdsec:8080", @@ -55,18 +58,17 @@ func CreateConfig() *Config { UpdateIntervalSeconds: 60, DefaultDecisionSeconds: 60, ForwardedHeadersTrustedIPs: []string{}, - ForwardedHeadersCustomName: xForwardedFor, + ForwardedHeadersCustomName: "X-Forwarded-For", } } -// Bouncer a Bouncer plugin. +// Bouncer a Bouncer struct. type Bouncer struct { next http.Handler name string template *template.Template enabled bool - crowdsecStreamHealthy bool crowdsecScheme string crowdsecHost string crowdsecKey string @@ -76,13 +78,14 @@ type Bouncer struct { customHeader string poolStrategy *ip.PoolStrategy client *http.Client - cache *ttl_map.Heap } // New creates the crowdsec bouncer plugin. func New(ctx context.Context, next http.Handler, config *Config, name string) (http.Handler, error) { + logger.Init(config.LogLevel) err := validateParams(config) if err != nil { + logger.Info(fmt.Sprintf("%w", err)) return nil, err } @@ -94,7 +97,6 @@ func New(ctx context.Context, next http.Handler, config *Config, name string) (h template: template.New("CrowdsecBouncer").Delims("[[", "]]"), enabled: config.Enabled, - crowdsecStreamHealthy: false, crowdsecMode: config.CrowdsecMode, crowdsecScheme: config.CrowdsecLapiScheme, crowdsecHost: config.CrowdsecLapiHost, @@ -112,18 +114,14 @@ func New(ctx context.Context, next http.Handler, config *Config, name string) (h }, Timeout: 5 * time.Second, }, - cache: ttl_map.New(), } if config.CrowdsecMode == streamMode { go func() { - rand.Seed(time.Now().UnixNano()) - timeout := rand.Int63n(30) - logger(fmt.Sprintf("Wait: %v", timeout)) - time.Sleep(time.Duration(timeout) * time.Second) - go handleStreamCache(bouncer) - ticker := time.NewTicker(time.Duration(config.UpdateIntervalSeconds) * time.Second) - for range ticker.C { + if ticker == nil { go handleStreamCache(bouncer) + ticker = startTicker(config, func() { + handleStreamCache(bouncer) + }) } }() } @@ -137,15 +135,18 @@ func (bouncer *Bouncer) ServeHTTP(rw http.ResponseWriter, req *http.Request) { return } - remoteHost, err := getRemoteIP(bouncer, req) + remoteHost, err := ip.GetRemoteIP(req, bouncer.poolStrategy, bouncer.customHeader) if err != nil { + logger.Info(fmt.Sprintf("%w", err)) bouncer.next.ServeHTTP(rw, req) return } + logger.Debug(fmt.Sprintf("ServeHTTP ip:%v", remoteHost)) if bouncer.crowdsecMode != noneMode { - isBanned, err := getDecision(bouncer.cache, remoteHost) + isBanned, err := cache.GetDecision(remoteHost) if err == nil { + logger.Debug(fmt.Sprintf("ServeHTTP cacheHit isBanned:%v", isBanned)) if isBanned { rw.WriteHeader(http.StatusForbidden) } else { @@ -157,7 +158,7 @@ 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 == streamMode { - if bouncer.crowdsecStreamHealthy { + if crowdsecStreamHealthy { bouncer.next.ServeHTTP(rw, req) } else { rw.WriteHeader(http.StatusForbidden) @@ -188,10 +189,6 @@ type Stream struct { New []Decision `json:"new"` } -func logger(str string) { - log.Printf("Crowdsec Bouncer Traefik Plugin - %s", str) -} - func contains(source []string, target string) bool { for _, item := range source { if item == target { @@ -201,42 +198,21 @@ func contains(source []string, target string) bool { return false } -// It returns the first IP that is not in the pool, or the empty string otherwise. -func getRemoteIP(bouncer *Bouncer, req *http.Request) (string, error) { - remoteIP := bouncer.poolStrategy.GetIP(req, bouncer.customHeader) - if len(remoteIP) != 0 { - return remoteIP, nil - } - remoteIP, _, err := net.SplitHostPort(req.RemoteAddr) - if err != nil { - logger(fmt.Sprintf("failed to extract ip from remote address: %v", err)) - return "", err - } - return remoteIP, nil -} - -// 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) - bannedString, isValid := banned.(string) - if isCached && isValid && len(bannedString) > 0 { - return bannedString == cacheBannedValue, nil - } - return false, fmt.Errorf("no cache data") -} - -func setDecision(cache *ttl_map.Heap, clientIP string, isBanned bool, duration int64) { - if isBanned { - logger(fmt.Sprintf("%v banned", clientIP)) - cache.Set(clientIP, cacheBannedValue, duration) - } else { - cache.Set(clientIP, cacheNoBannedValue, duration) - } -} - -func deleteDecision(cache *ttl_map.Heap, clientIP string) { - cache.Del(clientIP) +func startTicker(config *Config, work func()) chan bool { + ticker := time.NewTicker(time.Duration(config.UpdateIntervalSeconds) * time.Second) + stop := make(chan bool, 1) + go func() { + defer logger.Debug("ticker:stopped") + for { + select { + case <-ticker.C: + go work() + case <-stop: + return + } + } + }() + return stop } // We are now in none or live mode. @@ -247,26 +223,31 @@ func handleNoStreamCache(bouncer *Bouncer, rw http.ResponseWriter, req *http.Req Path: crowdsecLapiRoute, RawQuery: fmt.Sprintf("ip=%v&banned=true", remoteHost), } - body := crowdsecQuery(bouncer, routeURL.String()) + body, err := crowdsecQuery(bouncer, routeURL.String()) + if err != nil { + logger.Info(fmt.Sprintf("%w", err)) + rw.WriteHeader(http.StatusForbidden) + return + } if bytes.Equal(body, []byte("null")) { if bouncer.crowdsecMode == liveMode { - setDecision(bouncer.cache, remoteHost, false, bouncer.defaultDecisionTimeout) + cache.SetDecision(remoteHost, false, bouncer.defaultDecisionTimeout) } bouncer.next.ServeHTTP(rw, req) return } var decisions []Decision - err := json.Unmarshal(body, &decisions) + err = json.Unmarshal(body, &decisions) if err != nil { - logger(fmt.Sprintf("failed to parse body: %s", err)) + logger.Info(fmt.Sprintf("failed to parse body: %s", err)) rw.WriteHeader(http.StatusForbidden) return } if len(decisions) == 0 { if bouncer.crowdsecMode == liveMode { - setDecision(bouncer.cache, remoteHost, false, bouncer.defaultDecisionTimeout) + cache.SetDecision(remoteHost, false, bouncer.defaultDecisionTimeout) } bouncer.next.ServeHTTP(rw, req) return @@ -274,70 +255,78 @@ func handleNoStreamCache(bouncer *Bouncer, rw http.ResponseWriter, req *http.Req rw.WriteHeader(http.StatusForbidden) duration, err := time.ParseDuration(decisions[0].Duration) if err != nil { - logger(fmt.Sprintf("failed to parse duration: %s", err)) + logger.Info(fmt.Sprintf("failed to parse duration: %s", err)) return } if bouncer.crowdsecMode == liveMode { - setDecision(bouncer.cache, remoteHost, true, int64(duration.Seconds())) + cache.SetDecision(remoteHost, true, int64(duration.Seconds())) } } func handleStreamCache(bouncer *Bouncer) { // TODO clean properly on exit. + // Instead of blocking the goroutine interval for all the secondary node, + // if the master service is shut down, other goroutine can take the lead + // because updated routine information is in the cache + logger.Debug("handleStreamCache") + _, err := cache.GetDecision(cacheTimeoutKey) + if err == nil { + return + } + cache.SetDecision(cacheTimeoutKey, false, bouncer.updateInterval-1) streamRouteURL := url.URL{ Scheme: bouncer.crowdsecScheme, Host: bouncer.crowdsecHost, Path: crowdsecLapiStreamRoute, - RawQuery: fmt.Sprintf("startup=%t", !bouncer.crowdsecStreamHealthy), + RawQuery: fmt.Sprintf("startup=%t", !crowdsecStreamHealthy), } - body := crowdsecQuery(bouncer, streamRouteURL.String()) - var stream Stream - err := json.Unmarshal(body, &stream) + body, err := crowdsecQuery(bouncer, streamRouteURL.String()) if err != nil { - logger(fmt.Sprintf("error while parsing body: %s", err)) - bouncer.crowdsecStreamHealthy = false + logger.Info(fmt.Sprintf("%w", err)) + crowdsecStreamHealthy = false + return + } + var stream Stream + err = json.Unmarshal(body, &stream) + if err != nil { + logger.Info(fmt.Sprintf("error while parsing body: %s", err)) + crowdsecStreamHealthy = false return } for _, decision := range stream.New { duration, err := time.ParseDuration(decision.Duration) if err == nil { - setDecision(bouncer.cache, decision.Value, true, int64(duration.Seconds())) + cache.SetDecision(decision.Value, true, int64(duration.Seconds())) } } for _, decision := range stream.Deleted { - deleteDecision(bouncer.cache, decision.Value) + cache.DeleteDecision(decision.Value) } - bouncer.crowdsecStreamHealthy = true + crowdsecStreamHealthy = true } -func crowdsecQuery(bouncer *Bouncer, stringURL string) []byte { +func crowdsecQuery(bouncer *Bouncer, stringURL string) ([]byte, error) { var req *http.Request req, _ = http.NewRequest(http.MethodGet, stringURL, nil) req.Header.Add(crowdsecLapiHeader, bouncer.crowdsecKey) res, err := bouncer.client.Do(req) if err != nil { - logger(fmt.Sprintf("error while fetching %v: %s", stringURL, err)) - bouncer.crowdsecStreamHealthy = false - return nil + return nil, fmt.Errorf("error while fetching %v: %s", stringURL, err) } if res.StatusCode != http.StatusOK { - logger(fmt.Sprintf("error while fetching %v, status code: %d", stringURL, res.StatusCode)) - bouncer.crowdsecStreamHealthy = false - return nil + return nil, fmt.Errorf("error while fetching %v, status code: %d", stringURL, res.StatusCode) } defer func(body io.ReadCloser) { err = body.Close() if err != nil { - logger(fmt.Sprintf("failed to close body reader: %s", err)) + logger.Info(fmt.Sprintf("failed to close body reader: %s", err)) } }(res.Body) body, err := ioutil.ReadAll(res.Body) if err != nil { - logger(fmt.Sprintf("error while reading body: %s", err)) - bouncer.crowdsecStreamHealthy = false - return nil + return nil, fmt.Errorf("error while reading body: %s", err) } - return body + return body, nil } func validateParams(config *Config) error { @@ -378,10 +367,10 @@ func validateParams(config *Config) error { if len(config.ForwardedHeadersTrustedIPs) > 0 { _, err = ip.NewChecker(config.ForwardedHeadersTrustedIPs) if err != nil { - return fmt.Errorf("ForwardedHeadersTrustedIPs must be a list of IP/CIDR :%v", err) + return fmt.Errorf("ForwardedHeadersTrustedIPs must be a list of IP/CIDR :%w", err) } } else { - logger("No IP provided for ForwardedHeadersTrustedIPs") + logger.Debug("No IP provided for ForwardedHeadersTrustedIPs") } return nil diff --git a/docker-compose.local.yml b/docker-compose.local.yml index 9cd4494..fc8f834 100644 --- a/docker-compose.local.yml +++ b/docker-compose.local.yml @@ -35,7 +35,7 @@ services: - "traefik.http.services.service1.loadbalancer.server.port=80" - "traefik.http.middlewares.crowdsec1.plugin.bouncer.enabled=true" - "traefik.http.middlewares.crowdsec1.plugin.bouncer.crowdseclapikey=40796d93c2958f9e58345514e67740e5" - - "traefik.http.middlewares.crowdsec1.plugin.bouncer.forwardedheaderstrustedips=172.26.0.1" + - "traefik.http.middlewares.crowdsec1.plugin.bouncer.forwardedheaderstrustedips=172.26.0.5" whoami2: image: traefik/whoami @@ -48,6 +48,7 @@ services: - "traefik.http.services.service2.loadbalancer.server.port=80" - "traefik.http.middlewares.crowdsec2.plugin.bouncer.enabled=true" - "traefik.http.middlewares.crowdsec2.plugin.bouncer.crowdseclapikey=44c36dac5c4140af9f06f397508e82c7" + - "traefik.http.middlewares.crowdsec2.plugin.bouncer.forwardedheaderstrustedips=172.26.0.5" crowdsec: image: crowdsecurity/crowdsec:v1.4.1 diff --git a/pkg/cache/cache.go b/pkg/cache/cache.go new file mode 100644 index 0000000..c734c89 --- /dev/null +++ b/pkg/cache/cache.go @@ -0,0 +1,39 @@ +package cache + +import ( + "fmt" + + ttl_map "github.com/leprosus/golang-ttl-map" + + logger "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/logger" +) +const ( + cacheBannedValue = "t" + cacheNoBannedValue = "f" +) + +var cache = ttl_map.New() + +// 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(clientIP string) (bool, error) { + banned, isCached := cache.Get(clientIP) + bannedString, isValid := banned.(string) + if isCached && isValid && len(bannedString) > 0 { + return bannedString == cacheBannedValue, nil + } + return false, fmt.Errorf("no cache data") +} + +func SetDecision(clientIP string, isBanned bool, duration int64) { + if isBanned { + logger.Debug(fmt.Sprintf("%v banned", clientIP)) + cache.Set(clientIP, cacheBannedValue, duration) + } else { + cache.Set(clientIP, cacheNoBannedValue, duration) + } +} + +func DeleteDecision(clientIP string) { + cache.Del(clientIP) +} diff --git a/ip/ip.go b/pkg/ip/ip.go similarity index 81% rename from ip/ip.go rename to pkg/ip/ip.go index 226416a..622d8ba 100644 --- a/ip/ip.go +++ b/pkg/ip/ip.go @@ -8,7 +8,7 @@ import ( "strings" ) -//CHECKER +// CHECKER // Checker allows to check that addresses are in a trusted IPs. type Checker struct { @@ -91,7 +91,7 @@ type PoolStrategy struct { // GetIP checks the list of Forwarded IPs (most recent first) against the // Checker pool of IPs. It returns the first IP that is not in the pool, or the // empty string otherwise. -func (s *PoolStrategy) GetIP(req *http.Request, customHeader string) string { +func (s *PoolStrategy) getIP(req *http.Request, customHeader string) string { if s.Checker == nil { return "" } @@ -112,3 +112,16 @@ func (s *PoolStrategy) GetIP(req *http.Request, customHeader string) string { return "" } + +// GetRemoteIP It returns the first IP that is not in the pool, or the empty string otherwise. +func GetRemoteIP(req *http.Request, strategy *PoolStrategy, customHeader string) (string, error) { + remoteIP := strategy.getIP(req, customHeader) + if len(remoteIP) != 0 { + return remoteIP, nil + } + remoteIP, _, err := net.SplitHostPort(req.RemoteAddr) + if err != nil { + return "", fmt.Errorf("failed to extract ip from remote address: %w", err) + } + return remoteIP, nil +} diff --git a/pkg/logger/logger.go b/pkg/logger/logger.go new file mode 100644 index 0000000..b82977a --- /dev/null +++ b/pkg/logger/logger.go @@ -0,0 +1,35 @@ +package logger + +import ( + "io" + "log" + "os" +) + +var ( + loggerInfo = log.New(io.Discard, "INFO: CrowdsecBouncerTraefikPlugin: ", log.Ldate|log.Ltime) + loggerDebug = log.New(io.Discard, "DEBUG: CrowdsecBouncerTraefikPlugin: ", log.Ldate|log.Ltime) +) + +// Init Set Default log level to info in case log level to defined +func Init(logLevel string) { + switch logLevel { + case "INFO": + loggerInfo.SetOutput(os.Stdout) + case "DEBUG": + loggerInfo.SetOutput(os.Stdout) + loggerDebug.SetOutput(os.Stdout) + default: + loggerInfo.SetOutput(os.Stdout) + } +} + +// Info Log info +func Info(str string) { + loggerInfo.Printf(str) +} + +// Info Log debug +func Debug(str string) { + loggerDebug.Printf(str) +}