functional tests (#48)

Signed-off-by: Shivam Sandbhor <[email protected]>
Signed-off-by: Shivam Sandbhor <[email protected]>
Co-authored-by: Shivam Sandbhor <[email protected]>
Co-authored-by: Shivam Sandbhor <[email protected]>

.github/workflows/build-binary-package.yml

@@ -11,12 +11,12 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - name: Set up Go 1.19
-      uses: actions/setup-go@v1
+      uses: actions/setup-go@v3
       with:
         go-version: 1.19
       id: go
     - name: Check out code into the Go module directory
-      uses: actions/checkout@v2
+      uses: actions/checkout@v3
     - name: Build the binaries
       run: make release
     - name: Upload to release

.github/workflows/functional_tests.yaml

@@ -0,0 +1,34 @@
+name: Functional Tests
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    branches: [ main ]
+
+jobs:
+  install_crowdsec:
+    name: Build
+    runs-on: ubuntu-latest
+    steps:
+    - name: Set up Go 1.19
+      uses: actions/setup-go@v3
+      with:
+        go-version: 1.19
+      id: go
+    - name: Check out code into the Go module directory
+      uses: actions/checkout@v3
+    - id: cache-pipenv
+      uses: actions/cache@v3
+      with:
+        path: ~/.local/share/virtualenvs
+        key: ${{ runner.os }}-pipenv-${{ hashFiles('**/Pipfile.lock') }}
+    - name: Install deps for tests
+      run: |
+        sudo apt install -y nftables iptables ipset
+        python3 -m pip install --upgrade pipenv wheel
+    - name: Run tests
+      run: |
+        make build
+        pipenv install --deploy
+        pipenv run pytest

.github/workflows/go.yml

@@ -18,8 +18,14 @@ jobs:
         go-version: 1.19
       id: go
     - name: Check out code into the Go module directory
-      uses: actions/checkout@v2
+      uses: actions/checkout@v3
     - name: Build
       run: make build
     - name: Test
       run: go test -v
+    - name: golangci-lint
+      uses: golangci/golangci-lint-action@v3
+      with:
+        version: v1.51
+        args: --issues-exit-code=1 --timeout 10m
+        only-new-issues: false

.gitignore

@@ -1,9 +1,5 @@
 # Binaries for programs and plugins
-*.exe
-*.exe~
-*.dll
-*.so
-*.dylib
+crowdsec-custom-bouncer
 
 # Test binary, built with `go test -c`
 *.test

Makefile

@@ -20,6 +20,9 @@ export LD_OPTS=-ldflags "-s -w -X github.com/crowdsecurity/cs-custom-bouncer/pkg
 
 RELDIR = "crowdsec-custom-bouncer-${BUILD_VERSION}"
 
+PYTHON=python3
+PIP=pip
+
 all: clean test build
 
 static: clean
@@ -51,4 +54,8 @@ release: build
 	@chmod +x $(RELDIR)/uninstall.sh
 	@chmod +x $(RELDIR)/upgrade.sh
 	@tar cvzf crowdsec-custom-bouncer.tgz $(RELDIR)
-
+
+.PHONY: func-tests
+func-tests: build
+	pipenv install --dev
+	pipenv run pytest -v

Pipfile

@@ -0,0 +1,12 @@
+[packages]
+pytest = "*"
+flask = "*"
+pytimeparse = "*"
+psutil = "*"
+
+[dev-packages]
+gnureadline = "*"
+ipdb = "*"
+
+[requires]
+python_version = "3.10"

config/crowdsec-custom-bouncer.yaml

@@ -1,8 +1,8 @@
 bin_path: ${BINARY_PATH}
 feed_via_stdin: false # Invokes binary once and feeds incoming decisions to it's stdin.
 total_retries: 0 # number of times to restart binary. relevant if feed_via_stdin=true . Set to -1 for infinite retries.
-scenarios_containing: []
-scenarios_not_containing: []
+scenarios_containing: [] # ignore IPs banned for triggering scenarios not containing either of provided word, eg ["ssh", "http"]
+scenarios_not_containing: [] # ignore IPs banned for triggering scenarios containing either of provided word
 origins: []
 piddir: /var/run/
 update_frequency: 10s

custom.go

@@ -20,6 +20,7 @@ type DecisionKey struct {
 
 type DecisionWithAction struct {
 	models.Decision
+	ID     int64  `json:"id"`
 	Action string `json:"action,omitempty"`
 }
 
@@ -73,6 +74,7 @@ func (c *customBouncer) Add(decision *models.Decision) error {
 	}
 	if c.feedViaStdin {
 		fmt.Fprintln(c.binaryStdin, str)
+		c.newDecisionValueSet[decisionToDecisionKey(decision)] = struct{}{}
 		return nil
 	}
 	cmd := exec.Command(c.path, "add", *decision.Value, strconv.Itoa(int(banDuration.Seconds())), *decision.Scenario, str)
@@ -99,6 +101,7 @@ func (c *customBouncer) Delete(decision *models.Decision) error {
 	}
 	if c.feedViaStdin {
 		fmt.Fprintln(c.binaryStdin, str)
+		c.expiredDecisionValueSet[decisionToDecisionKey(decision)] = struct{}{}
 		return nil
 	}
 	if err != nil {
@@ -118,7 +121,7 @@ func (c *customBouncer) ShutDown() error {
 }
 
 func serializeDecision(decision *models.Decision, action string) (string, error) {
-	d := DecisionWithAction{Decision: *decision, Action: action}
+	d := DecisionWithAction{Decision: *decision, Action: action, ID: decision.ID}
 	serbyte, err := json.Marshal(d)
 	if err != nil {
 		return "", fmt.Errorf("serialize error : %s", err)

custom_test.go

@@ -59,7 +59,7 @@ func parseLine(line string) parsedLine {
 
 func cleanup() {
 	if _, err := os.Stat(binaryOutputFile); err != nil {
-		fmt.Println("didnt found the file")
+		fmt.Println("didn't found the file")
 		return
 	}
 	os.Remove(binaryOutputFile)

main.go

@@ -125,8 +125,10 @@ func main() {
 		return
 	}
 	cacheResetTicker := time.NewTicker(config.CacheRetentionDuration)
-
-	go bouncer.Run()
+	go func() {
+		bouncer.Run()
+		t.Kill(fmt.Errorf("stream init failed"))
+	}()
 	if config.PrometheusConfig.Enabled {
 		listenOn := net.JoinHostPort(
 			config.PrometheusConfig.ListenAddress,
@@ -178,7 +180,6 @@ func main() {
 				log.Error("maximum retries exceeded for binary. Exiting")
 				t.Kill(err)
 				return err
-
 			},
 		)
 
@@ -192,7 +193,9 @@ func main() {
 				log.Infoln("terminating bouncer process")
 				if config.PrometheusConfig.Enabled {
 					log.Infoln("terminating prometheus server")
-					promServer.Shutdown(context.Background())
+					if err := promServer.Shutdown(context.Background()); err != nil {
+						log.Errorf("unable to shutdown prometheus server: %s", err)
+					}
 				}
 				return nil
 			case decisions := <-bouncer.Stream:

pytest-debug.ini

@@ -0,0 +1,3 @@
+[pytest]
+# drop to pdb on first failure
+addopts = --pdb --pdbcls=IPython.terminal.debugger:Pdb

pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+addopts =

tests/mock_lapi.py

@@ -0,0 +1,139 @@
+import datetime
+import logging
+from datetime import timedelta
+from ipaddress import ip_address
+from threading import Thread
+from time import sleep
+
+from flask import Flask, abort, request
+from pytimeparse.timeparse import timeparse
+from werkzeug.serving import make_server
+
+
+# This is the "database" of our dummy LAPI
+class DataStore:
+    def __init__(self) -> None:
+        self.id = 0
+        self.decisions = []
+        self.bouncer_lastpull_by_api_key = {}
+
+    def insert_decisions(self, decisions):
+        for i, _ in enumerate(decisions):
+            decisions[i]["created_at"] = datetime.datetime.now()
+            decisions[i]["deleted_at"] = self.get_decision_expiry_time(decisions[i])
+            decisions[i]["id"] = self.id
+            self.id += 1
+        self.decisions.extend(decisions)
+
+    # This methods can be made more generic by taking lambda expr as input for filtering
+    # decisions to delete
+    def delete_decisions_by_ip(self, ip):
+        for i, decision in enumerate(self.decisions):
+            if ip_address(decision["value"]) == ip_address(ip):
+                self.decisions[i]["deleted_at"] = datetime.datetime.now()
+
+    def delete_decision_by_id(self, id):
+        for i, decision in enumerate(self.decisions):
+            if decision["id"] == id:
+                self.decisions[i]["deleted_at"] = datetime.datetime.now()
+                break
+
+    def update_bouncer_pull(self, api_key):
+        self.bouncer_lastpull_by_api_key[api_key] = datetime.datetime.now()
+
+    def get_active_and_expired_decisions_since(self, since):
+        expired_decisions = []
+        active_decisions = []
+
+        for decision in self.decisions:
+            # decision["deleted_at"] > datetime.datetime.now()  means that decision hasn't yet expired
+            if decision["deleted_at"] > since and decision["deleted_at"] < datetime.datetime.now():
+                expired_decisions.append(decision)
+
+            elif decision["created_at"] > since:
+                active_decisions.append(decision)
+        return active_decisions, expired_decisions
+
+    def get_decisions_for_bouncer(self, api_key, startup=False):
+        if startup or api_key not in self.bouncer_lastpull_by_api_key:
+            since = datetime.datetime.min
+            self.bouncer_lastpull_by_api_key[api_key] = since
+        else:
+            since = self.bouncer_lastpull_by_api_key[api_key]
+
+        self.update_bouncer_pull(api_key)
+        return self.get_active_and_expired_decisions_since(since)
+
+    @staticmethod
+    def get_decision_expiry_time(decision):
+        return decision["created_at"] + timedelta(seconds=timeparse(decision["duration"]))
+
+
+class MockLAPI:
+    def __init__(self) -> None:
+        self.app = Flask(__name__)
+        self.app.add_url_rule("/v1/decisions/stream", view_func=self.decisions)
+        log = logging.getLogger("werkzeug")
+        log.setLevel(logging.ERROR)
+        self.app.logger.disabled = True
+        log.disabled = True
+        self.ds = DataStore()
+
+    def decisions(self):
+        api_key = request.headers.get("x-api-key")
+        if not api_key:
+            abort(404)
+        startup = True if request.args.get("startup") == "true" else False
+        active_decisions, expired_decisions = self.ds.get_decisions_for_bouncer(api_key, startup)
+        return {
+            "new": formatted_decisions(active_decisions),
+            "deleted": formatted_decisions(expired_decisions),
+        }
+
+    def start(self, port=8081):
+        self.server_thread = ServerThread(self.app, port=port)
+        self.server_thread.start()
+
+    def stop(self):
+        self.server_thread.shutdown()
+
+
+def formatted_decisions(decisions):
+    formatted_decisions = []
+    for decision in decisions:
+        expiry_time = decision["created_at"] + timedelta(seconds=timeparse(decision["duration"]))
+        duration = expiry_time - datetime.datetime.now()
+        formatted_decisions.append(
+            {
+                "duration": f"{duration.total_seconds()}s",
+                "id": decision["id"],
+                "origin": decision["origin"],
+                "scenario": "cscli",
+                "scope": decision["scope"],
+                "type": decision["type"],
+                "value": decision["value"],
+            }
+        )
+    return formatted_decisions
+
+
+# Copied from https://stackoverflow.com/a/45017691 .
+# We run server inside thread instead of process to avoid
+# huge complexity of sharing objects
+class ServerThread(Thread):
+    def __init__(self, app, port=8081):
+        Thread.__init__(self)
+        self.server = make_server("127.0.0.1", port, app)
+        self.ctx = app.app_context()
+        self.ctx.push()
+
+    def run(self):
+        self.server.serve_forever()
+
+    def shutdown(self):
+        self.server.shutdown()
+
+
+if __name__ == "__main__":
+    MockLAPI().start()
+    sleep(100)

tests/stdinmode/crowdsec-custom-bouncer.yaml

@@ -0,0 +1,20 @@
+bin_path: tests/stdinmode/custombinary
+feed_via_stdin: true # Invokes binary once and feeds incoming decisions to it's stdin.
+total_retries: 2 # number of times to restart binary. relevant if feed_via_stdin=true . Set to -1 for infinite retries.
+scenarios_containing: [] # ignore IPs banned for triggering scenarios not containing either of provided word, eg ["ssh", "http"]
+scenarios_not_containing: [] # ignore IPs banned for triggering scenarios containing either of provided word
+origins: []
+piddir: /var/run/
+update_frequency: 0.1s
+cache_retention_duration: 10s
+daemonize: false
+log_mode: stdout
+log_dir: /var/log/
+log_level: info
+api_url: http://localhost:8081/
+api_key: 1237adaf7a1724ac68a3288828820a67
+
+prometheus:
+  enabled: false
+  listen_addr: 127.0.0.1
+  listen_port: 60602 

tests/stdinmode/custombinary

@@ -0,0 +1,7 @@
+#!/bin/bash 
+rm data.txt > /dev/null
+while read line
+do
+  echo "$line" >> data.txt
+done
+# rm data.txt > /dev/null