feat(*): Autonomous mode

.github/workflows/update-js.yml

@@ -1,9 +1,10 @@
 name: Generate js
 
-on: 
+on:
   push:
     paths:
       - 'pkg/cloudflare/worker/**'
+      - 'pkg/cloudflare/decisions-sync-worker/**'
       - '.github/workflows/update-js.yml'
 jobs:
   build:
@@ -21,7 +22,7 @@ jobs:
       run: |
         git config --local user.email "[email protected]"
         git config --local user.name "GitHub Action"
-        git commit -m "Update dist js" pkg/cloudflare/worker/dist/main.js || exit 0
+        git commit -m "Update dist js" pkg/cloudflare/worker/dist/main.js pkg/cloudflare/decisions-sync-worker/dist/main.js || exit 0
 #token to expire on 09/19/2024
     - name: Push changes
       if: ${{ github.event_name == 'push'}}

.gitignore

@@ -17,6 +17,7 @@ venv/
 
 crowdsec-cloudflare-worker-bouncer
 pkg/cloudflare/worker/node_modules
+pkg/cloudflare/decisions-sync-worker/node_modules
 
 # built by make
 /crowdsec-cloudflare-worker-bouncer
@@ -36,3 +37,9 @@ pkg/cloudflare/worker/node_modules
 /rpm/RPMS
 /rpm/SOURCES/*.tar.gz
 /rpm/SRPMS
+
+# Claude
+.claude
+
+# Dev
+dev/

Makefile

@@ -61,7 +61,10 @@ binary:
 
 .PHONY: build-worker-js
 build-worker-js:
+	@echo "Building bouncer worker..."
 	@cd pkg/cloudflare/worker && npm install && npm run build
+	@echo "Building decisions-sync-worker..."
+	@cd pkg/cloudflare/decisions-sync-worker && npm install && npm run build
 
 .PHONY: build-all
 build-all: clean build-worker-js binary

cmd/root.go

@@ -209,7 +209,7 @@ func getConfigFromPath(configPath string) (*cfg.BouncerConfig, error) {
 func CloudflareManagersFromConfig(ctx context.Context, config cfg.CloudflareConfig) ([]*cf.CloudflareAccountManager, error) {
 	cfManagers := make([]*cf.CloudflareAccountManager, 0, len(config.Accounts))
 	for _, accountCfg := range config.Accounts {
-		manager, err := cf.NewCloudflareManager(ctx, accountCfg, &config.Worker)
+		manager, err := cf.NewCloudflareManager(ctx, accountCfg, &config.Worker, config.IgnoreBindingErrorsOnDeploy)
 		if err != nil {
 			return nil, fmt.Errorf("unable to create cloudflare manager: %w", err)
 		}
@@ -218,39 +218,115 @@ func CloudflareManagersFromConfig(ctx context.Context, config cfg.CloudflareConf
 	return cfManagers, nil
 }
 
-func Execute(configTokens *string, configOutputPath *string, configPath *string, ver *bool, testConfig *bool, showConfig *bool, deleteOnly *bool, setupOnly *bool) error {
-	if ver != nil && *ver {
-		fmt.Fprint(os.Stdout, version.FullString())
-		return nil
-	}
-
-	if configPath == nil || *configPath == "" {
-		configPath = new(string)
-		*configPath = DEFAULT_CONFIG_PATH
+func handleConfigTokens(configTokens, configOutputPath, configPath string) error {
+	cfgTokenString, err := cfg.ConfigTokens(configTokens, configPath)
+	if err != nil {
+		return err
 	}
-
-	if configTokens != nil && *configTokens != "" {
-		cfgTokenString, err := cfg.ConfigTokens(*configTokens, *configPath)
+	if configOutputPath != "" {
+		err := os.WriteFile(configOutputPath, []byte(cfgTokenString), 0o664)
 		if err != nil {
 			return err
 		}
-		if configOutputPath != nil && *configOutputPath != "" {
-			err := os.WriteFile(*configOutputPath, []byte(cfgTokenString), 0o664)
+		log.Printf("Config successfully generated in %s", configOutputPath)
+	} else {
+		fmt.Fprint(os.Stdout, cfgTokenString)
+	}
+	return nil
+}
+
+func deployInfraForManagers(g *errgroup.Group, cfManagers []*cf.CloudflareAccountManager, deleteOnly, setupAutonomous bool, crowdSecConfig cfg.CrowdSecConfig, cronSchedule string) {
+	for _, cfManager := range cfManagers {
+		manager := cfManager
+		g.Go(func() error {
+			err := manager.CleanUpExistingWorkers(true)
 			if err != nil {
-				return err
+				return fmt.Errorf("unable to cleanup existing workers: %w for account %s", err, manager.AccountCfg.Name)
 			}
-			log.Printf("Config successfully generated in %s", *configOutputPath)
-		} else {
-			fmt.Fprint(os.Stdout, cfgTokenString)
+			if deleteOnly {
+				return nil
+			}
+			if err := manager.DeployInfra(); err != nil {
+				return fmt.Errorf("unable to deploy infra: %w for account %s", err, manager.AccountCfg.Name)
+			}
+			log.Infof("Successfully deployed infra for account %s", manager.AccountCfg.Name)
+
+			// Deploy decisions sync worker if autonomous setup (-S flag)
+			if setupAutonomous {
+				log.Infof("Autonomous setup mode, deploying decisions sync worker for account %s", manager.AccountCfg.Name)
+				if err := manager.DeployDecisionsSyncWorker(crowdSecConfig, cronSchedule); err != nil {
+					return fmt.Errorf("unable to deploy decisions sync worker: %w for account %s", err, manager.AccountCfg.Name)
+				}
+			}
+
+			return nil
+		})
+	}
+}
+
+func startPrometheusServer(g *errgroup.Group, ctx context.Context, conf *cfg.BouncerConfig, mHandler *metricsHandler) {
+	if !conf.PrometheusConfig.Enabled {
+		return
+	}
+	mux := http.NewServeMux()
+	mux.Handle("/metrics", mHandler.computeMetricsHandler(promhttp.Handler()))
+	server := &http.Server{
+		Addr:    net.JoinHostPort(conf.PrometheusConfig.ListenAddress, conf.PrometheusConfig.ListenPort),
+		Handler: mux,
+	}
+	g.Go(func() error {
+		if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+			return err
 		}
 		return nil
+	})
+	g.Go(func() error {
+		<-ctx.Done()
+		log.Info("Shutting down Prometheus HTTP server")
+		// Use a fresh context for shutdown since ctx is already canceled
+		shutdownCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
+		defer cancel()
+		return server.Shutdown(shutdownCtx)
+	})
+}
+
+// ExecuteOptions holds configuration options for the Execute function.
+type ExecuteOptions struct {
+	ConfigTokens     *string
+	ConfigOutputPath *string
+	ConfigPath       *string
+	Ver              *bool
+	TestConfig       *bool
+	ShowConfig       *bool
+	DeleteOnly       *bool
+	SetupOnly        *bool
+	SetupAutonomous  *bool
+}
+
+func Execute(opts ExecuteOptions) error {
+	if opts.Ver != nil && *opts.Ver {
+		fmt.Fprint(os.Stdout, version.FullString())
+		return nil
+	}
+
+	if opts.ConfigPath == nil || *opts.ConfigPath == "" {
+		opts.ConfigPath = new(string)
+		*opts.ConfigPath = DEFAULT_CONFIG_PATH
+	}
+
+	if opts.ConfigTokens != nil && *opts.ConfigTokens != "" {
+		outputPath := ""
+		if opts.ConfigOutputPath != nil {
+			outputPath = *opts.ConfigOutputPath
+		}
+		return handleConfigTokens(*opts.ConfigTokens, outputPath, *opts.ConfigPath)
 	}
 
-	conf, err := getConfigFromPath(*configPath)
+	conf, err := getConfigFromPath(*opts.ConfigPath)
 	if err != nil {
 		return err
 	}
-	if showConfig != nil && *showConfig {
+	if opts.ShowConfig != nil && *opts.ShowConfig {
 		fmt.Fprintf(os.Stdout, "%+v", conf)
 		return nil
 	}
@@ -271,13 +347,13 @@ func Execute(configTokens *string, configOutputPath *string, configPath *string,
 		CAPath:   conf.CrowdSecConfig.CAPath,
 	}
 
-	if (testConfig != nil && *testConfig) || (setupOnly == nil || !*setupOnly) || (deleteOnly == nil || !*deleteOnly) {
+	if (opts.TestConfig != nil && *opts.TestConfig) || (opts.SetupOnly == nil || !*opts.SetupOnly) || (opts.DeleteOnly == nil || !*opts.DeleteOnly) {
 		if err := csLAPI.Init(); err != nil {
 			return fmt.Errorf("unable to initialize crowdsec bouncer: %w", err)
 		}
 	}
 
-	if testConfig != nil && *testConfig {
+	if opts.TestConfig != nil && *opts.TestConfig {
 		log.Info("config is valid")
 		return nil
 	}
@@ -288,31 +364,18 @@ func Execute(configTokens *string, configOutputPath *string, configPath *string,
 	if err != nil {
 		return err
 	}
-	for _, cfManager := range cfManagers {
-		manager := cfManager
-		g.Go(func() error {
-			err := manager.CleanUpExistingWorkers(true)
-			if err != nil {
-				return fmt.Errorf("unable to cleanup existing workers: %w for account %s", err, manager.AccountCfg.Name)
-			}
-			if deleteOnly != nil && *deleteOnly {
-				return nil
-			}
-			if err := manager.DeployInfra(); err != nil {
-				return fmt.Errorf("unable to deploy infra: %w for account %s", err, manager.AccountCfg.Name)
-			}
-			log.Infof("Successfully deployed infra for account %s", manager.AccountCfg.Name)
-			return nil
-		})
-	}
+
+	isDeleteOnly := opts.DeleteOnly != nil && *opts.DeleteOnly
+	isSetupAutonomous := opts.SetupAutonomous != nil && *opts.SetupAutonomous
+	deployInfraForManagers(g, cfManagers, isDeleteOnly, isSetupAutonomous, conf.CrowdSecConfig, conf.CloudflareConfig.DecisionsSyncWorker.Cron)
 	if err := g.Wait(); err != nil {
 		return err
 	}
-	if deleteOnly != nil && *deleteOnly {
+	if opts.DeleteOnly != nil && *opts.DeleteOnly {
 		return nil
 	}
 	log.Info("Successfully deployed infra for all accounts")
-	if setupOnly != nil && *setupOnly {
+	if (opts.SetupOnly != nil && *opts.SetupOnly) || (opts.SetupAutonomous != nil && *opts.SetupAutonomous) {
 		return nil
 	}
 
@@ -335,10 +398,6 @@ func Execute(configTokens *string, configOutputPath *string, configPath *string,
 		return HandleSignals(ctx)
 	})
 
-	g.Go(func() error {
-		return csLAPI.Run(ctx)
-	})
-
 	mHandler := metricsHandler{
 		cfManagers: cfManagers,
 	}
@@ -354,12 +413,15 @@ func Execute(configTokens *string, configOutputPath *string, configPath *string,
 
 	prometheus.MustRegister(csbouncer.TotalLAPICalls, csbouncer.TotalLAPIError, metrics.CloudflareAPICallsByAccount, metrics.TotalKeysByAccount,
 		metrics.TotalActiveDecisions, metrics.TotalBlockedRequests, metrics.TotalProcessedRequests)
-	if conf.PrometheusConfig.Enabled {
-		g.Go(func() error {
-			http.Handle("/metrics", mHandler.computeMetricsHandler(promhttp.Handler()))
-			return http.ListenAndServe(net.JoinHostPort(conf.PrometheusConfig.ListenAddress, conf.PrometheusConfig.ListenPort), nil)
-		})
-	}
+	startPrometheusServer(g, ctx, conf, &mHandler)
+
+	// Process decisions from CrowdSec LAPI
+	g.Go(func() error {
+		if err := csLAPI.Run(ctx); err != nil {
+			return fmt.Errorf("crowdsec bouncer error: %w", err)
+		}
+		return errors.New("crowdsec bouncer stopped")
+	})
 
 	for {
 		select {

cmd/root_test.go

@@ -305,7 +305,10 @@ func TestBouncer(t *testing.T) {
 
 	// generate config
 	configPath := "/tmp/crowdsec-cloudflare-worker-bouncer.yaml"
-	if err := Execute(&cloudflareToken, &configPath, nil, nil, nil, nil, nil, nil); err != nil {
+	if err := Execute(ExecuteOptions{
+		ConfigTokens:     &cloudflareToken,
+		ConfigOutputPath: &configPath,
+	}); err != nil {
 		t.Fatal(err)
 	}
 

main.go

@@ -17,8 +17,19 @@ func main() {
 	showConfig := flag.Bool("T", false, "show full config (.yaml + .yaml.local) and exit")
 	deleteOnly := flag.Bool("d", false, "delete all the created infra and exit")
 	setupOnly := flag.Bool("s", false, "setup the infra and exit")
+	setupAutonomous := flag.Bool("S", false, "setup the infra in autonomous mode (with decisions-sync-worker) and exit")
 	flag.Parse()
-	err := cmd.Execute(configTokens, configOutputPath, configPath, ver, testConfig, showConfig, deleteOnly, setupOnly)
+	err := cmd.Execute(cmd.ExecuteOptions{
+		ConfigTokens:     configTokens,
+		ConfigOutputPath: configOutputPath,
+		ConfigPath:       configPath,
+		Ver:              ver,
+		TestConfig:       testConfig,
+		ShowConfig:       showConfig,
+		DeleteOnly:       deleteOnly,
+		SetupOnly:        setupOnly,
+		SetupAutonomous:  setupAutonomous,
+	})
 	if err != nil {
 		log.Fatal(err)
 	}