mirror of
https://github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin.git
synced 2026-09-03 12:38:51 +02:00
+50
-49
@@ -31,7 +31,7 @@ const (
|
|||||||
|
|
||||||
//nolint:gochecknoglobals
|
//nolint:gochecknoglobals
|
||||||
var (
|
var (
|
||||||
crowdsecStreamHealthy = false
|
isCrowdsecStreamHealthy = false
|
||||||
ticker chan bool
|
ticker chan bool
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -126,7 +126,7 @@ func New(ctx context.Context, next http.Handler, config *Config, name string) (h
|
|||||||
MaxIdleConns: 10,
|
MaxIdleConns: 10,
|
||||||
IdleConnTimeout: 30 * time.Second,
|
IdleConnTimeout: 30 * time.Second,
|
||||||
},
|
},
|
||||||
Timeout: 5 * time.Second,
|
Timeout: 2 * time.Second,
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
if config.RedisCacheEnabled {
|
if config.RedisCacheEnabled {
|
||||||
@@ -154,30 +154,31 @@ func (bouncer *Bouncer) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
|
|||||||
// Here we check for the trusted IPs in the customHeader
|
// Here we check for the trusted IPs in the customHeader
|
||||||
remoteIP, err := ip.GetRemoteIP(req, bouncer.serverPoolStrategy, bouncer.customHeader)
|
remoteIP, err := ip.GetRemoteIP(req, bouncer.serverPoolStrategy, bouncer.customHeader)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logger.Error(fmt.Sprintf("ServeHTTP ip:%s %s", remoteIP, err.Error()))
|
logger.Error(fmt.Sprintf("ServeHTTP:getRemoteIp ip:%s %s", remoteIP, err.Error()))
|
||||||
rw.WriteHeader(http.StatusForbidden)
|
rw.WriteHeader(http.StatusForbidden)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
trusted, err := bouncer.clientPoolStrategy.Checker.Contains(remoteIP)
|
isTrusted, err := bouncer.clientPoolStrategy.Checker.Contains(remoteIP)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logger.Info(err.Error())
|
logger.Error(fmt.Sprintf("ServeHTTP:checkerContains ip:%s %s", remoteIP, err.Error()))
|
||||||
|
rw.WriteHeader(http.StatusForbidden)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
// if our IP is in the trusted list we bypass the next checks
|
// if our IP is in the trusted list we bypass the next checks
|
||||||
logger.Debug(fmt.Sprintf("ServeHTTP ip:%s isTrusted:%v", remoteIP, trusted))
|
logger.Debug(fmt.Sprintf("ServeHTTP ip:%s isTrusted:%v", remoteIP, isTrusted))
|
||||||
if trusted {
|
if isTrusted {
|
||||||
bouncer.next.ServeHTTP(rw, req)
|
bouncer.next.ServeHTTP(rw, req)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO This should be simplified
|
// TODO This should be simplified
|
||||||
healthy := crowdsecStreamHealthy
|
|
||||||
if bouncer.crowdsecMode != noneMode {
|
if bouncer.crowdsecMode != noneMode {
|
||||||
isBanned, err := cache.GetDecision(remoteIP)
|
isBanned, erro := cache.GetDecision(remoteIP)
|
||||||
if err != nil {
|
if erro != nil {
|
||||||
logger.Error(err.Error())
|
logger.Debug(fmt.Sprintf("ServeHTTP:getDecision ip:%s %s", remoteIP, erro.Error()))
|
||||||
if err.Error() == simpleredis.RedisUnreachable {
|
if erro.Error() == simpleredis.RedisUnreachable {
|
||||||
healthy = false
|
rw.WriteHeader(http.StatusForbidden)
|
||||||
|
return
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
logger.Debug(fmt.Sprintf("ServeHTTP ip:%s cache:hit isBanned:%v", remoteIP, isBanned))
|
logger.Debug(fmt.Sprintf("ServeHTTP ip:%s cache:hit isBanned:%v", remoteIP, isBanned))
|
||||||
@@ -192,13 +193,20 @@ 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.
|
// Right here if we cannot join the stream we forbid the request to go on.
|
||||||
if bouncer.crowdsecMode == streamMode {
|
if bouncer.crowdsecMode == streamMode {
|
||||||
if healthy {
|
if isCrowdsecStreamHealthy {
|
||||||
bouncer.next.ServeHTTP(rw, req)
|
logger.Error(fmt.Sprintf("ServeHTTP:isCrowdsecStreamHealthy ip:%s", remoteIP))
|
||||||
} else {
|
|
||||||
rw.WriteHeader(http.StatusForbidden)
|
rw.WriteHeader(http.StatusForbidden)
|
||||||
|
} else {
|
||||||
|
bouncer.next.ServeHTTP(rw, req)
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
handleNoStreamCache(bouncer, rw, req, remoteIP)
|
err = handleNoStreamCache(bouncer, remoteIP)
|
||||||
|
if err != nil {
|
||||||
|
logger.Debug(err.Error())
|
||||||
|
rw.WriteHeader(http.StatusForbidden)
|
||||||
|
} else {
|
||||||
|
bouncer.next.ServeHTTP(rw, req)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -250,7 +258,8 @@ func startTicker(config *Config, work func()) chan bool {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// We are now in none or live mode.
|
// We are now in none or live mode.
|
||||||
func handleNoStreamCache(bouncer *Bouncer, rw http.ResponseWriter, req *http.Request, remoteIP string) {
|
func handleNoStreamCache(bouncer *Bouncer, remoteIP string) error {
|
||||||
|
isLiveMode := bouncer.crowdsecMode == liveMode
|
||||||
routeURL := url.URL{
|
routeURL := url.URL{
|
||||||
Scheme: bouncer.crowdsecScheme,
|
Scheme: bouncer.crowdsecScheme,
|
||||||
Host: bouncer.crowdsecHost,
|
Host: bouncer.crowdsecHost,
|
||||||
@@ -259,42 +268,35 @@ func handleNoStreamCache(bouncer *Bouncer, rw http.ResponseWriter, req *http.Req
|
|||||||
}
|
}
|
||||||
body, err := crowdsecQuery(bouncer, routeURL.String())
|
body, err := crowdsecQuery(bouncer, routeURL.String())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logger.Info(err.Error())
|
return err
|
||||||
rw.WriteHeader(http.StatusForbidden)
|
|
||||||
return
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if bytes.Equal(body, []byte("null")) {
|
if bytes.Equal(body, []byte("null")) {
|
||||||
if bouncer.crowdsecMode == liveMode {
|
if isLiveMode {
|
||||||
cache.SetDecision(remoteIP, false, bouncer.defaultDecisionTimeout)
|
cache.SetDecision(remoteIP, false, bouncer.defaultDecisionTimeout)
|
||||||
}
|
}
|
||||||
bouncer.next.ServeHTTP(rw, req)
|
return nil
|
||||||
return
|
|
||||||
}
|
}
|
||||||
|
|
||||||
var decisions []Decision
|
var decisions []Decision
|
||||||
err = json.Unmarshal(body, &decisions)
|
err = json.Unmarshal(body, &decisions)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logger.Info(fmt.Sprintf("failed to parse body: %s", err))
|
return fmt.Errorf("handleNoStreamCache:parseBody %w", err)
|
||||||
rw.WriteHeader(http.StatusForbidden)
|
|
||||||
return
|
|
||||||
}
|
}
|
||||||
if len(decisions) == 0 {
|
if len(decisions) == 0 {
|
||||||
if bouncer.crowdsecMode == liveMode {
|
if isLiveMode {
|
||||||
cache.SetDecision(remoteIP, false, bouncer.defaultDecisionTimeout)
|
cache.SetDecision(remoteIP, false, bouncer.defaultDecisionTimeout)
|
||||||
}
|
}
|
||||||
bouncer.next.ServeHTTP(rw, req)
|
return nil
|
||||||
return
|
|
||||||
}
|
}
|
||||||
rw.WriteHeader(http.StatusForbidden)
|
|
||||||
duration, err := time.ParseDuration(decisions[0].Duration)
|
duration, err := time.ParseDuration(decisions[0].Duration)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logger.Info(fmt.Sprintf("failed to parse duration: %s", err))
|
return fmt.Errorf("handleNoStreamCache:parseDuration %w", err)
|
||||||
return
|
|
||||||
}
|
}
|
||||||
if bouncer.crowdsecMode == liveMode {
|
if isLiveMode {
|
||||||
cache.SetDecision(remoteIP, true, int64(duration.Seconds()))
|
cache.SetDecision(remoteIP, true, int64(duration.Seconds()))
|
||||||
}
|
}
|
||||||
|
return fmt.Errorf("handleNoStreamCache:banned")
|
||||||
}
|
}
|
||||||
|
|
||||||
func handleStreamCache(bouncer *Bouncer) {
|
func handleStreamCache(bouncer *Bouncer) {
|
||||||
@@ -302,9 +304,9 @@ func handleStreamCache(bouncer *Bouncer) {
|
|||||||
// Instead of blocking the goroutine interval for all the secondary node,
|
// Instead of blocking the goroutine interval for all the secondary node,
|
||||||
// if the master service is shut down, other goroutine can take the lead
|
// if the master service is shut down, other goroutine can take the lead
|
||||||
// because updated routine information is in the cache
|
// because updated routine information is in the cache
|
||||||
logger.Debug("handleStreamCache")
|
|
||||||
_, err := cache.GetDecision(cacheTimeoutKey)
|
_, err := cache.GetDecision(cacheTimeoutKey)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
|
logger.Debug("handleStreamCache:alreadyUpdated")
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
cache.SetDecision(cacheTimeoutKey, false, bouncer.updateInterval-1)
|
cache.SetDecision(cacheTimeoutKey, false, bouncer.updateInterval-1)
|
||||||
@@ -312,19 +314,19 @@ func handleStreamCache(bouncer *Bouncer) {
|
|||||||
Scheme: bouncer.crowdsecScheme,
|
Scheme: bouncer.crowdsecScheme,
|
||||||
Host: bouncer.crowdsecHost,
|
Host: bouncer.crowdsecHost,
|
||||||
Path: crowdsecLapiStreamRoute,
|
Path: crowdsecLapiStreamRoute,
|
||||||
RawQuery: fmt.Sprintf("startup=%t", !crowdsecStreamHealthy),
|
RawQuery: fmt.Sprintf("startup=%t", !isCrowdsecStreamHealthy),
|
||||||
}
|
}
|
||||||
body, err := crowdsecQuery(bouncer, streamRouteURL.String())
|
body, err := crowdsecQuery(bouncer, streamRouteURL.String())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logger.Info(err.Error())
|
logger.Error(err.Error())
|
||||||
crowdsecStreamHealthy = false
|
isCrowdsecStreamHealthy = false
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
var stream Stream
|
var stream Stream
|
||||||
err = json.Unmarshal(body, &stream)
|
err = json.Unmarshal(body, &stream)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logger.Info(fmt.Sprintf("error while parsing body: %s", err))
|
logger.Error(fmt.Sprintf("handleStreamCache:parsingBody %s", err.Error()))
|
||||||
crowdsecStreamHealthy = false
|
isCrowdsecStreamHealthy = false
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
for _, decision := range stream.New {
|
for _, decision := range stream.New {
|
||||||
@@ -336,7 +338,7 @@ func handleStreamCache(bouncer *Bouncer) {
|
|||||||
for _, decision := range stream.Deleted {
|
for _, decision := range stream.Deleted {
|
||||||
cache.DeleteDecision(decision.Value)
|
cache.DeleteDecision(decision.Value)
|
||||||
}
|
}
|
||||||
crowdsecStreamHealthy = true
|
isCrowdsecStreamHealthy = true
|
||||||
}
|
}
|
||||||
|
|
||||||
func crowdsecQuery(bouncer *Bouncer, stringURL string) ([]byte, error) {
|
func crowdsecQuery(bouncer *Bouncer, stringURL string) ([]byte, error) {
|
||||||
@@ -345,21 +347,20 @@ func crowdsecQuery(bouncer *Bouncer, stringURL string) ([]byte, error) {
|
|||||||
req.Header.Add(crowdsecLapiHeader, bouncer.crowdsecKey)
|
req.Header.Add(crowdsecLapiHeader, bouncer.crowdsecKey)
|
||||||
res, err := bouncer.client.Do(req)
|
res, err := bouncer.client.Do(req)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("error while fetching %v: %w", stringURL, err)
|
return nil, fmt.Errorf("crowdsecQuery url:%s %w", stringURL, err)
|
||||||
}
|
}
|
||||||
if res.StatusCode != http.StatusOK {
|
if res.StatusCode != http.StatusOK {
|
||||||
return nil, fmt.Errorf("error while fetching %v, status code: %d", stringURL, res.StatusCode)
|
return nil, fmt.Errorf("crowdsecQuery url:%s, statusCode:%d", stringURL, res.StatusCode)
|
||||||
}
|
}
|
||||||
defer func(body io.ReadCloser) {
|
defer func() {
|
||||||
err = body.Close()
|
if err = res.Body.Close(); err != nil {
|
||||||
if err != nil {
|
logger.Error(fmt.Sprintf("crowdsecQuery:closeBody %s", err.Error()))
|
||||||
logger.Error(fmt.Sprintf("failed to close body reader: %s", err.Error()))
|
|
||||||
}
|
}
|
||||||
}(res.Body)
|
}()
|
||||||
body, err := io.ReadAll(res.Body)
|
body, err := io.ReadAll(res.Body)
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("error while reading body: %w", err)
|
return nil, fmt.Errorf("crowdsecQuery:readBody %w", err)
|
||||||
}
|
}
|
||||||
return body, nil
|
return body, nil
|
||||||
}
|
}
|
||||||
|
|||||||
+4
-5
@@ -3,7 +3,6 @@
|
|||||||
package ip
|
package ip
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"errors"
|
|
||||||
"fmt"
|
"fmt"
|
||||||
"net"
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
@@ -45,12 +44,12 @@ func NewChecker(trustedIPs []string) (*Checker, error) {
|
|||||||
// Contains checks if provided address is in the trusted IPs.
|
// Contains checks if provided address is in the trusted IPs.
|
||||||
func (ip *Checker) Contains(addr string) (bool, error) {
|
func (ip *Checker) Contains(addr string) (bool, error) {
|
||||||
if len(addr) == 0 {
|
if len(addr) == 0 {
|
||||||
return false, errors.New("empty IP address")
|
return false, fmt.Errorf("Contains:noAddress")
|
||||||
}
|
}
|
||||||
|
|
||||||
ipAddr, err := parseIP(addr)
|
ipAddr, err := parseIP(addr)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, fmt.Errorf("unable to parse address: %s: %w", addr, err)
|
return false, fmt.Errorf("Contains:parseAddress addr:%s %w", addr, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
return ip.ContainsIP(ipAddr), nil
|
return ip.ContainsIP(ipAddr), nil
|
||||||
@@ -76,7 +75,7 @@ func (ip *Checker) ContainsIP(addr net.IP) bool {
|
|||||||
func parseIP(addr string) (net.IP, error) {
|
func parseIP(addr string) (net.IP, error) {
|
||||||
userIP := net.ParseIP(addr)
|
userIP := net.ParseIP(addr)
|
||||||
if userIP == nil {
|
if userIP == nil {
|
||||||
return nil, fmt.Errorf("can't parse IP from address %s", addr)
|
return nil, fmt.Errorf("parseIP:parseAddress %s", addr)
|
||||||
}
|
}
|
||||||
|
|
||||||
return userIP, nil
|
return userIP, nil
|
||||||
@@ -123,7 +122,7 @@ func GetRemoteIP(req *http.Request, strategy *PoolStrategy, customHeader string)
|
|||||||
}
|
}
|
||||||
remoteIP, _, err := net.SplitHostPort(req.RemoteAddr)
|
remoteIP, _, err := net.SplitHostPort(req.RemoteAddr)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", fmt.Errorf("failed to extract ip from remote address: %w", err)
|
return "", fmt.Errorf("GetRemoteIP:extractIP: %w", err)
|
||||||
}
|
}
|
||||||
return remoteIP, nil
|
return remoteIP, nil
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user