🍱 fix logic + logs

This commit is contained in:
Max Lerebourg
2022-11-20 12:04:05 +01:00
parent 80d49e3969
commit d1d64689af
2 changed files with 42 additions and 46 deletions
+39 -43
View File
@@ -31,8 +31,8 @@ const (
//nolint:gochecknoglobals //nolint:gochecknoglobals
var ( var (
crowdsecStreamHealthy = false isCrowdsecStreamHealthy = false
ticker chan bool ticker chan bool
) )
// Config the plugin configuration. // Config the plugin configuration.
@@ -154,51 +154,45 @@ 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 %w", remoteIP, err)) logger.Error(fmt.Sprintf("ServeHTTP 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(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, err := cache.GetDecision(remoteIP)
if err != nil { if err != nil {
logger.Debug(err.Error()) logger.Debug(err.Error())
if err.Error() == simpleredis.RedisUnreachable { if err.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))
if isBanned { response(isBanned, bouncer, rw, req)
rw.WriteHeader(http.StatusForbidden)
} else {
bouncer.next.ServeHTTP(rw, req)
}
return return
} }
} }
// 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 { response(isCrowdsecStreamHealthy, bouncer, rw, req)
bouncer.next.ServeHTTP(rw, req)
} else {
rw.WriteHeader(http.StatusForbidden)
}
} else { } else {
handleNoStreamCache(bouncer, rw, req, remoteIP) err = handleNoStreamCache(bouncer, remoteIP)
response(err != nil, bouncer, rw, req)
} }
} }
@@ -223,6 +217,14 @@ type Stream struct {
New []Decision `json:"new"` New []Decision `json:"new"`
} }
func response(isValid bool, bouncer *Bouncer, rw http.ResponseWriter, req *http.Request) {
if isValid {
rw.WriteHeader(http.StatusForbidden)
} else {
bouncer.next.ServeHTTP(rw, req)
}
}
func contains(source []string, target string) bool { func contains(source []string, target string) bool {
for _, item := range source { for _, item := range source {
if item == target { if item == target {
@@ -250,7 +252,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 +262,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.Error(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.Error(fmt.Sprintf("handleNoStreamCache:parseBody: %w", 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.Error(fmt.Sprintf("handleNoStreamCache:parseDuration %w", 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) {
@@ -312,19 +308,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.Error(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.Error(fmt.Sprintf("handleStreamCache:parsingBody %w", 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 +332,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) {
@@ -352,7 +348,7 @@ func crowdsecQuery(bouncer *Bouncer, stringURL string) ([]byte, error) {
} }
defer func() { defer func() {
if err = res.Body.Close(); err != nil { if err = res.Body.Close(); err != nil {
logger.Error(fmt.Sprintf("crowdsecQuery:closeBody %w", err)) logger.Error(fmt.Sprintf("crowdsecQuery:closeBody %s", err.Error()))
} }
}() }()
body, err := io.ReadAll(res.Body) body, err := io.ReadAll(res.Body)
@@ -14,9 +14,9 @@ services:
- "--providers.docker.exposedbydefault=false" - "--providers.docker.exposedbydefault=false"
- "--entrypoints.web.address=:80" - "--entrypoints.web.address=:80"
- "--experimental.plugins.bouncer.modulename=github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin" # - "--experimental.plugins.bouncer.modulename=github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin"
- "--experimental.plugins.bouncer.version=v1.1.3" # - "--experimental.plugins.bouncer.version=v1.1.3"
# - "--experimental.localplugins.bouncer.modulename=github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin" - "--experimental.localplugins.bouncer.modulename=github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin"
volumes: volumes:
- /var/run/docker.sock:/var/run/docker.sock:ro - /var/run/docker.sock:/var/run/docker.sock:ro
- logs-redis:/var/log/traefik - logs-redis:/var/log/traefik