Enable/appsec

cmd/root.go

@@ -2,6 +2,7 @@ package cmd
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"flag"
 	"fmt"
@@ -62,21 +63,25 @@ func Execute() error {
 	showConfig := flag.Bool("T", false, "show full config (.yaml + .yaml.local) and exit")
 
 	// Worker flags
-	tcpAddr := flag.String("tcp", "", "tcp listener address")
-	unixAddr := flag.String("unix", "", "unix listener address")
+	workerConfigString := flag.String("config", "", "whole worker configuration as json")
 	workerMode := flag.Bool("worker", false, "run as worker")
 
 	flag.Parse()
 
-	if !*workerMode && (*tcpAddr != "" || *unixAddr != "") {
-		return fmt.Errorf("parent process cannot have listener address")
+	if !*workerMode && (*workerConfigString != "") {
+		return fmt.Errorf("parent process cannot have worker config")
 	}
 
 	if *workerMode {
-		if *tcpAddr == "" && *unixAddr == "" {
-			return fmt.Errorf("worker process must have one listener address")
+		var workerConfig *worker.Worker = &worker.Worker{}
+		err := json.Unmarshal([]byte(*workerConfigString), workerConfig)
+		if err != nil {
+			return fmt.Errorf("unable to unmarshal worker config: %w", err)
+		}
+		if workerConfig.ListenAddr == "" && workerConfig.ListenSocket == "" {
+			return fmt.Errorf("worker process must have one listener address: %+v", *workerConfigString)
 		}
-		return WorkerExecute(*tcpAddr, *unixAddr)
+		return WorkerExecute(workerConfig)
 	}
 
 	if *bouncerVersion {
@@ -110,7 +115,8 @@ func Execute() error {
 	}
 
 	log.Infof("Starting %s %s", name, version.String())
-
+	log.Debugf("Configuration: %+v", config.Workers[0])
+	log.Debugf("Configuration: %+v", *config.AppSec)
 	bouncer := &csbouncer.StreamBouncer{}
 
 	err = bouncer.ConfigReader(strings.NewReader(configExpanded))
@@ -225,7 +231,8 @@ func Execute() error {
 		defer adminServer.Close()
 	}
 
-	workerManager := worker.NewManager(ctx, workerServer, config.WorkerUid, config.WorkerGid)
+	ctx, cancel := context.WithCancel(context.Background())
+	workerManager := worker.NewManager(ctx, &cancel, workerServer, config.WorkerUid, config.WorkerGid)
 
 	g.Go(func() error {
 		return workerManager.Run()
@@ -234,6 +241,7 @@ func Execute() error {
 	apiServer := api.NewApi(ctx, workerManager, HostManager, dataSet, &config.Geo, socketConnChan)
 
 	for _, worker := range config.Workers {
+		log.Debugf("Adding worker %v", worker)
 		workerManager.CreateChan <- worker
 	}
 
@@ -259,15 +267,15 @@ func Execute() error {
 	return nil
 }
 
-func WorkerExecute(tcpAddr, unixAddr string) error {
+func WorkerExecute(workerConfig *worker.Worker) error {
 
 	g, ctx := errgroup.WithContext(context.Background())
 
 	g.Go(func() error {
 		return HandleSignals(ctx)
 	})
 
-	spoad, err := spoa.New(tcpAddr, unixAddr)
+	spoad, err := spoa.New(workerConfig)
 
 	if err != nil {
 		return fmt.Errorf("failed to create SPOA: %w", err)

config/crowdsec-spoa-bouncer.yaml

@@ -18,8 +18,8 @@ insecure_skip_verify: false
 #   - name: spoa1
 #     listen_addr: 0.0.0.0:9000
 #     listen_socket: /run/crowdsec-spoa/spoa-1.sock
-# worker_user: crowdsec-spoa
-# worker_group: crowdsec-spoa
+worker_user: crowdsec-spoa
+worker_group: crowdsec-spoa
 # asn_database_path: /var/lib/crowdsec/data/GeoLite2-ASN.mmdb
 # city_database_path: /var/lib/crowdsec/data/GeoLite2-City.mmdb
 

go.sum

@@ -13,7 +13,7 @@ github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XL
 github.com/coreos/go-systemd/v22 v22.5.0 h1:RrqgGjYQKalulkV8NGVIfkXQf6YYmOyiJKk8iXXhfZs=
 github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
-github.com/crowdsecurity/crowdsec v1.6.3 h1:L/6iT2/Gfl9bc9DQkHJz2BbpKM3P+yW6ocCKRyF4j1g=
+github.com/crowdsecurity/crowdsec v1.6.3 h1:2eMK2SPSOCTafeMBZmffYrgC9Y13LMIyCO9MEUgjD7o=
 github.com/crowdsecurity/crowdsec v1.6.3/go.mod h1:LrdAX9l4vgaExQbNUVnvZIu/DPwD9pSE9gBj14D4MTo=
 github.com/crowdsecurity/go-cs-bouncer v0.0.14 h1:0hxOaa59pMT274qDzJXNxps4QfMnhSNss+oUn36HTpw=
 github.com/crowdsecurity/go-cs-bouncer v0.0.14/go.mod h1:4nSF37v7i98idHM6cw1o0V0XgiY25EjTLfFFXvqg6OA=
@@ -110,6 +110,8 @@ github.com/google/go-querystring v1.1.0/go.mod h1:Kcdr2DB4koayq7X8pmAG4sNG59So17
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/hashicorp/go-version v1.2.1 h1:zEfKbn2+PDgroKdiOzqiE8rsmLqU2uwi5PB5pBJ3TkI=
+github.com/hashicorp/go-version v1.2.1/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/joho/godotenv v1.3.0/go.mod h1:7hK45KPybAkOC6peb+G5yklZfMxEjkZhHbwpqxOKXbg=
 github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=

internal/appsec/root.go

@@ -1,21 +1,145 @@
 package appsec
 
 import (
+	"bytes"
 	"context"
+	"fmt"
+	"net/http"
+	"time"
 
 	log "github.com/sirupsen/logrus"
 )
 
-type AppSec struct {
-	AlwaysSend bool       `yaml:"always_send"`
-	logger     *log.Entry `yaml:"-"`
+type AppsecRequest struct {
+	headers       *http.Header
+	body          []byte
+	method        string
+	httpValidated bool
+	tcpValidated  bool
+	httpChan      chan<- *http.Request
+	url           string
+	timeout       time.Duration
+	logger        *log.Entry
 }
 
-func (a *AppSec) Init(logger *log.Entry, ctx context.Context) error {
+type AppsecConfig struct {
+	Enabled        bool                      `yaml:"enabled"`
+	AppsecUrl      string                    `yaml:"appsec_url"`
+	logger         *log.Entry                `yaml:"-"`
+	AppsecRequests map[string]*AppsecRequest `yaml:"-"`
+	httpChan       <-chan *http.Request      `yaml:"-"`
+	ApiKey         string                    `yaml:"api_key"`
+}
+
+// idempotent function, can be called multiple times
+func (a *AppsecRequest) Isready() bool {
+	return a.httpValidated && a.tcpValidated
+}
+
+func (a *AppsecRequest) GenerateHTTPRequest() *http.Request {
+	ctx, cancel := context.WithTimeout(context.Background(), a.timeout)
+	defer cancel()
+	req, err := http.NewRequestWithContext(ctx, a.method, a.url, bytes.NewReader(a.body))
+	if err != nil {
+		a.logger.Errorf("failed to create request: %v", err)
+		return nil
+	}
+	req.Header = *a.headers
+	return req
+}
+
+func (a *AppsecRequest) ValidateHTTP() {
+	a.httpValidated = true
+	if a.tcpValidated {
+		a.httpChan <- a.GenerateHTTPRequest()
+	}
+}
+
+func (a *AppsecRequest) ValidateTCP() {
+	a.tcpValidated = true
+	if a.httpValidated {
+		a.httpChan <- a.GenerateHTTPRequest()
+	}
+}
+
+func (a *AppsecRequest) AddHeaders(header *http.Header) {
+	if a.headers != nil {
+		a.headers = &http.Header{}
+	}
+	for key, values := range *header {
+		for _, value := range values {
+			a.headers.Add(key, value)
+		}
+	}
+}
+
+func (a *AppsecRequest) SetMethod(method string) {
+	a.method = method
+}
+
+func (a *AppsecRequest) SetBody(body []byte) {
+	a.body = body
+}
+
+func (a *AppsecConfig) Init(logger *log.Entry) error {
 	a.InitLogger(logger)
+
+	if a.Enabled && a.AppsecUrl == "" {
+		a.logger.Errorf("appsec is enabled but no appsec_url is set")
+		return fmt.Errorf("appsec is enabled but no appsec_url is set")
+	}
+
+	if !a.Enabled {
+		a.logger.Warnf("appsec is disabled")
+		return nil
+	}
+
+	a.logger.Debugf("appsec is enabled")
+	a.AppsecRequests = make(map[string]*AppsecRequest)
 	return nil
 }
 
-func (a *AppSec) InitLogger(logger *log.Entry) {
+func (a *AppsecConfig) InitLogger(logger *log.Entry) {
 	a.logger = logger.WithField("type", "appsec")
 }
+
+// beware of malformed requests
+func (a *AppsecConfig) SetMethodAndBody(id string, method string, body []byte) {
+	_, ok := a.AppsecRequests[id]
+	if ok {
+		a.AppsecRequests[id].SetMethod(method)
+		a.AppsecRequests[id].SetBody(body)
+	} else {
+		a.AppsecRequests[id] = &AppsecRequest{
+			method: method,
+			body:   body,
+			logger: a.logger,
+			url:    a.AppsecUrl,
+		}
+	}
+}
+
+func (a *AppsecConfig) AddHeaders(id string, headers *http.Header) {
+	_, ok := a.AppsecRequests[id]
+	if ok {
+		a.AppsecRequests[id].AddHeaders(headers)
+	} else {
+		a.AppsecRequests[id] = &AppsecRequest{
+			headers: headers,
+			logger:  a.logger,
+			url:     a.AppsecUrl,
+		}
+	}
+}
+
+func (a *AppsecConfig) ValidateConfiguration() error {
+
+	if a.AppsecUrl == "" {
+		return fmt.Errorf("appsec_url is empty")
+	}
+
+	if a.ApiKey == "" {
+		return fmt.Errorf("api_key is empty")
+	}
+	return nil
+}

internal/worker/root.go

@@ -2,36 +2,40 @@ package worker
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"os"
 	"os/exec"
 	"syscall"
 
+	"github.com/crowdsecurity/crowdsec-spoa/internal/appsec"
 	"github.com/crowdsecurity/crowdsec-spoa/pkg/server"
 	log "github.com/sirupsen/logrus"
 )
 
 type Worker struct {
-	Name         string     `yaml:"name"`
-	ListenAddr   string     `yaml:"listen_addr"`
-	ListenSocket string     `yaml:"listen_socket"`
-	LogLevel     *log.Level `yaml:"log_level"`
-	Uid          int        `yaml:"-"` // Set by the worker manager
-	Gid          int        `yaml:"-"` // Set by the worker manager
-	Command      *exec.Cmd  `yaml:"-"`
-	SocketPath   string     `yaml:"-"` // Set by combining the socket dir and the worker name
+	Name         string               `yaml:"name"`
+	LogLevel     *log.Level           `yaml:"log_level"`
+	ListenAddr   string               `yaml:"listen_addr"`
+	ListenSocket string               `yaml:"listen_socket"`
+	AppSecConfig *appsec.AppsecConfig `yaml:"appsec_config"`
+	Uid          int                  `yaml:"-"` // Set by the worker manager
+	Gid          int                  `yaml:"-"` // Set by the worker manager
+	Command      *exec.Cmd            `yaml:"-"`
+	SocketPath   string               `yaml:"-"` // Set by combining the socket dir and the worker name
 }
 
 func (w *Worker) Run(socket string) error {
 	args := []string{
 		"-worker",
 	}
-	if w.ListenAddr != "" {
-		args = append(args, "-tcp", w.ListenAddr)
-	}
-	if w.ListenSocket != "" {
-		args = append(args, "-unix", w.ListenSocket)
+
+	config, err := json.Marshal(*w)
+	if err != nil {
+		return fmt.Errorf("failed to marshal appsec config: %w", err)
 	}
+
+	args = append(args, "-config", string(config))
 	command := exec.Command(os.Args[0], args...)
 
 	command.Env = []string{
@@ -57,19 +61,21 @@ func (w *Worker) Run(socket string) error {
 }
 
 type Manager struct {
-	Workers    []*Worker       `yaml:"-"`
-	CreateChan chan *Worker    `yaml:"-"`
-	Ctx        context.Context `yaml:"-"`
-	Server     *server.Server  `yaml:"-"`
-	WorkerUid  int             `yaml:"-"`
-	WorkerGid  int             `yaml:"-"`
+	Workers    []*Worker           `yaml:"-"`
+	CreateChan chan *Worker        `yaml:"-"`
+	Ctx        context.Context     `yaml:"-"`
+	Cancel     *context.CancelFunc `yaml:"-"`
+	Server     *server.Server      `yaml:"-"`
+	WorkerUid  int                 `yaml:"-"`
+	WorkerGid  int                 `yaml:"-"`
 }
 
-func NewManager(ctx context.Context, s *server.Server, uid, gid int) *Manager {
+func NewManager(ctx context.Context, cancel *context.CancelFunc, s *server.Server, uid, gid int) *Manager {
 	return &Manager{
 		CreateChan: make(chan *Worker),
 		Workers:    make([]*Worker, 0),
 		Ctx:        ctx,
+		Cancel:     cancel,
 		Server:     s,
 		WorkerUid:  uid,
 		WorkerGid:  gid,
@@ -96,7 +102,13 @@ func (m *Manager) AddWorker(w *Worker) {
 		log.Errorf("failed to create worker listener: %s", err)
 		return
 	}
-	go w.Run(socketString)
+	go func() {
+		err := w.Run(socketString)
+		if err != nil {
+			log.Errorf("worker failed with: %s", err)
+		}
+		defer (*m.Cancel)()
+	}()
 	m.Workers = append(m.Workers, w)
 }
 

pkg/cfg/config.go

@@ -9,6 +9,7 @@ import (
 
 	"gopkg.in/yaml.v2"
 
+	"github.com/crowdsecurity/crowdsec-spoa/internal/appsec"
 	"github.com/crowdsecurity/crowdsec-spoa/internal/geo"
 	"github.com/crowdsecurity/crowdsec-spoa/internal/worker"
 	"github.com/crowdsecurity/crowdsec-spoa/pkg/host"
@@ -35,6 +36,7 @@ type BouncerConfig struct {
 	WorkerSocketDir  string                  `yaml:"worker_socket"`
 	WorkerUid        int                     `yaml:"-"`
 	WorkerGid        int                     `yaml:"-"`
+	AppSec           *appsec.AppsecConfig    `yaml:"appsec"`
 }
 
 // MergedConfig() returns the byte content of the patched configuration file (with .yaml.local).
@@ -94,5 +96,15 @@ func NewConfig(reader io.Reader) (*BouncerConfig, error) {
 		config.WorkerSocketDir = config.WorkerSocketDir + "/"
 	}
 
+	if config.AppSec == nil {
+		config.AppSec = &appsec.AppsecConfig{
+			Enabled: false}
+	}
+	if config.AppSec.Enabled {
+		err = config.AppSec.ValidateConfiguration()
+		if err != nil {
+			return nil, fmt.Errorf("failed to validate appsec configuration: %w", err)
+		}
+	}
 	return config, nil
 }