appsec: release the listeners when one of them fails

pkg/acquisition/modules/appsec/run.go

@@ -25,6 +25,13 @@ func (w *Source) listenAndServe(ctx context.Context, t *tomb.Tomb) error {
 	serverError := make(chan error, 2)
 
 	startServer := func(listener net.Listener, canTLS bool) {
+		// The listener is this goroutine's to close. http.Server closes the ones
+		// it tracks, but it never gets to track this one when the TLS config is
+		// rejected below or by ServeTLS itself (a cert that fails to load returns
+		// before the listener reaches Serve), and an untracked listener is one
+		// Shutdown cannot release either.
+		defer listener.Close()
+
 		var err error
 
 		if canTLS && (w.config.CertFilePath != "" || w.config.KeyFilePath != "") {
@@ -93,34 +100,45 @@ func (w *Source) listenAndServe(ctx context.Context, t *tomb.Tomb) error {
 		startServer(listener, true)
 	}(w.config.ListenAddr)
 
+	// Whichever way this returns, the server goes with it. One listener failing
+	// to bind used to return straight away and leave the other one serving: the
+	// port stayed held for the lifetime of the process, and because nothing
+	// rebinds, every later reload failed on an address that was still in use
+	// while the rest of the process carried on looking healthy.
+	defer w.shutdownServer(ctx)
+
 	select {
 	case err := <-serverError:
 		return err
 	case <-t.Dying():
-		w.logger.Info("Shutting down Appsec server")
-		// xx let's clean up the appsec runners :)
-		appsec.AppsecRulesDetails = make(map[int]appsec.RulesDetails)
+		return nil
+	}
+}
 
-		if err := w.server.Shutdown(ctx); err != nil {
-			w.logger.Errorf("Error shutting down Appsec server: %s", err.Error())
-		}
+// shutdownServer stops the HTTP server and releases what the listeners hold.
+func (w *Source) shutdownServer(ctx context.Context) {
+	w.logger.Info("Shutting down Appsec server")
 
-		if w.AppsecRuntime != nil && w.AppsecRuntime.ChallengeRuntime != nil {
-			if err := w.AppsecRuntime.ChallengeRuntime.Close(ctx); err != nil {
-				w.logger.Errorf("Error closing challenge runtime: %s", err)
-			}
+	// xx let's clean up the appsec runners :)
+	appsec.AppsecRulesDetails = make(map[int]appsec.RulesDetails)
+
+	if err := w.server.Shutdown(ctx); err != nil {
+		w.logger.Errorf("Error shutting down Appsec server: %s", err.Error())
+	}
+
+	if w.AppsecRuntime != nil && w.AppsecRuntime.ChallengeRuntime != nil {
+		if err := w.AppsecRuntime.ChallengeRuntime.Close(ctx); err != nil {
+			w.logger.Errorf("Error closing challenge runtime: %s", err)
 		}
+	}
 
-		if w.config.ListenSocket != "" {
-			if err := os.Remove(w.config.ListenSocket); err != nil {
-				if !errors.Is(err, fs.ErrNotExist) {
-					w.logger.Errorf("can't remove socket %s: %s", w.config.ListenSocket, err)
-				}
+	if w.config.ListenSocket != "" {
+		if err := os.Remove(w.config.ListenSocket); err != nil {
+			if !errors.Is(err, fs.ErrNotExist) {
+				w.logger.Errorf("can't remove socket %s: %s", w.config.ListenSocket, err)
 			}
 		}
 	}
-
-	return nil
 }
 
 func (w *Source) StreamingAcquisition(ctx context.Context, out chan pipeline.Event, t *tomb.Tomb) error {

pkg/acquisition/modules/appsec/run_test.go

@@ -0,0 +1,190 @@
+package appsecacquisition
+
+import (
+	"context"
+	"net"
+	"net/http"
+	"path/filepath"
+	"testing"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/require"
+	"gopkg.in/tomb.v2"
+)
+
+// freeAddr returns a loopback address that was bindable a moment ago.
+func freeAddr(t *testing.T) string {
+	t.Helper()
+
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+
+	addr := l.Addr().String()
+	require.NoError(t, l.Close())
+
+	return addr
+}
+
+func newTestSource(t *testing.T, cfg Configuration) *Source {
+	t.Helper()
+
+	w := &Source{
+		logger: log.NewEntry(log.StandardLogger()),
+		mux:    http.NewServeMux(),
+		config: cfg,
+	}
+	w.server = &http.Server{Addr: cfg.ListenAddr, Handler: w.mux}
+
+	return w
+}
+
+// runListenAndServe runs listenAndServe to completion and returns its error.
+func runListenAndServe(t *testing.T, w *Source) error {
+	t.Helper()
+
+	var tb tomb.Tomb
+
+	done := make(chan error, 1)
+
+	tb.Go(func() error {
+		done <- w.listenAndServe(context.Background(), &tb)
+		return nil
+	})
+
+	defer func() {
+		tb.Kill(nil)
+		_ = tb.Wait()
+	}()
+
+	select {
+	case err := <-done:
+		return err
+	case <-time.After(30 * time.Second):
+		t.Fatal("listenAndServe did not return")
+		return nil
+	}
+}
+
+// requireAddrFree fails unless addr can be bound again.
+func requireAddrFree(t *testing.T, addr string) {
+	t.Helper()
+
+	// the goroutine that owns the listener releases it as it unwinds
+	require.Eventually(t, func() bool {
+		l, err := net.Listen("tcp", addr)
+		if err != nil {
+			return false
+		}
+
+		l.Close()
+
+		return true
+	}, 10*time.Second, 50*time.Millisecond, "%s is still bound after listenAndServe returned", addr)
+}
+
+// A listener that fails to bind must not leave the other one holding its port.
+// Before this was fixed, listenAndServe returned the error without shutting the
+// server down, so the surviving listener kept the port for the lifetime of the
+// process and every later reload failed with "address already in use" while the
+// rest of CrowdSec carried on looking healthy.
+func TestListenAndServeReleasesTCPWhenTheSocketFails(t *testing.T) {
+	addr := freeAddr(t)
+
+	w := newTestSource(t, Configuration{
+		ListenAddr: addr,
+		// a directory that does not exist, so the bind fails
+		ListenSocket: filepath.Join(t.TempDir(), "absent", "appsec.sock"),
+	})
+
+	require.Error(t, runListenAndServe(t, w))
+	requireAddrFree(t, addr)
+}
+
+// The mirror of the above: the TCP bind fails and the socket must be released.
+func TestListenAndServeReleasesSocketWhenTCPFails(t *testing.T) {
+	// hold the address so the appsec listener cannot have it
+	blocker, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+
+	defer blocker.Close()
+
+	socket := filepath.Join(t.TempDir(), "appsec.sock")
+
+	w := newTestSource(t, Configuration{
+		ListenAddr:   blocker.Addr().String(),
+		ListenSocket: socket,
+	})
+
+	require.Error(t, runListenAndServe(t, w))
+
+	// the socket file is removed on the way out, so it can be bound again
+	require.Eventually(t, func() bool {
+		l, err := net.Listen("unix", socket)
+		if err != nil {
+			return false
+		}
+
+		l.Close()
+
+		return true
+	}, 10*time.Second, 50*time.Millisecond, "%s was not released", socket)
+}
+
+// A TLS listener that never reaches Serve is one the server never tracks, so
+// Shutdown cannot release it either - the goroutine that opened it has to.
+func TestListenAndServeReleasesTCPWhenTLSIsMisconfigured(t *testing.T) {
+	for name, cfg := range map[string]Configuration{
+		"cert file that does not exist": {
+			CertFilePath: filepath.Join(t.TempDir(), "absent.pem"),
+			KeyFilePath:  filepath.Join(t.TempDir(), "absent.key"),
+		},
+		"key file not configured": {
+			CertFilePath: filepath.Join(t.TempDir(), "absent.pem"),
+		},
+		"cert file not configured": {
+			KeyFilePath: filepath.Join(t.TempDir(), "absent.key"),
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			cfg.ListenAddr = freeAddr(t)
+			w := newTestSource(t, cfg)
+
+			require.Error(t, runListenAndServe(t, w))
+			requireAddrFree(t, cfg.ListenAddr)
+		})
+	}
+}
+
+// The ordinary path still works: a clean shutdown releases the port.
+func TestListenAndServeReleasesTCPOnShutdown(t *testing.T) {
+	addr := freeAddr(t)
+	w := newTestSource(t, Configuration{ListenAddr: addr})
+
+	var tb tomb.Tomb
+
+	done := make(chan error, 1)
+
+	tb.Go(func() error {
+		done <- w.listenAndServe(context.Background(), &tb)
+		return nil
+	})
+
+	// wait until it is actually serving
+	require.Eventually(t, func() bool {
+		c, err := net.Dial("tcp", addr)
+		if err != nil {
+			return false
+		}
+
+		c.Close()
+
+		return true
+	}, 10*time.Second, 50*time.Millisecond, "appsec never started listening on %s", addr)
+
+	tb.Kill(nil)
+	require.NoError(t, <-done)
+	require.NoError(t, tb.Wait())
+
+	requireAddrFree(t, addr)
+}