Compare commits

...
17 Commits
Author SHA1 Message Date
mathieuHa 24c7801dac ⬆️ Bump to go 1.19 (#37)
* ⬆️ Bump to go 1.19

* ⬆️ Remove depreciated function and replace io/utils calls
2022-11-19 20:40:03 +01:00
mathieuHa be150e8aca 📝 Update README fix wrong version (#35)
Change version 1.10 to 1.1.0 on the change that deduplicated the cache
2022-11-19 19:37:23 +01:00
mathieuHaandMax Lerebourg 0a54f7b09f Update CI following latest plugindemo version (#34)
* Update CI following latest plugindemo version

* disable some linter

* Disable some linter

* Fix linter in redis, add error logger

* 🚨 Fix lint errors in private packages

* 🚨 Update lints in package and core code

* 🚨 exclude unfixable lint errors

* 🚨 update lint at package level and ignore necessary global variable

* 🍱 fix lint qnd merge

* 🚨 Fix Lint for string

* 🚨 Fix lint for bouncer

* 🚨 Fix error

* Fix weird linter error

Co-authored-by: Max Lerebourg <maxlerebourg@gmail.com>
2022-11-19 19:35:01 +01:00
maxlerebourg 2d0bea8eec Merge pull request #32 from maxlerebourg/28-bypassing-crowdsec-for-local-ips
28 bypassing crowdsec for local ips
2022-11-19 15:47:30 +01:00
MathieuHa bb714b5dfd Change debug to info log 2022-11-19 15:17:56 +01:00
MathieuHa bba5620187 Update all documentation and exemple 2022-11-19 13:46:15 +01:00
MathieuHa 73374ccefe Rework variables name, and logic to check IPs with strategy 2022-11-19 13:27:51 +01:00
MathieuHa 6b7f8655ac Remove Fixme In Doc 2022-11-19 12:28:50 +01:00
MathieuHa c9a3f1f8a8 Remove duration for remove csli crowdsec 2022-11-19 12:26:49 +01:00
MathieuHa 490e5e934f Merge remote-tracking branch 'origin' into 28-bypassing-crowdsec-for-local-ips 2022-11-19 12:22:56 +01:00
MathieuHa 5b0d4b533f Add logic return to bypass when IP is unfiltered, update documentation 2022-11-19 12:16:20 +01:00
Max Lerebourg 400de0d552 add redis timeout 2022-11-18 20:36:42 +01:00
MathieuHa 3742b6b540 Fix nil trusted checker 2022-11-17 09:34:16 +01:00
MathieuHa d2f0a9416d Add check to load param 2022-11-17 09:21:57 +01:00
MathieuHa c1f2131bc2 Add doc for trusted IP 2022-11-16 23:19:57 +01:00
MathieuHa 3f70e2b256 First implementation of the bypass feature for trusted IP 2022-11-16 23:17:41 +01:00
Max Lerebourg bf76ca9ef5 🍱 change logging 2022-11-15 19:50:22 +01:00
22 changed files with 445 additions and 735 deletions
+1 -1
View File
@@ -11,7 +11,7 @@ jobs:
strategy: strategy:
matrix: matrix:
go-version: [ 1.17, 1.x ] go-version: [ 1.19, 1.x ]
os: [ubuntu-latest, macos-latest, windows-latest] os: [ubuntu-latest, macos-latest, windows-latest]
steps: steps:
+3 -3
View File
@@ -12,9 +12,9 @@ jobs:
name: Main Process name: Main Process
runs-on: ubuntu-latest runs-on: ubuntu-latest
env: env:
GO_VERSION: 1.17 GO_VERSION: 1.19
GOLANGCI_LINT_VERSION: v1.46.2 GOLANGCI_LINT_VERSION: v1.50.0
YAEGI_VERSION: v0.13.0 YAEGI_VERSION: v0.14.2
CGO_ENABLED: 0 CGO_ENABLED: 0
defaults: defaults:
run: run:
+14 -2
View File
@@ -29,11 +29,20 @@ linters-settings:
linters: linters:
enable-all: true enable-all: true
disable: disable:
- deadcode # deprecated
- exhaustivestruct # deprecated
- golint # deprecated
- ifshort # deprecated
- interfacer # deprecated - interfacer # deprecated
- maligned # deprecated - maligned # deprecated
- nosnakecase # deprecated
- scopelint # deprecated - scopelint # deprecated
- golint # deprecated - scopelint # deprecated
- exhaustivestruct # deprecated - structcheck # deprecated
- varcheck # deprecated
- sqlclosecheck # not relevant (SQL)
- rowserrcheck # not relevant (SQL)
- execinquery # not relevant (SQL)
- cyclop # duplicate of gocyclo - cyclop # duplicate of gocyclo
- bodyclose # Too many false positives: https://github.com/timakin/bodyclose/issues/30 - bodyclose # Too many false positives: https://github.com/timakin/bodyclose/issues/30
- dupl - dupl
@@ -52,6 +61,9 @@ linters:
- gomnd - gomnd
- forbidigo - forbidigo
- varnamelen - varnamelen
- wastedassign # is disabled because of generics
- gofumpt
- gci
issues: issues:
exclude-use-default: false exclude-use-default: false
+14 -1
View File
@@ -31,6 +31,9 @@ run_behindproxy:
run_cacheredis: run_cacheredis:
docker-compose -f exemples/redis-cache/docker-compose.redis.yml up -d --remove-orphans docker-compose -f exemples/redis-cache/docker-compose.redis.yml up -d --remove-orphans
run_trustedips:
docker-compose -f exemples/trusted-ips/docker-compose.trusted.yml up -d --remove-orphans
run: run:
docker-compose -f docker-compose.yml up -d --remove-orphans docker-compose -f docker-compose.yml up -d --remove-orphans
@@ -43,6 +46,15 @@ restart_local:
restart: restart:
docker-compose -f docker-compose.yml restart docker-compose -f docker-compose.yml restart
restart_behindproxy:
docker-compose -f exemples/behind-proxy/docker-compose.cloudflare.yml restart
restart_cacheredis:
docker-compose -f exemples/redis-cache/docker-compose.redis.yml restart
restart_trustedips:
docker-compose -f exemples/trusted-ips/docker-compose.trusted.yml restart
show_logs: show_logs:
docker-compose -f docker-compose.yml restart docker-compose -f docker-compose.yml restart
@@ -54,7 +66,8 @@ show_dev_logs:
clean_all_docker: clean_all_docker:
docker-compose -f exemples/behind-proxy/docker-compose.cloudflare.yml down --remove-orphans docker-compose -f exemples/behind-proxy/docker-compose.cloudflare.yml down --remove-orphans
docker-compose -f exemples/behind-proxy/docker-compose.redis.yml down --remove-orphans docker-compose -f exemples/redis-cache/docker-compose.redis.yml down --remove-orphans
docker-compose -f exemples/trusted-ips/docker-compose.trusted.yml down --remove-orphans
docker-compose -f docker-compose.local.yml down --remove-orphans docker-compose -f docker-compose.local.yml down --remove-orphans
docker-compose -f docker-compose.yml down --remove-orphans docker-compose -f docker-compose.yml down --remove-orphans
+50 -5
View File
@@ -40,7 +40,7 @@ make run
### Note ### Note
**/!\ Since Release 1.10, cache is no longer duplicated but shared by all services** **/!\ Since Release 1.1.0, cache is no longer duplicated but shared by all services**
*This lowers the overhead of the cache in memory and the numbers of cache to fetch it from crowdsec in situation with many services* *This lowers the overhead of the cache in memory and the numbers of cache to fetch it from crowdsec in situation with many services*
@@ -89,7 +89,10 @@ make run
- string - string
- default: "redis:6379" - default: "redis:6379"
- hostname and port for the redis service - hostname and port for the redis service
- ClientTrustedIPs
- string
- default: []
- List of client IPs we trust, they will bypass any more check to the bouncer or cache (usefull for LAN or VPN IP)
### Configuration ### Configuration
@@ -139,6 +142,8 @@ http:
forwardedHeadersTrustedIPs: forwardedHeadersTrustedIPs:
- 10.0.10.23/32 - 10.0.10.23/32
- 10.0.20.0/24 - 10.0.20.0/24
clientTrustedIPs:
- 192.168.1.0/24
forwardedHeadersCustomName: X-Custom-Header forwardedHeadersCustomName: X-Custom-Header
redisCacheEnabled: false redisCacheEnabled: false
redisCacheHost: "redis:6379" redisCacheHost: "redis:6379"
@@ -178,7 +183,7 @@ docker-compose up -d
```bash ```bash
docker-compose up -d crowdsec docker-compose up -d crowdsec
docker exec crowdsec cscli decisions add --ip 10.0.0.10 # this will be effective 4h docker exec crowdsec cscli decisions add --ip 10.0.0.10 -d 10m # this will be effective 10min
docker exec crowdsec cscli decisions remove --ip 10.0.0.10 docker exec crowdsec cscli decisions remove --ip 10.0.0.10
``` ```
@@ -224,6 +229,7 @@ You need to configure your Traefik to trust Forwarded headers by your front prox
In the exemple we use another instance of traefik with the container named cloudflare to simulate a front proxy In the exemple we use another instance of traefik with the container named cloudflare to simulate a front proxy
The "internal" Traefik instance is configured to trust the cloudflare forward headers The "internal" Traefik instance is configured to trust the cloudflare forward headers
This helps Traefik choose the right IP of the client: see https://doc.traefik.io/traefik/routing/entrypoints/#forwarded-headers
```yaml ```yaml
- "--entrypoints.web.forwardedheaders.trustedips=172.21.0.5" - "--entrypoints.web.forwardedheaders.trustedips=172.21.0.5"
``` ```
@@ -233,7 +239,7 @@ We configure the middleware to trust as well the IP:
- "traefik.http.middlewares.crowdsec1.plugin.bouncer.forwardedheaderstrustedips=172.21.0.5" - "traefik.http.middlewares.crowdsec1.plugin.bouncer.forwardedheaderstrustedips=172.21.0.5"
``` ```
To run the environnement run: To play the demo environnement run:
```bash ```bash
make run_behindproxy make run_behindproxy
``` ```
@@ -246,11 +252,50 @@ The plugin must be configured to connect to a redis instance
``` ```
Here **redis** is the hostname of a container located in the same network as Traefik and **6379** the default port of redis Here **redis** is the hostname of a container located in the same network as Traefik and **6379** the default port of redis
To run the demo environnement run: To play the demo environnement run:
```bash ```bash
make run_cacheredis make run_cacheredis
``` ```
3. Using Trusted IP (ex: LAN OR VPN) that won't get filtered by crowdsec
You need to configure your Traefik to trust Forwarded headers by your front proxy
In the exemple we use a whoami container protected by crowdsec, and we ban or IP before allowing using TrustedIPs
If you are using another proxy in front, you need to add it's IP in the trusted IP for the forwarded headers.
This helps Traefik choose the right IP of the client: see https://doc.traefik.io/traefik/routing/entrypoints/#forwarded-headers
The "internal" Traefik instance is configured to trust the forward headers
```yaml
- "--entrypoints.web.forwardedheaders.trustedips=172.21.0.5"
```
We configure the middleware to trust as well the IP of the intermediate proxy if needed:
```yaml
- "traefik.http.middlewares.crowdsec.plugin.bouncer.forwardedheaderstrustedips=172.21.0.5"
```
Add your IP to the ban list
```bash
docker exec crowdsec cscli decisions add --ip 10.0.10.30 -d 10m
```
You should get a 403 on http://localhost/foo
> Replace *10.0.10.30* by your IP
Add the IPs that will not be filtered by the plugin
```yaml
- "traefik.http.middlewares.crowdsec.plugin.bouncer.clientTrustedips=10.0.10.30/32"
```
> Replace *10.0.10.30/32* by your IP or IP range, so it's not getting checked against ban cache of crowdsec
You should get a 200 on http://localhost/foo even if you are on the ban cache
To play the demo environnement run:
```bash
make run_trustedips
```
### About ### About
Me and [mathieuHa](https://github.com/mathieuHa) have been using traefik since 2020 at [Primadviz](https://primadviz.com). Me and [mathieuHa](https://github.com/mathieuHa) have been using traefik since 2020 at [Primadviz](https://primadviz.com).
+66 -27
View File
@@ -1,4 +1,6 @@
package crowdsec_bouncer_traefik_plugin // Package crowdsec_bouncer_traefik_plugin implements a middleware that communicates with crowdsec.
// It can cache results to filesystem or redis, or even ask crowdsec for every requests.
package crowdsec_bouncer_traefik_plugin //nolint:revive,stylecheck
import ( import (
"bytes" "bytes"
@@ -6,7 +8,6 @@ import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"io" "io"
"io/ioutil"
"net/http" "net/http"
"net/url" "net/url"
"text/template" "text/template"
@@ -15,6 +16,7 @@ import (
cache "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/cache" cache "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/cache"
ip "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/ip" ip "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/ip"
logger "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/logger" logger "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/logger"
simpleredis "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/simpleredis"
) )
const ( const (
@@ -27,6 +29,7 @@ const (
cacheTimeoutKey = "updated" cacheTimeoutKey = "updated"
) )
//nolint:gochecknoglobals
var ( var (
crowdsecStreamHealthy = false crowdsecStreamHealthy = false
ticker chan bool ticker chan bool
@@ -40,10 +43,11 @@ type Config struct {
CrowdsecLapiScheme string `json:"crowdsecLapiScheme,omitempty"` CrowdsecLapiScheme string `json:"crowdsecLapiScheme,omitempty"`
CrowdsecLapiHost string `json:"crowdsecLapiHost,omitempty"` CrowdsecLapiHost string `json:"crowdsecLapiHost,omitempty"`
CrowdsecLapiKey string `json:"crowdsecLapiKey,omitempty"` CrowdsecLapiKey string `json:"crowdsecLapiKey,omitempty"`
ForwardedHeadersCustomName string `json:"forwardedheaderscustomheader,omitempty"`
UpdateIntervalSeconds int64 `json:"updateIntervalSeconds,omitempty"` UpdateIntervalSeconds int64 `json:"updateIntervalSeconds,omitempty"`
DefaultDecisionSeconds int64 `json:"defaultDecisionSeconds,omitempty"` DefaultDecisionSeconds int64 `json:"defaultDecisionSeconds,omitempty"`
ForwardedHeadersCustomName string `json:"forwardedheaderscustomheader,omitempty"`
ForwardedHeadersTrustedIPs []string `json:"forwardedHeadersTrustedIps,omitempty"` ForwardedHeadersTrustedIPs []string `json:"forwardedHeadersTrustedIps,omitempty"`
ClientTrustedIPs []string `json:"clientTrustedIps,omitempty"`
RedisCacheEnabled bool `json:"redisCacheEnabled,omitempty"` RedisCacheEnabled bool `json:"redisCacheEnabled,omitempty"`
RedisCacheHost string `json:"redisCacheHost,omitempty"` RedisCacheHost string `json:"redisCacheHost,omitempty"`
} }
@@ -59,6 +63,7 @@ func CreateConfig() *Config {
CrowdsecLapiKey: "", CrowdsecLapiKey: "",
UpdateIntervalSeconds: 60, UpdateIntervalSeconds: 60,
DefaultDecisionSeconds: 60, DefaultDecisionSeconds: 60,
ClientTrustedIPs: []string{},
ForwardedHeadersTrustedIPs: []string{}, ForwardedHeadersTrustedIPs: []string{},
ForwardedHeadersCustomName: "X-Forwarded-For", ForwardedHeadersCustomName: "X-Forwarded-For",
RedisCacheEnabled: false, RedisCacheEnabled: false,
@@ -80,7 +85,8 @@ type Bouncer struct {
updateInterval int64 updateInterval int64
defaultDecisionTimeout int64 defaultDecisionTimeout int64
customHeader string customHeader string
poolStrategy *ip.PoolStrategy clientPoolStrategy *ip.PoolStrategy
serverPoolStrategy *ip.PoolStrategy
client *http.Client client *http.Client
} }
@@ -89,11 +95,12 @@ func New(ctx context.Context, next http.Handler, config *Config, name string) (h
logger.Init(config.LogLevel) logger.Init(config.LogLevel)
err := validateParams(config) err := validateParams(config)
if err != nil { if err != nil {
logger.Info(fmt.Sprintf("%w", err)) logger.Info(err.Error())
return nil, err return nil, err
} }
checker, _ := ip.NewChecker(config.ForwardedHeadersTrustedIPs) serverChecker, _ := ip.NewChecker(config.ForwardedHeadersTrustedIPs)
clientChecker, _ := ip.NewChecker(config.ClientTrustedIPs)
bouncer := &Bouncer{ bouncer := &Bouncer{
next: next, next: next,
@@ -108,8 +115,11 @@ func New(ctx context.Context, next http.Handler, config *Config, name string) (h
updateInterval: config.UpdateIntervalSeconds, updateInterval: config.UpdateIntervalSeconds,
customHeader: config.ForwardedHeadersCustomName, customHeader: config.ForwardedHeadersCustomName,
defaultDecisionTimeout: config.DefaultDecisionSeconds, defaultDecisionTimeout: config.DefaultDecisionSeconds,
poolStrategy: &ip.PoolStrategy{ serverPoolStrategy: &ip.PoolStrategy{
Checker: checker, Checker: serverChecker,
},
clientPoolStrategy: &ip.PoolStrategy{
Checker: clientChecker,
}, },
client: &http.Client{ client: &http.Client{
Transport: &http.Transport{ Transport: &http.Transport{
@@ -133,24 +143,44 @@ func New(ctx context.Context, next http.Handler, config *Config, name string) (h
} }
// ServeHTTP principal function of plugin. // ServeHTTP principal function of plugin.
//
//nolint:nestif
func (bouncer *Bouncer) ServeHTTP(rw http.ResponseWriter, req *http.Request) { func (bouncer *Bouncer) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
if !bouncer.enabled { if !bouncer.enabled {
bouncer.next.ServeHTTP(rw, req) bouncer.next.ServeHTTP(rw, req)
return return
} }
remoteHost, err := ip.GetRemoteIP(req, bouncer.poolStrategy, bouncer.customHeader) // Here we check for the trusted IPs in the customHeader
remoteIP, err := ip.GetRemoteIP(req, bouncer.serverPoolStrategy, bouncer.customHeader)
if err != nil { if err != nil {
logger.Info(fmt.Sprintf("%w", err)) logger.Error(fmt.Sprintf("ServeHTTP ip:%s %s", remoteIP, err.Error()))
rw.WriteHeader(http.StatusForbidden)
return
}
trusted, err := bouncer.clientPoolStrategy.Checker.Contains(remoteIP)
if err != nil {
logger.Info(err.Error())
return
}
// if our IP is in the trusted list we bypass the next checks
logger.Debug(fmt.Sprintf("ServeHTTP ip:%s isTrusted:%v", remoteIP, trusted))
if trusted {
bouncer.next.ServeHTTP(rw, req) bouncer.next.ServeHTTP(rw, req)
return return
} }
logger.Debug(fmt.Sprintf("ServeHTTP ip:%v", remoteHost))
// TODO This should be simplified
healthy := crowdsecStreamHealthy
if bouncer.crowdsecMode != noneMode { if bouncer.crowdsecMode != noneMode {
isBanned, err := cache.GetDecision(remoteHost) isBanned, err := cache.GetDecision(remoteIP)
if err == nil { if err != nil {
logger.Debug(fmt.Sprintf("ServeHTTP cacheHit isBanned:%v", isBanned)) logger.Error(err.Error())
if err.Error() == simpleredis.RedisUnreachable {
healthy = false
}
} else {
logger.Debug(fmt.Sprintf("ServeHTTP ip:%s cache:hit isBanned:%v", remoteIP, isBanned))
if isBanned { if isBanned {
rw.WriteHeader(http.StatusForbidden) rw.WriteHeader(http.StatusForbidden)
} else { } else {
@@ -162,13 +192,13 @@ 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 crowdsecStreamHealthy { if healthy {
bouncer.next.ServeHTTP(rw, req) bouncer.next.ServeHTTP(rw, req)
} else { } else {
rw.WriteHeader(http.StatusForbidden) rw.WriteHeader(http.StatusForbidden)
} }
} else { } else {
handleNoStreamCache(bouncer, rw, req, remoteHost) handleNoStreamCache(bouncer, rw, req, remoteIP)
} }
} }
@@ -220,23 +250,23 @@ 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, remoteHost string) { func handleNoStreamCache(bouncer *Bouncer, rw http.ResponseWriter, req *http.Request, remoteIP string) {
routeURL := url.URL{ routeURL := url.URL{
Scheme: bouncer.crowdsecScheme, Scheme: bouncer.crowdsecScheme,
Host: bouncer.crowdsecHost, Host: bouncer.crowdsecHost,
Path: crowdsecLapiRoute, Path: crowdsecLapiRoute,
RawQuery: fmt.Sprintf("ip=%v&banned=true", remoteHost), RawQuery: fmt.Sprintf("ip=%v&banned=true", remoteIP),
} }
body, err := crowdsecQuery(bouncer, routeURL.String()) body, err := crowdsecQuery(bouncer, routeURL.String())
if err != nil { if err != nil {
logger.Info(fmt.Sprintf("%w", err)) logger.Info(err.Error())
rw.WriteHeader(http.StatusForbidden) rw.WriteHeader(http.StatusForbidden)
return return
} }
if bytes.Equal(body, []byte("null")) { if bytes.Equal(body, []byte("null")) {
if bouncer.crowdsecMode == liveMode { if bouncer.crowdsecMode == liveMode {
cache.SetDecision(remoteHost, false, bouncer.defaultDecisionTimeout) cache.SetDecision(remoteIP, false, bouncer.defaultDecisionTimeout)
} }
bouncer.next.ServeHTTP(rw, req) bouncer.next.ServeHTTP(rw, req)
return return
@@ -251,7 +281,7 @@ func handleNoStreamCache(bouncer *Bouncer, rw http.ResponseWriter, req *http.Req
} }
if len(decisions) == 0 { if len(decisions) == 0 {
if bouncer.crowdsecMode == liveMode { if bouncer.crowdsecMode == liveMode {
cache.SetDecision(remoteHost, false, bouncer.defaultDecisionTimeout) cache.SetDecision(remoteIP, false, bouncer.defaultDecisionTimeout)
} }
bouncer.next.ServeHTTP(rw, req) bouncer.next.ServeHTTP(rw, req)
return return
@@ -263,7 +293,7 @@ func handleNoStreamCache(bouncer *Bouncer, rw http.ResponseWriter, req *http.Req
return return
} }
if bouncer.crowdsecMode == liveMode { if bouncer.crowdsecMode == liveMode {
cache.SetDecision(remoteHost, true, int64(duration.Seconds())) cache.SetDecision(remoteIP, true, int64(duration.Seconds()))
} }
} }
@@ -286,7 +316,7 @@ func handleStreamCache(bouncer *Bouncer) {
} }
body, err := crowdsecQuery(bouncer, streamRouteURL.String()) body, err := crowdsecQuery(bouncer, streamRouteURL.String())
if err != nil { if err != nil {
logger.Info(fmt.Sprintf("%w", err)) logger.Info(err.Error())
crowdsecStreamHealthy = false crowdsecStreamHealthy = false
return return
} }
@@ -315,7 +345,7 @@ 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: %s", stringURL, err) return nil, fmt.Errorf("error while fetching %v: %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("error while fetching %v, status code: %d", stringURL, res.StatusCode)
@@ -323,12 +353,13 @@ func crowdsecQuery(bouncer *Bouncer, stringURL string) ([]byte, error) {
defer func(body io.ReadCloser) { defer func(body io.ReadCloser) {
err = body.Close() err = body.Close()
if err != nil { if err != nil {
logger.Info(fmt.Sprintf("failed to close body reader: %s", err)) logger.Error(fmt.Sprintf("failed to close body reader: %s", err.Error()))
} }
}(res.Body) }(res.Body)
body, err := ioutil.ReadAll(res.Body) body, err := io.ReadAll(res.Body)
if err != nil { if err != nil {
return nil, fmt.Errorf("error while reading body: %s", err) return nil, fmt.Errorf("error while reading body: %w", err)
} }
return body, nil return body, nil
} }
@@ -376,6 +407,14 @@ func validateParams(config *Config) error {
} else { } else {
logger.Debug("No IP provided for ForwardedHeadersTrustedIPs") logger.Debug("No IP provided for ForwardedHeadersTrustedIPs")
} }
if len(config.ClientTrustedIPs) > 0 {
_, err = ip.NewChecker(config.ClientTrustedIPs)
if err != nil {
return fmt.Errorf("TrustedIPs must be a list of IP/CIDR :%w", err)
}
} else {
logger.Debug("No IP provided for TrustedIPs")
}
return nil return nil
} }
+1 -1
View File
@@ -1,4 +1,4 @@
package crowdsec_bouncer_traefik_plugin package crowdsec_bouncer_traefik_plugin //nolint:revive,stylecheck
import ( import (
"context" "context"
+20 -16
View File
@@ -4,8 +4,9 @@ services:
traefik: traefik:
image: "traefik:v2.9.4" image: "traefik:v2.9.4"
container_name: "traefik" container_name: "traefik"
restart: unless-stopped
command: command:
# - "--log.level=DEBUG" - "--log.level=DEBUG"
- "--accesslog" - "--accesslog"
- "--accesslog.filepath=/var/log/traefik/access.log" - "--accesslog.filepath=/var/log/traefik/access.log"
- "--api.insecure=true" - "--api.insecure=true"
@@ -24,33 +25,36 @@ services:
depends_on: depends_on:
- crowdsec - crowdsec
whoami1: whoami-foo:
image: traefik/whoami image: traefik/whoami
container_name: "simple-service1" container_name: "simple-service-foo"
restart: unless-stopped
labels: labels:
- "traefik.enable=true" - "traefik.enable=true"
- "traefik.http.routers.router1.rule=Host(`localhost`) && Path(`/foo`)" - "traefik.http.routers.router-foo.rule=Path(`/foo`)"
- "traefik.http.routers.router1.entrypoints=web" - "traefik.http.routers.router-foo.entrypoints=web"
- "traefik.http.routers.router1.middlewares=crowdsec1@docker" - "traefik.http.routers.router-foo.middlewares=crowdsec-foo@docker"
- "traefik.http.services.service1.loadbalancer.server.port=80" - "traefik.http.services.service-foo.loadbalancer.server.port=80"
- "traefik.http.middlewares.crowdsec1.plugin.bouncer.enabled=true" - "traefik.http.middlewares.crowdsec-foo.plugin.bouncer.enabled=true"
- "traefik.http.middlewares.crowdsec1.plugin.bouncer.crowdseclapikey=40796d93c2958f9e58345514e67740e5" - "traefik.http.middlewares.crowdsec-foo.plugin.bouncer.crowdseclapikey=40796d93c2958f9e58345514e67740e5"
whoami2: whoami2:
image: traefik/whoami image: traefik/whoami
container_name: "simple-service2" container_name: "simple-service-bar"
restart: unless-stopped
labels: labels:
- "traefik.enable=true" - "traefik.enable=true"
- "traefik.http.routers.router2.rule=Host(`localhost`) && Path(`/bar`)" - "traefik.http.routers.router-bar.rule=Path(`/bar`)"
- "traefik.http.routers.router2.entrypoints=web" - "traefik.http.routers.router-bar.entrypoints=web"
- "traefik.http.routers.router2.middlewares=crowdsec2@docker" - "traefik.http.routers.router-bar.middlewares=crowdsec-bar@docker"
- "traefik.http.services.service2.loadbalancer.server.port=80" - "traefik.http.services.service-bar.loadbalancer.server.port=80"
- "traefik.http.middlewares.crowdsec2.plugin.bouncer.enabled=true" - "traefik.http.middlewares.crowdsec-bar.plugin.bouncer.enabled=true"
- "traefik.http.middlewares.crowdsec2.plugin.bouncer.crowdseclapikey=44c36dac5c4140af9f06f397508e82c7" - "traefik.http.middlewares.crowdsec-bar.plugin.bouncer.crowdseclapikey=44c36dac5c4140af9f06f397508e82c7"
crowdsec: crowdsec:
image: crowdsecurity/crowdsec:v1.4.1 image: crowdsecurity/crowdsec:v1.4.1
container_name: "crowdsec" container_name: "crowdsec"
restart: unless-stopped
environment: environment:
COLLECTIONS: crowdsecurity/traefik COLLECTIONS: crowdsecurity/traefik
CUSTOM_HOSTNAME: crowdsec CUSTOM_HOSTNAME: crowdsec
+21 -17
View File
@@ -4,6 +4,7 @@ services:
traefik: traefik:
image: "traefik:v2.9.4" image: "traefik:v2.9.4"
container_name: "traefik" container_name: "traefik"
restart: unless-stopped
command: command:
- "--accesslog" - "--accesslog"
- "--accesslog.filepath=/var/log/traefik/access.log" - "--accesslog.filepath=/var/log/traefik/access.log"
@@ -13,7 +14,7 @@ services:
- "--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.0" - "--experimental.plugins.bouncer.version=v1.1.3"
volumes: volumes:
- "/var/run/docker.sock:/var/run/docker.sock:ro" - "/var/run/docker.sock:/var/run/docker.sock:ro"
- "logs:/var/log/traefik" - "logs:/var/log/traefik"
@@ -25,43 +26,46 @@ services:
whoami1: whoami1:
image: traefik/whoami image: traefik/whoami
container_name: "simple-service-1" container_name: "simple-service-foo"
restart: unless-stopped
labels: labels:
- "traefik.enable=true" - "traefik.enable=true"
# Definition of the router # Definition of the router
- "traefik.http.routers.router1.rule=Path(`/foo`)" - "traefik.http.routers.router-foo.rule=Path(`/foo`)"
- "traefik.http.routers.router1.entrypoints=web" - "traefik.http.routers.router-foo.entrypoints=web"
- "traefik.http.routers.router1.middlewares=crowdsec2@docker" - "traefik.http.routers.router-foo.middlewares=crowdsec-foo@docker"
# Definition of the service # Definition of the service
- "traefik.http.services.service1.loadbalancer.server.port=80" - "traefik.http.services.service-foo.loadbalancer.server.port=80"
# Definition of the middleware # Definition of the middleware
- "traefik.http.middlewares.crowdsec1.plugin.bouncer.enabled=true" - "traefik.http.middlewares.crowdsec-foo.plugin.bouncer.enabled=true"
# crowdseclapikey must be unique to the middleware attached to the service # crowdseclapikey must be unique to the middleware attached to the service
- "traefik.http.middlewares.crowdsec1.plugin.bouncer.crowdseclapikey=FIXME-LAPI-KEY-1" - "traefik.http.middlewares.crowdsec-foo.plugin.bouncer.crowdseclapikey=FIXME-LAPI-KEY-1"
# forwardedheaderstrustedips should be the IP of the proxy that is in front of traefik (if any) # forwardedheaderstrustedips should be the IP of the proxy that is in front of traefik (if any)
- "traefik.http.middlewares.crowdsec1.plugin.bouncer.forwardedheaderstrustedips=172.21.0.5" - "traefik.http.middlewares.crowdsec-foo.plugin.bouncer.forwardedheaderstrustedips=172.21.0.5"
whoami2: whoami2:
image: traefik/whoami image: traefik/whoami
container_name: "simple-service-2" container_name: "simple-service-bar"
restart: unless-stopped
labels: labels:
- "traefik.enable=true" - "traefik.enable=true"
# Definition of the router # Definition of the router
- "traefik.http.routers.router2.rule=Path(`/bar`)" - "traefik.http.routers.router-bar.rule=Path(`/bar`)"
- "traefik.http.routers.router2.entrypoints=web" - "traefik.http.routers.router-bar.entrypoints=web"
- "traefik.http.routers.router2.middlewares=crowdsec2@docker" - "traefik.http.routers.router-bar.middlewares=crowdsec-bar@docker"
# Definition of the service # Definition of the service
- "traefik.http.services.service2.loadbalancer.server.port=80" - "traefik.http.services.service-bar.loadbalancer.server.port=80"
# Definitin of the middleware # Definitin of the middleware
- "traefik.http.middlewares.crowdsec2.plugin.bouncer.enabled=true" - "traefik.http.middlewares.crowdsec-bar.plugin.bouncer.enabled=true"
# crowdseclapikey must be unique to the middleware attached to the service # crowdseclapikey must be unique to the middleware attached to the service
- "traefik.http.middlewares.crowdsec2.plugin.bouncer.crowdseclapikey=FIXME-LAPI-KEY-2" - "traefik.http.middlewares.crowdsec-bar.plugin.bouncer.crowdseclapikey=FIXME-LAPI-KEY-2"
# forwardedheaderstrustedips should be the IP of the proxy that is in front of traefik (if any) # forwardedheaderstrustedips should be the IP of the proxy that is in front of traefik (if any)
- "traefik.http.middlewares.crowdsec2.plugin.bouncer.forwardedheaderstrustedips=172.21.0.5" - "traefik.http.middlewares.crowdsec-bar.plugin.bouncer.forwardedheaderstrustedips=172.21.0.5"
crowdsec: crowdsec:
image: crowdsecurity/crowdsec:v1.4.1 image: crowdsecurity/crowdsec:v1.4.1
container_name: "crowdsec" container_name: "crowdsec"
restart: unless-stopped
environment: environment:
COLLECTIONS: crowdsecurity/traefik COLLECTIONS: crowdsecurity/traefik
CUSTOM_HOSTNAME: crowdsec CUSTOM_HOSTNAME: crowdsec
@@ -2,8 +2,9 @@ version: "3.8"
services: services:
cloudflare: cloudflare:
image: "traefik:v2.9.1" image: "traefik:v2.9.4"
container_name: "cloudflare" container_name: "cloudflare"
restart: unless-stopped
command: command:
# - "--log.level=DEBUG" # - "--log.level=DEBUG"
- "--accesslog" - "--accesslog"
@@ -21,8 +22,9 @@ services:
- 8080:8080 - 8080:8080
traefik: traefik:
image: "traefik:v2.9.1" image: "traefik:v2.9.4"
container_name: "traefik" container_name: "traefik"
restart: unless-stopped
command: command:
# - "--log.level=DEBUG" # - "--log.level=DEBUG"
- "--accesslog" - "--accesslog"
@@ -34,10 +36,10 @@ services:
- "--entrypoints.web.forwardedheaders.trustedips=172.21.0.5" - "--entrypoints.web.forwardedheaders.trustedips=172.21.0.5"
- "--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.0" - "--experimental.plugins.bouncer.version=v1.1.3"
volumes: volumes:
- /var/run/docker.sock:/var/run/docker.sock:ro - /var/run/docker.sock:/var/run/docker.sock:ro
- logs-dev:/var/log/traefik - logs-traefik:/var/log/traefik
ports: ports:
- 90:80 - 90:80
- 9080:8080 - 9080:8080
@@ -46,26 +48,49 @@ services:
whoami1: whoami1:
image: traefik/whoami image: traefik/whoami
container_name: "simple-service1" container_name: "simple-service-foo"
restart: unless-stopped
labels: labels:
- "traefik.enable=true" - "traefik.enable=true"
# Definition of the router # Definition of the router
- "traefik.http.routers.router1.rule=Path(`/foo`)" - "traefik.http.routers.router-foo.rule=Path(`/foo`)"
- "traefik.http.routers.router1.entrypoints=web" - "traefik.http.routers.router-foo.entrypoints=web"
- "traefik.http.routers.router1.middlewares=crowdsec1@docker" - "traefik.http.routers.router-foo.middlewares=crowdsec-foo@docker"
# Definition of the service # Definition of the service
- "traefik.http.services.service1.loadbalancer.server.port=80" - "traefik.http.services.service-foo.loadbalancer.server.port=80"
# Definitin of the middleware # Definitin of the middleware
- "traefik.http.middlewares.crowdsec1.plugin.bouncer.enabled=true" - "traefik.http.middlewares.crowdsec-foo.plugin.bouncer.enabled=true"
# crowdseclapikey must be uniq to the middleware attached to the service # crowdseclapikey must be uniq to the middleware attached to the service
- "traefik.http.middlewares.crowdsec1.plugin.bouncer.crowdseclapikey=40796d93c2958f9e58345514e67740e5" - "traefik.http.middlewares.crowdsec-foo.plugin.bouncer.crowdseclapikey=40796d93c2958f9e58345514e67740e5"
- "traefik.http.middlewares.crowdsec1.plugin.bouncer.crowdsecmode=live" - "traefik.http.middlewares.crowdsec-foo.plugin.bouncer.crowdsecmode=live"
- "traefik.http.middlewares.crowdsec1.plugin.bouncer.forwardedheaderstrustedips=172.21.0.5" - "traefik.http.middlewares.crowdsec-foo.plugin.bouncer.forwardedheaderstrustedips=172.21.0.5"
- "traefik.http.middlewares.crowdsec-foo.plugin.bouncer.loglevel=DEBUG"
whoami2:
image: traefik/whoami
container_name: "simple-service-bar"
restart: unless-stopped
labels:
- "traefik.enable=true"
# Definition of the router
- "traefik.http.routers.router-bar.rule=Path(`/bar`)"
- "traefik.http.routers.router-bar.entrypoints=web"
- "traefik.http.routers.router-bar.middlewares=crowdsec-bar@docker"
# Definition of the service
- "traefik.http.services.service-bar.loadbalancer.server.port=80"
# Definitin of the middleware
- "traefik.http.middlewares.crowdsec-bar.plugin.bouncer.enabled=true"
# crowdseclapikey must be uniq to the middleware attached to the service
- "traefik.http.middlewares.crowdsec-bar.plugin.bouncer.crowdseclapikey=44c36dac5c4140af9f06f397508e82c7"
- "traefik.http.middlewares.crowdsec-bar.plugin.bouncer.crowdsecmode=live"
- "traefik.http.middlewares.crowdsec-bar.plugin.bouncer.forwardedheaderstrustedips=172.21.0.5"
- "traefik.http.middlewares.crowdsec-bar.plugin.bouncer.loglevel=DEBUG"
crowdsec: crowdsec:
image: crowdsecurity/crowdsec:v1.4.1 image: crowdsecurity/crowdsec:v1.4.1
container_name: "crowdsec" container_name: "crowdsec"
restart: unless-stopped
environment: environment:
COLLECTIONS: crowdsecurity/traefik COLLECTIONS: crowdsecurity/traefik
CUSTOM_HOSTNAME: crowdsec CUSTOM_HOSTNAME: crowdsec
@@ -73,14 +98,14 @@ services:
BOUNCER_KEY_TRAEFIK_DEV_2: 44c36dac5c4140af9f06f397508e82c7 BOUNCER_KEY_TRAEFIK_DEV_2: 44c36dac5c4140af9f06f397508e82c7
volumes: volumes:
- ./acquis.yaml:/etc/crowdsec/acquis.yaml:ro - ./acquis.yaml:/etc/crowdsec/acquis.yaml:ro
- logs-dev:/var/log/traefik:ro - logs-cloudflare:/var/log/traefik:ro
- crowdsec-db-dev:/var/lib/crowdsec/data/ - crowdsec-db-cloudflare:/var/lib/crowdsec/data/
- crowdsec-config-dev:/etc/crowdsec/ - crowdsec-config-cloudflare:/etc/crowdsec/
labels: labels:
- "traefik.enable=false" - "traefik.enable=false"
volumes: volumes:
logs-dev: logs-traefik:
logs-cloudflare: logs-cloudflare:
crowdsec-db-dev: crowdsec-db-cloudflare:
crowdsec-config-dev: crowdsec-config-cloudflare:
+29 -24
View File
@@ -4,6 +4,7 @@ services:
traefik: traefik:
image: "traefik:v2.9.4" image: "traefik:v2.9.4"
container_name: "traefik" container_name: "traefik"
restart: unless-stopped
command: command:
# - "--log.level=DEBUG" # - "--log.level=DEBUG"
- "--accesslog" - "--accesslog"
@@ -13,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.0.9" - "--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
@@ -27,46 +28,49 @@ services:
- crowdsec - crowdsec
- redis - redis
whoami1: whoami-foo:
image: traefik/whoami image: traefik/whoami
container_name: "simple-service1" container_name: "simple-service-foo"
restart: unless-stopped
labels: labels:
- "traefik.enable=true" - "traefik.enable=true"
# Definition of the router # Definition of the router
- "traefik.http.routers.router1.rule=Path(`/foo`)" - "traefik.http.routers.router-foo.rule=Path(`/foo`)"
- "traefik.http.routers.router1.entrypoints=web" - "traefik.http.routers.router-foo.entrypoints=web"
- "traefik.http.routers.router1.middlewares=crowdsec1@docker" - "traefik.http.routers.router-foo.middlewares=crowdsec-foo@docker"
# Definition of the service # Definition of the service
- "traefik.http.services.service1.loadbalancer.server.port=80" - "traefik.http.services.service-foo.loadbalancer.server.port=80"
# Definition of the middleware # Definition of the middleware
- "traefik.http.middlewares.crowdsec1.plugin.bouncer.enabled=true" - "traefik.http.middlewares.crowdsec-foo.plugin.bouncer.enabled=true"
# crowdseclapikey must be uniq to the middleware attached to the service # crowdseclapikey must be uniq to the middleware attached to the service
- "traefik.http.middlewares.crowdsec1.plugin.bouncer.crowdseclapikey=40796d93c2958f9e58345514e67740e5" - "traefik.http.middlewares.crowdsec-foo.plugin.bouncer.crowdseclapikey=40796d93c2958f9e58345514e67740e5"
- "traefik.http.middlewares.crowdsec1.plugin.bouncer.rediscacheenabled=true" - "traefik.http.middlewares.crowdsec-foo.plugin.bouncer.rediscacheenabled=true"
- "traefik.http.middlewares.crowdsec-foo.plugin.bouncer.loglevel=DEBUG"
whoami2: whoami-bar:
image: 4206969/spiderfoot image: traefik/whoami
container_name: "simple-service2" container_name: "simple-service-bar"
volumes: restart: unless-stopped
- ./spiderfoot-data:/var/lib/spiderfoot
labels: labels:
- "traefik.enable=true" - "traefik.enable=true"
# Definition of the router # Definition of the router
- "traefik.http.routers.router2.rule=PathPrefix(`/`)" - "traefik.http.routers.router-bar.rule=Path(`/bar`)"
- "traefik.http.routers.router2.entrypoints=web" - "traefik.http.routers.router-bar.entrypoints=web"
- "traefik.http.routers.router2.middlewares=crowdsec1@docker" - "traefik.http.routers.router-bar.middlewares=crowdsec-bar@docker"
# Definition of the service # Definition of the service
- "traefik.http.services.service2.loadbalancer.server.port=5001" - "traefik.http.services.service-bar.loadbalancer.server.port=80"
# Definition of the middleware # Definition of the middleware
- "traefik.http.middlewares.crowdsec2.plugin.bouncer.enabled=true" - "traefik.http.middlewares.crowdsec-bar.plugin.bouncer.enabled=true"
# crowdseclapikey must be uniq to the middleware attached to the service # crowdseclapikey must be uniq to the middleware attached to the service
- "traefik.http.middlewares.crowdsec2.plugin.bouncer.crowdseclapikey=44c36dac5c4140af9f06f397508e82c7" - "traefik.http.middlewares.crowdsec-bar.plugin.bouncer.crowdseclapikey=44c36dac5c4140af9f06f397508e82c7"
- "traefik.http.middlewares.crowdsec2.plugin.bouncer.rediscacheenabled=true" - "traefik.http.middlewares.crowdsec-bar.plugin.bouncer.rediscacheenabled=true"
- "traefik.http.middlewares.crowdsec-bar.plugin.bouncer.loglevel=DEBUG"
crowdsec: crowdsec:
image: crowdsecurity/crowdsec:v1.4.1 image: crowdsecurity/crowdsec:v1.4.1
container_name: "crowdsec" container_name: "crowdsec"
restart: unless-stopped
environment: environment:
COLLECTIONS: crowdsecurity/traefik COLLECTIONS: crowdsecurity/traefik
CUSTOM_HOSTNAME: crowdsec CUSTOM_HOSTNAME: crowdsec
@@ -83,6 +87,7 @@ services:
redis: redis:
image: "redis:7.0.5-alpine" image: "redis:7.0.5-alpine"
container_name: "redis" container_name: "redis"
restart: unless-stopped
command: "redis-server --save 60 1" command: "redis-server --save 60 1"
volumes: volumes:
- redis-data:/data - redis-data:/data
+4
View File
@@ -0,0 +1,4 @@
filenames:
- /var/log/traefik/access.log
labels:
type: traefik
@@ -0,0 +1,91 @@
version: "3.8"
services:
traefik:
image: "traefik:v2.9.4"
container_name: "traefik"
restart: unless-stopped
command:
# - "--log.level=DEBUG"
- "--accesslog"
- "--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.1.3"
# - "--experimental.localplugins.bouncer.modulename=github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin"
volumes:
- /var/run/docker.sock:/var/run/docker.sock:ro
- logs-trustedips:/var/log/traefik
- ./../../:/plugins-local/src/github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin
ports:
- 80:80
- 8080:8080
depends_on:
- crowdsec
whoami1:
image: traefik/whoami
container_name: "simple-service-foo"
restart: unless-stopped
labels:
- "traefik.enable=true"
# Definition of the router
- "traefik.http.routers.router-foo.rule=Path(`/foo`)"
- "traefik.http.routers.router-foo.entrypoints=web"
- "traefik.http.routers.router-foo.middlewares=crowdsec-foo@docker"
# Definition of the service
- "traefik.http.services.service-foo.loadbalancer.server.port=80"
# Definition of the middleware
- "traefik.http.middlewares.crowdsec-foo.plugin.bouncer.enabled=true"
# crowdseclapikey must be uniq to the middleware attached to the service
- "traefik.http.middlewares.crowdsec-foo.plugin.bouncer.crowdseclapikey=40796d93c2958f9e58345514e67740e5"
# Replace 10.0.10.30/32 by your IP range which is "trusted"
- "traefik.http.middlewares.crowdsec-foo.plugin.bouncer.clienttrustedips=10.0.10.30/32"
- "traefik.http.middlewares.crowdsec-foo.plugin.bouncer.loglevel=DEBUG"
whoami2:
image: traefik/whoami
container_name: "simple-service-bar"
restart: unless-stopped
labels:
- "traefik.enable=true"
# Definition of the router
- "traefik.http.routers.router-bar.rule=Path(`/bar`)"
- "traefik.http.routers.router-bar.entrypoints=web"
- "traefik.http.routers.router-bar.middlewares=crowdsec-bar@docker"
# Definition of the service
- "traefik.http.services.service-bar.loadbalancer.server.port=80"
# Definition of the middleware
- "traefik.http.middlewares.crowdsec-bar.plugin.bouncer.enabled=true"
# crowdseclapikey must be uniq to the middleware attached to the service
- "traefik.http.middlewares.crowdsec-bar.plugin.bouncer.crowdseclapikey=44c36dac5c4140af9f06f397508e82c7"
# Replace 10.0.10.30/32 by your IP range which is "trusted"
- "traefik.http.middlewares.crowdsec-bar.plugin.bouncer.clienttrustedips=10.0.10.30/32"
- "traefik.http.middlewares.crowdsec-bar.plugin.bouncer.loglevel=DEBUG"
crowdsec:
image: crowdsecurity/crowdsec:v1.4.1
container_name: "crowdsec"
restart: unless-stopped
environment:
COLLECTIONS: crowdsecurity/traefik
CUSTOM_HOSTNAME: crowdsec
BOUNCER_KEY_TRAEFIK_DEV_1: 40796d93c2958f9e58345514e67740e5
BOUNCER_KEY_TRAEFIK_DEV_2: 44c36dac5c4140af9f06f397508e82c7
volumes:
- ./acquis.yaml:/etc/crowdsec/acquis.yaml:ro
- logs-trustedips:/var/log/traefik:ro
- crowdsec-db-trustedips:/var/lib/crowdsec/data/
- crowdsec-config-trustedips:/etc/crowdsec/
labels:
- "traefik.enable=false"
volumes:
logs-trustedips:
crowdsec-db-trustedips:
crowdsec-config-trustedips:
+1 -3
View File
@@ -1,7 +1,5 @@
module github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin module github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin
go 1.17 go 1.19
require github.com/leprosus/golang-ttl-map v1.1.7 require github.com/leprosus/golang-ttl-map v1.1.7
require github.com/tehnerd/goUtils v0.0.0-20150515130609-5a2d8fb2ded8 // indirect
-2
View File
@@ -1,4 +1,2 @@
github.com/leprosus/golang-ttl-map v1.1.7 h1:cF4AAFDDnJTFSV+/42sKLhmMluvLdRlCGS2UaifH6UM= github.com/leprosus/golang-ttl-map v1.1.7 h1:cF4AAFDDnJTFSV+/42sKLhmMluvLdRlCGS2UaifH6UM=
github.com/leprosus/golang-ttl-map v1.1.7/go.mod h1:4QWHJPeVBbrkhOhXdhCv9IEiyj/YzkO04/iexy4vSe0= github.com/leprosus/golang-ttl-map v1.1.7/go.mod h1:4QWHJPeVBbrkhOhXdhCv9IEiyj/YzkO04/iexy4vSe0=
github.com/tehnerd/goUtils v0.0.0-20150515130609-5a2d8fb2ded8 h1:/b777evAfRRdUJHasZLgQ/w8D/s1HtbaeDXsCVsV0B0=
github.com/tehnerd/goUtils v0.0.0-20150515130609-5a2d8fb2ded8/go.mod h1:UuuqaOb+pZOxJZtjF1mBWTo8HYa7HQCbNkwUEaG9uU0=
+25 -16
View File
@@ -1,3 +1,5 @@
// Package cache implements utility routines for manipulating cache.
// It supports currently local file and redis cache.
package cache package cache
import ( import (
@@ -6,7 +8,7 @@ import (
ttl_map "github.com/leprosus/golang-ttl-map" ttl_map "github.com/leprosus/golang-ttl-map"
logger "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/logger" logger "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/logger"
simpleredis "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/redis" simpleredis "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/simpleredis"
) )
const ( const (
@@ -14,12 +16,14 @@ const (
cacheNoBannedValue = "f" cacheNoBannedValue = "f"
) )
var cache = ttl_map.New() //nolint:gochecknoglobals
var redis simpleredis.SimpleRedis var (
cache = ttl_map.New()
redis simpleredis.SimpleRedis
redisEnabled = false
)
var redisEnabled = false // FileSystem Cache
// CLASSIC
func getDecisionLocalCache(clientIP string) (bool, error) { func getDecisionLocalCache(clientIP string) (bool, error) {
banned, isCached := cache.Get(clientIP) banned, isCached := cache.Get(clientIP)
@@ -27,7 +31,7 @@ func getDecisionLocalCache(clientIP string) (bool, error) {
if isCached && isValid && len(bannedString) > 0 { if isCached && isValid && len(bannedString) > 0 {
return bannedString == cacheBannedValue, nil return bannedString == cacheBannedValue, nil
} }
return false, fmt.Errorf("no cache data") return false, fmt.Errorf("cache:miss")
} }
func setDecisionLocalCache(clientIP string, value string, duration int64) { func setDecisionLocalCache(clientIP string, value string, duration int64) {
@@ -38,7 +42,7 @@ func deleteDecisionLocalCache(clientIP string) {
cache.Del(clientIP) cache.Del(clientIP)
} }
// REDIS // Redis Cache
func getDecisionRedisCache(clientIP string) (bool, error) { func getDecisionRedisCache(clientIP string) (bool, error) {
banned, err := redis.Get(clientIP) banned, err := redis.Get(clientIP)
@@ -46,18 +50,22 @@ func getDecisionRedisCache(clientIP string) (bool, error) {
if err == nil && len(bannedString) > 0 { if err == nil && len(bannedString) > 0 {
return bannedString == cacheBannedValue, nil return bannedString == cacheBannedValue, nil
} }
return false, fmt.Errorf("no cache data") return false, err
} }
func setDecisionRedisCache(clientIP string, value string, duration int64) { func setDecisionRedisCache(clientIP string, value string, duration int64) {
redis.Set(clientIP, []byte(value), duration) if err := redis.Set(clientIP, []byte(value), duration); err != nil {
logger.Error(fmt.Sprintf("cache:setDecisionRedisCache %s", err.Error()))
}
} }
func deleteDecisionRedisCache(clientIP string) { func deleteDecisionRedisCache(clientIP string) {
redis.Del(clientIP) if err := redis.Del(clientIP); err != nil {
logger.Error(fmt.Sprintf("cache:deleteDecisionRedisCache %s", err.Error()))
}
} }
// DeleteDecision delete decision in cache // DeleteDecision delete decision in cache.
func DeleteDecision(clientIP string) { func DeleteDecision(clientIP string) {
if redisEnabled { if redisEnabled {
deleteDecisionRedisCache(clientIP) deleteDecisionRedisCache(clientIP)
@@ -71,15 +79,15 @@ func DeleteDecision(clientIP string) {
func GetDecision(clientIP string) (bool, error) { func GetDecision(clientIP string) (bool, error) {
if redisEnabled { if redisEnabled {
return getDecisionRedisCache(clientIP) return getDecisionRedisCache(clientIP)
} else {
return getDecisionLocalCache(clientIP)
} }
return getDecisionLocalCache(clientIP)
} }
// SetDecision update the cache with the IP as key and the value banned / not banned.
func SetDecision(clientIP string, isBanned bool, duration int64) { func SetDecision(clientIP string, isBanned bool, duration int64) {
var value string var value string
if isBanned { if isBanned {
logger.Debug(fmt.Sprintf("%v banned", clientIP)) logger.Debug(fmt.Sprintf("cache:SetDecision ip:%v banned", clientIP))
value = cacheBannedValue value = cacheBannedValue
} else { } else {
value = cacheNoBannedValue value = cacheNoBannedValue
@@ -91,8 +99,9 @@ func SetDecision(clientIP string, isBanned bool, duration int64) {
} }
} }
// InitRedisClient loads variables.
func InitRedisClient(host string) { func InitRedisClient(host string) {
redisEnabled = true redisEnabled = true
redis.Init(host) redis.Init(host)
logger.Debug("Redis initialized") logger.Debug("cache:InitRedisClient redis:initialized")
} }
+6 -4
View File
@@ -1,3 +1,5 @@
// Package ip implements utility routines for to manipulates IP and CIDR.
// It allows to find on IP on a list, and find if an IP is part of a list of CIDR.
package ip package ip
import ( import (
@@ -6,6 +8,8 @@ import (
"net" "net"
"net/http" "net/http"
"strings" "strings"
"github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/logger"
) )
// CHECKER // CHECKER
@@ -18,15 +22,12 @@ type Checker struct {
// NewChecker builds a new Checker given a list of CIDR-Strings to trusted IPs. // NewChecker builds a new Checker given a list of CIDR-Strings to trusted IPs.
func NewChecker(trustedIPs []string) (*Checker, error) { func NewChecker(trustedIPs []string) (*Checker, error) {
if len(trustedIPs) == 0 {
return nil, errors.New("no trusted IPs provided")
}
checker := &Checker{} checker := &Checker{}
for _, ipMask := range trustedIPs { for _, ipMask := range trustedIPs {
if ipAddr := net.ParseIP(ipMask); ipAddr != nil { if ipAddr := net.ParseIP(ipMask); ipAddr != nil {
checker.authorizedIPs = append(checker.authorizedIPs, &ipAddr) checker.authorizedIPs = append(checker.authorizedIPs, &ipAddr)
logger.Debug(fmt.Sprintf("IP %v is trusted", ipAddr))
continue continue
} }
@@ -35,6 +36,7 @@ func NewChecker(trustedIPs []string) (*Checker, error) {
return nil, fmt.Errorf("parsing CIDR trusted IPs %s: %w", ipAddr, err) return nil, fmt.Errorf("parsing CIDR trusted IPs %s: %w", ipAddr, err)
} }
checker.authorizedIPsNet = append(checker.authorizedIPsNet, ipAddr) checker.authorizedIPsNet = append(checker.authorizedIPsNet, ipAddr)
logger.Debug(fmt.Sprintf("IP network %v is trusted", ipAddr))
} }
return checker, nil return checker, nil
+16 -12
View File
@@ -1,3 +1,5 @@
// Package logger implements utility routines to write to stdout and stderr.
// It supports debug, info and error level
package logger package logger
import ( import (
@@ -7,29 +9,31 @@ import (
) )
var ( var (
loggerInfo = log.New(io.Discard, "INFO: CrowdsecBouncerTraefikPlugin: ", log.Ldate|log.Ltime) loggerInfo = log.New(io.Discard, "INFO: CrowdsecBouncerTraefikPlugin: ", log.Ldate|log.Ltime) //nolint:gochecknoglobals
loggerDebug = log.New(io.Discard, "DEBUG: CrowdsecBouncerTraefikPlugin: ", log.Ldate|log.Ltime) loggerDebug = log.New(io.Discard, "DEBUG: CrowdsecBouncerTraefikPlugin: ", log.Ldate|log.Ltime) //nolint:gochecknoglobals
loggerError = log.New(io.Discard, "ERROR: CrowdsecBouncerTraefikPlugin: ", log.Ldate|log.Ltime) //nolint:gochecknoglobals
) )
// Init Set Default log level to info in case log level to defined // Init Set Default log level to info in case log level to defined.
func Init(logLevel string) { func Init(logLevel string) {
switch logLevel { loggerError.SetOutput(os.Stderr)
case "INFO": loggerInfo.SetOutput(os.Stdout)
loggerInfo.SetOutput(os.Stdout) if logLevel == "DEBUG" {
case "DEBUG":
loggerInfo.SetOutput(os.Stdout)
loggerDebug.SetOutput(os.Stdout) loggerDebug.SetOutput(os.Stdout)
default:
loggerInfo.SetOutput(os.Stdout)
} }
} }
// Info Log info // Info log to Stdout.
func Info(str string) { func Info(str string) {
loggerInfo.Printf(str) loggerInfo.Printf(str)
} }
// Info Log debug // Debug log to Stdout.
func Debug(str string) { func Debug(str string) {
loggerDebug.Printf(str) loggerDebug.Printf(str)
} }
// Error log to Stderr.
func Error(str string) {
loggerError.Printf(str)
}
@@ -1,3 +1,6 @@
// Package simpleredis implements utility routines for interacting.
// It supports currently the following operations: GET, SET, DELETE,
// and support timetoleave for keys.
package simpleredis package simpleredis
import ( import (
@@ -11,6 +14,14 @@ import (
logger "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/logger" logger "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/logger"
) )
// Error strings for redis.
const (
RedisUnreachable = "redis:unreachable"
RedisMiss = "redis:miss"
RedisTimeout = "redis:timeout"
)
// A RedisCmd is used to communicate with redis at low level using commands.
type RedisCmd struct { type RedisCmd struct {
Command string Command string
Name string Name string
@@ -19,6 +30,7 @@ type RedisCmd struct {
Error error Error error
} }
// A SimpleRedis is used to communicate with redis.
type SimpleRedis struct { type SimpleRedis struct {
redisHost string redisHost string
} }
@@ -33,38 +45,49 @@ func genRedisArray(params ...[]byte) []byte {
return []byte(MSG) return []byte(MSG)
} }
func send(wr *textproto.Writer, method string, data []byte) {
if err := wr.PrintfLine(string(data)); err != nil {
logger.Error(fmt.Sprintf("redis:%s %s", method, err.Error()))
} else {
logger.Debug(fmt.Sprintf("redis:%s", method))
}
}
func askRedis(hostnamePort string, cmd RedisCmd, channel chan RedisCmd) { func askRedis(hostnamePort string, cmd RedisCmd, channel chan RedisCmd) {
conn, err := net.Dial("tcp", hostnamePort) dialer := net.Dialer{Timeout: 2 * time.Second}
conn, err := dialer.Dial("tcp", hostnamePort)
if err != nil { if err != nil {
channel <- RedisCmd{Error: fmt.Errorf(RedisUnreachable)}
return return
} }
defer conn.Close() defer func() {
if err := conn.Close(); err != nil {
logger.Error(fmt.Sprintf("redis:connClose %s", err.Error()))
}
}()
writer := textproto.NewWriter(bufio.NewWriter(conn)) writer := textproto.NewWriter(bufio.NewWriter(conn))
reader := textproto.NewReader(bufio.NewReader(conn)) reader := textproto.NewReader(bufio.NewReader(conn))
switch cmd.Command { switch cmd.Command {
case "SET": case "SET":
data := genRedisArray([]byte("SET"), []byte(cmd.Name), []byte(cmd.Data), []byte("EX"), []byte(fmt.Sprintf("%v", cmd.Duration))) data := genRedisArray([]byte("SET"), []byte(cmd.Name), cmd.Data, []byte("EX"), []byte(fmt.Sprintf("%d", cmd.Duration)))
writer.PrintfLine(string(data)) send(writer, "set", data)
logger.Info("set")
case "DEL": case "DEL":
data := genRedisArray([]byte("DEL"), []byte(cmd.Name)) data := genRedisArray([]byte("DEL"), []byte(cmd.Name))
writer.PrintfLine(string(data)) send(writer, "del", data)
logger.Info("del")
case "GET": case "GET":
data := genRedisArray([]byte("GET"), []byte(cmd.Name)) data := genRedisArray([]byte("GET"), []byte(cmd.Name))
writer.PrintfLine(string(data)) send(writer, "get", data)
logger.Info("get")
for { for {
select { select {
case <-time.After(time.Second * 1): case <-time.After(time.Second * 1):
channel <- RedisCmd{Error: fmt.Errorf("timeout")} channel <- RedisCmd{Error: fmt.Errorf(RedisTimeout)}
return return
default: default:
read, _ := reader.ReadLineBytes() read, _ := reader.ReadLineBytes()
if string(read) != "$1" { if string(read) != "$1" {
channel <- RedisCmd{Error: fmt.Errorf("miss")} channel <- RedisCmd{Error: fmt.Errorf(RedisMiss)}
return return
} }
read, _ = reader.ReadLineBytes() read, _ = reader.ReadLineBytes()
@@ -75,10 +98,12 @@ func askRedis(hostnamePort string, cmd RedisCmd, channel chan RedisCmd) {
} }
} }
// Init sets the redisHost used to connect to redis.
func (sr *SimpleRedis) Init(redisHost string) { func (sr *SimpleRedis) Init(redisHost string) {
sr.redisHost = redisHost sr.redisHost = redisHost
} }
// Get fetches the value for key name in redis.
func (sr *SimpleRedis) Get(name string) ([]byte, error) { func (sr *SimpleRedis) Get(name string) ([]byte, error) {
redisCmd := RedisCmd{ redisCmd := RedisCmd{
Command: "GET", Command: "GET",
@@ -93,6 +118,7 @@ func (sr *SimpleRedis) Get(name string) ([]byte, error) {
return resp.Data, nil return resp.Data, nil
} }
// Set update the value for key name in redis with value data for duration.
func (sr *SimpleRedis) Set(name string, data []byte, duration int64) error { func (sr *SimpleRedis) Set(name string, data []byte, duration int64) error {
redisCmd := RedisCmd{ redisCmd := RedisCmd{
Command: "SET", Command: "SET",
@@ -104,6 +130,7 @@ func (sr *SimpleRedis) Set(name string, data []byte, duration int64) error {
return nil return nil
} }
// Del remove the key name in redis.
func (sr *SimpleRedis) Del(name string) error { func (sr *SimpleRedis) Del(name string) error {
redisCmd := RedisCmd{ redisCmd := RedisCmd{
Command: "DEL", Command: "DEL",
-525
View File
@@ -1,525 +0,0 @@
package netutils
import (
"math/rand"
"net"
"strconv"
"strings"
"sync"
//"sync/atomic"
"time"
)
//Receive msg from tcp socket and send it as a []byte to readChan
func ReadFromTCP2(sock *net.TCPConn, msgBuf []byte, readChan chan []byte,
feedbackChanFromSocket chan int) {
loop := 1
for loop == 1 {
bytes, err := sock.Read(msgBuf)
if err != nil {
feedbackChanFromSocket <- 1
loop = 0
continue
}
b := make([]byte, 0)
b = append(b, msgBuf[:bytes]...)
readChan <- b
}
}
//Receive msg from tcp socket and send it as a []byte to readChan
func ReadFromTCP(sock *net.TCPConn, msgBuf []byte, readChan chan []byte,
feedbackChanFromSocket chan int) {
feedbackToSocket := make(chan bool)
feedbackFromSocket := make(chan bool)
reuseBufferChan := make(chan []byte, 1)
go readFromTCP(sock, readChan, feedbackFromSocket,
feedbackToSocket, reuseBufferChan)
go func() {
<-feedbackFromSocket
feedbackChanFromSocket <- 1
}()
}
/*simple write to TCP, for oneway connections only (no communitcation w/ "read" part of the socket
in terms of error propogation)*/
func WriteToTCPw2(sock *net.TCPConn, writeChan chan []byte,
feedbackChan chan int) {
loop := 1
for loop == 1 {
select {
case msg := <-writeChan:
_, err := sock.Write(msg)
if err != nil {
feedbackChan <- 1
continue
}
case <-feedbackChan:
loop = 0
}
}
}
/*simple write to TCP, for oneway connections only (no communitcation w/ "read" part of the socket
in terms of error propogation)*/
func WriteToTCPw(sock *net.TCPConn, writeChan chan []byte,
feedbackChan chan int) {
feedbackToSocket := make(chan bool)
feedbackFromSocket := make(chan bool)
go writeToTCP(sock, writeChan, feedbackFromSocket, feedbackToSocket)
go func() {
<-feedbackFromSocket
feedbackChan <- 1
}()
}
//simple write to tcp w/ erorr propagation to/from "read" part of the socket
func WriteToTCPrw2(sock *net.TCPConn, writeChan chan []byte,
feedbackChanFromSocket, feedbackChanToSocket chan int) {
loop := 1
for loop == 1 {
select {
case msg := <-writeChan:
_, err := sock.Write(msg)
if err != nil {
select {
case feedbackChanFromSocket <- 1:
continue
case loop = <-feedbackChanToSocket:
loop = 0
continue
}
}
case <-feedbackChanToSocket:
loop = 0
}
}
}
//simple write to tcp w/ erorr propagation to/from "read" part of the socket
func WriteToTCPrw(sock *net.TCPConn, writeChan chan []byte,
feedbackChanFromSocket, feedbackChanToSocket chan int) {
feedbackToSocket := make(chan bool)
feedbackFromSocket := make(chan bool)
go writeToTCP(sock, writeChan, feedbackFromSocket, feedbackToSocket)
go func() {
select {
case <-feedbackFromSocket:
feedbackChanFromSocket <- 1
case <-feedbackChanToSocket:
feedbackToSocket <- true
}
}()
}
func makeBuffer(reuseBufferChan chan []byte) []byte {
select {
case buf := <-reuseBufferChan:
return buf
default:
return make([]byte, 10000)
}
}
//Receive msg from tcp socket and send it as a []byte to readChan,with buffer reuse
func readFromTCP(sock *net.TCPConn, readChan chan []byte,
feedbackFromSocket, feedbackToSocket chan bool,
reuseBufferChan chan []byte) {
loop := 1
var buf []byte
for loop == 1 {
buf = makeBuffer(reuseBufferChan)
bytes, err := sock.Read(buf)
if err != nil {
select {
case feedbackFromSocket <- true:
loop = 0
continue
case <-feedbackToSocket:
loop = 0
continue
}
}
select {
case readChan <- buf[:bytes]:
case <-feedbackToSocket:
loop = 0
continue
}
}
}
//TCP's write routine, with feedback's chans
func writeToTCP(sock *net.TCPConn, writeChan chan []byte,
feedbackFromSocket, feedbackToSocket chan bool) {
loop := 1
for loop == 1 {
select {
case msg := <-writeChan:
_, err := sock.Write(msg)
if err != nil {
select {
case feedbackFromSocket <- true:
case <-feedbackToSocket:
}
loop = 0
continue
}
case <-feedbackToSocket:
loop = 0
continue
}
}
}
//reconnecting to remote host for both read and write purpose
func ReconnectTCPRW(ladr, radr *net.TCPAddr, msgBuf []byte, writeChan chan []byte,
readChan chan []byte, feedbackChanToSocket, feedbackChanFromSocket chan int,
init_msg []byte) {
loop := 1
for loop == 1 {
sock, err := net.DialTCP("tcp", ladr, radr)
if err != nil {
time.Sleep(time.Duration(20+rand.Intn(15)) * time.Second)
continue
}
//testing health of the new socket. GO sometimes doesnt rise the error when
// we receive RST from remote side
_, err = sock.Write(init_msg)
if err != nil {
sock.Close()
time.Sleep(time.Duration(20+rand.Intn(15)) * time.Second)
continue
}
loop = 0
go ReadFromTCP(sock, msgBuf, readChan, feedbackChanFromSocket)
go WriteToTCPrw(sock, writeChan, feedbackChanFromSocket, feedbackChanToSocket)
}
}
func ReconnectTCPRWReuse(ladr, radr *net.TCPAddr,
readChan, writeChan, reuseBufferChan chan []byte,
readFeedbackFrom, readFeedbackTo chan bool,
writeFeedbackFrom, writeFeedbackTo chan bool) {
loop := 1
for loop == 1 {
sock, err := net.DialTCP("tcp", ladr, radr)
if err != nil {
time.Sleep(time.Duration(20+rand.Intn(15)) * time.Second)
continue
}
loop = 0
go readFromTCP(sock, readChan, readFeedbackFrom, readFeedbackTo,
reuseBufferChan)
go writeToTCP(sock, writeChan, writeFeedbackFrom, writeFeedbackTo)
}
}
func AutoRecoonectedTCP(ladr, radr *net.TCPAddr, msgBuf, initMsg []byte,
writeChan, readChan chan []byte, flushChan chan int) {
feedbackChanFromSocket := make(chan int)
feedbackChanToSocket := make(chan int)
go ReconnectTCPRW(ladr, radr, msgBuf, writeChan, readChan, feedbackChanToSocket,
feedbackChanFromSocket, initMsg)
for {
select {
case feedbackFromSocket := <-feedbackChanFromSocket:
feedbackChanToSocket <- feedbackFromSocket
flushChan <- 1
go ReconnectTCPRW(ladr, radr, msgBuf, writeChan,
readChan, feedbackChanToSocket,
feedbackChanFromSocket, initMsg)
}
}
}
func AutoRecoonectedTCPReuse(ladr, radr *net.TCPAddr,
readChan, writeChan chan []byte,
reuseChan chan []byte,
flushChan chan bool) {
readFeedbackFrom := make(chan bool)
readFeedbackTo := make(chan bool)
writeFeedbackFrom := make(chan bool)
writeFeedbackTo := make(chan bool)
go ReconnectTCPRWReuse(ladr, radr, readChan, writeChan, reuseChan,
readFeedbackFrom, readFeedbackTo,
writeFeedbackFrom, writeFeedbackTo)
for {
select {
case <-readFeedbackFrom:
writeFeedbackTo <- true
case <-writeFeedbackFrom:
readFeedbackTo <- true
}
flushChan <- true
go ReconnectTCPRWReuse(ladr, radr, readChan, writeChan, reuseChan,
readFeedbackFrom, readFeedbackTo,
writeFeedbackFrom, writeFeedbackTo)
}
}
//reconnecting to remote host for write only
func ReconnectTCPW(radr net.TCPAddr, writeChan chan []byte, feedbackChan chan int) {
loop := 1
for loop == 1 {
time.Sleep(time.Duration(20+rand.Intn(15)) * time.Second)
sock, err := net.DialTCP("tcp", nil, &radr)
if err != nil {
continue
}
//testing health of the new socket. GO sometimes doesnt rise the error when
// we receive RST from remote side
_, err = sock.Write([]byte{1})
if err != nil {
sock.Close()
continue
}
loop = 0
go WriteToTCPw(sock, writeChan, feedbackChan)
}
}
/* --------------------- CONNECTION MANAGER -------------------------
Connection manager will allow send data and receive data from remote hosts.
it will have single ConnectionMsg (see below) read chan and single
ConnectionMsg write chan toward it's clients
as well as single read chan from sockets, but multiple write sockets.
it will route msgs according to Host field in connectionManager struct
(if it received from client, it will send this msg toward Host's sockets;
if recved from socket, will proxy it toward client(and client will know from which remote host
it was received)
-------------------------------------------------------------------- */
/*
MsgType's could be:
from Api's client to ConnectionManager:
"Data" - msg with Data to Host
"Connect" - connect to new Host
...
from ConnectionManager to Api's client:
"BufferFlush" - notification that connection to remote Host not longer working.
advice to flush all the msg buffers assosiated with remote host
*/
type ConnectionMsg struct {
Host string
Data []byte
Type string
}
/*
Receive msg from tcp socket and send it as a ConnectionMsg to readChan
TODO: think about more generic version to be more DRYer (to work in both CM and []byte chans
*/
func CMReadFromTCP(sock *net.TCPConn, readChan chan ConnectionMsg,
peerAddress string) {
msgBuf := make([]byte, 65000)
loop := 1
var msg ConnectionMsg
msg.Host = peerAddress
msg.Type = "Data"
for loop == 1 {
bytes, err := sock.Read(msgBuf)
if err != nil {
msg.Type = "ReadError"
readChan <- msg
loop = 0
continue
}
b := make([]byte, 0)
b = append(b, msgBuf[:bytes]...)
msg.Data = b
readChan <- msg
}
}
/*
ConnectionManager write instance to tcp w/ erorr propagation to/from "read" part of the socket
TODO: think about more generic version to be more DRYer (to work in both CM and []byte chans
*/
func CMWriteToTCP(sock *net.TCPConn, writeChan, readChan chan ConnectionMsg,
peerAddress string) {
loop := 1
var errorMsg ConnectionMsg
errorMsg.Host = peerAddress
errorMsg.Type = "WriteError"
for loop == 1 {
select {
case msg := <-writeChan:
switch msg.Type {
case "Data":
_, err := sock.Write(msg.Data)
if err != nil {
for loop == 1 {
select {
case readChan <- errorMsg:
case errorMsg := <-writeChan:
if errorMsg.Type != "ConnectionError" {
continue
}
}
loop = 0
}
}
case "ConnectionError":
loop = 0
default:
continue
}
}
}
}
func StartConnection(tcpConn *net.TCPConn, writeChan,
readChan chan ConnectionMsg, peerAddress string) {
go CMReadFromTCP(tcpConn, readChan, peerAddress)
go CMWriteToTCP(tcpConn, writeChan, readChan, peerAddress)
}
func CMListenForConnection(mutex *sync.RWMutex, localPort int,
writeChanMap map[string]chan ConnectionMsg,
connectionStateMap map[string]int,
readChan chan ConnectionMsg) {
laddr := strings.Join([]string{":", strconv.Itoa(localPort)}, "")
tcpLaddr, err := net.ResolveTCPAddr("tcp", laddr)
if err != nil {
panic("cant resolve local address for binding")
}
tcpListener, err := net.ListenTCP("tcp", tcpLaddr)
if err != nil {
panic("cant listen on local address for binding")
}
for {
tcpConn, err := tcpListener.AcceptTCP()
if err == nil {
radr := strings.Split(tcpConn.RemoteAddr().String(), ":")[0]
// check if we already has connection to remote peer as a client
mutex.Lock()
if val, exist := connectionStateMap[radr]; exist && val == 1 {
tcpConn.Close()
mutex.Unlock()
continue
}
connectionStateMap[radr] = 1
if writeChan, exist := writeChanMap[radr]; exist {
mutex.Unlock()
go StartConnection(tcpConn, writeChan, readChan, radr)
} else {
writeChanMap[radr] = make(chan ConnectionMsg)
mutex.Unlock()
go StartConnection(tcpConn, writeChanMap[radr], readChan, radr)
}
}
}
}
func CMConnectToRemotePeer(mutex *sync.RWMutex, peerTcpAddr *net.TCPAddr,
radr string,
writeChan chan ConnectionMsg,
readChan chan ConnectionMsg,
connectionStateMap map[string]int) {
connectLoop := 1
for connectLoop == 1 {
mutex.RLock()
if connectionStateMap[radr] == 1 {
connectLoop = 0
mutex.RUnlock()
continue
}
mutex.RUnlock()
tcpConn, err := net.DialTCP("tcp", nil, peerTcpAddr)
if err != nil {
time.Sleep(time.Second * time.Duration(rand.Int63n(15)))
continue
}
mutex.Lock()
if connectionStateMap[radr] == 0 {
connectionStateMap[radr] = 1
go StartConnection(tcpConn, writeChan, readChan, radr)
mutex.Unlock()
connectLoop = 0
continue
} else {
mutex.Unlock()
tcpConn.Close()
connectLoop = 0
continue
}
}
}
func ConnectionManager(msgChan chan ConnectionMsg, localPort int) {
writeChanMap := make(map[string]chan ConnectionMsg)
connectionStateMap := make(map[string]int)
readChan := make(chan ConnectionMsg)
var connectionMutex sync.RWMutex
go CMListenForConnection(&connectionMutex, localPort, writeChanMap,
connectionStateMap, readChan)
for {
select {
case msgToPeer := <-msgChan:
switch msgToPeer.Type {
case "Data":
if state, exists := connectionStateMap[msgToPeer.Host]; exists && state == 1 {
/*FIXME/THINK: There could be deadlock if connectin closes before we will
be able to send to the chan */
writeChan := writeChanMap[msgToPeer.Host]
writeChan <- msgToPeer
} else {
msgChan <- ConnectionMsg{Type: "ConnectionNotExist"}
}
case "Connect":
if len(strings.Split(msgToPeer.Host, ":")) > 1 {
radr := strings.Split(msgToPeer.Host, ":")[0]
connectionMutex.Lock()
if _, exist := writeChanMap[radr]; !exist {
writeChanMap[radr] = make(chan ConnectionMsg)
}
connectionMutex.Unlock()
peerTcpAddr, err := net.ResolveTCPAddr("tcp", msgToPeer.Host)
if err != nil {
//XXX: think about , mb make something less drastic
panic("cant resolve remote address")
}
go CMConnectToRemotePeer(&connectionMutex, peerTcpAddr, radr,
writeChanMap[radr], readChan, connectionStateMap)
} else {
connectionMutex.Lock()
if _, exist := writeChanMap[msgToPeer.Host]; !exist {
writeChanMap[msgToPeer.Host] = make(chan ConnectionMsg)
}
connectionMutex.Unlock()
remoteAddr := strings.Join([]string{msgToPeer.Host, strconv.Itoa(localPort)}, ":")
peerTcpAddr, err := net.ResolveTCPAddr("tcp", remoteAddr)
if err != nil {
//XXX: again panic could be overkill
panic("cant resolve remote address")
}
go CMConnectToRemotePeer(&connectionMutex, peerTcpAddr, msgToPeer.Host,
writeChanMap[msgToPeer.Host], readChan, connectionStateMap)
}
}
case msgFromPeer := <-readChan:
switch msgFromPeer.Type {
case "Data":
msgChan <- msgFromPeer
case "WriteError", "ReadError":
connectionMutex.Lock()
connectionStateMap[msgFromPeer.Host] = 0
connectionMutex.Unlock()
if msgFromPeer.Type == "ReadError" {
writeChanMap[msgFromPeer.Host] <- ConnectionMsg{Type: "ConnectionError"}
}
var msgToApiClient ConnectionMsg
msgToApiClient.Host = msgFromPeer.Host
msgToApiClient.Type = "BufferFlush"
msgChan <- msgToApiClient
}
}
}
}
-42
View File
@@ -1,42 +0,0 @@
package netutils
import (
"errors"
"net"
)
/*
we need to provide a function, which will read/write to/from socket and read/write to/from sockets feedback chans
*/
func ListenForConnection(port string, fn func(chan []byte, chan []byte, chan int, chan int)) error {
addr := ":" + port
tcpAddr, err := net.ResolveTCPAddr("tcp", addr)
if err != nil {
return errors.New("cant resolve local tcp address")
}
loop := 1
servSock, err := net.ListenTCP("tcp", tcpAddr)
if err != nil {
return errors.New("cant bind to local tcp address")
}
for loop == 1 {
sock, err := servSock.AcceptTCP()
if err != nil {
continue
}
go ServeTcpConn(sock, fn)
}
return nil
}
func ServeTcpConn(sock *net.TCPConn, fn func(chan []byte, chan []byte, chan int, chan int)) {
readChan := make(chan []byte)
writeChan := make(chan []byte)
feedbackFrom := make(chan int, 1)
feedbackTo := make(chan int, 1)
buf := make([]byte, 65535)
go ReadFromTCP(sock, buf, readChan, feedbackFrom)
go WriteToTCPrw(sock, writeChan, feedbackFrom, feedbackTo)
fn(readChan, writeChan, feedbackFrom, feedbackTo)
}
-3
View File
@@ -1,6 +1,3 @@
# github.com/leprosus/golang-ttl-map v1.1.7 # github.com/leprosus/golang-ttl-map v1.1.7
## explicit; go 1.15 ## explicit; go 1.15
github.com/leprosus/golang-ttl-map github.com/leprosus/golang-ttl-map
# github.com/tehnerd/goUtils v0.0.0-20150515130609-5a2d8fb2ded8
## explicit
github.com/tehnerd/goUtils/netutils