Rename the program to sww and its deciding daemon to discriminator

The program is now sww, built from cmd/sww, with two modes: sww rproxy and
sww discriminator. servewaf becomes discriminator everywhere it names that
daemon: the package (internal/discriminator), its types, functions, files, the
discriminator_line log field, and its docs and tests.

smallwebwaf is left only where sneak's directive allows it: the documents'
product name, the git.eeqj.de/sneak/smallwebwaf URLs, and the Go module path
sneak.berlin/go/smallwebwaf. Everywhere else a name containing it takes the sww
form: the package (internal/sww), metric names (sww_*), the X-Sww-* headers,
the /_sww/ admin endpoints, the OS user, and the state and rules directories.
The SWWAF_ environment variables take the SWW_ form.

No logic changes. make check passes.

Model: deepseek-v4-pro

internal/sww/discriminator_test.go

@@ -0,0 +1,204 @@
+package sww_test
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"io"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"syscall"
+	"testing"
+	"time"
+
+	"sneak.berlin/go/smallwebwaf/internal/discriminator"
+	"sneak.berlin/go/smallwebwaf/internal/sww"
+)
+
+// commandForMain, while set in the environment, has this test binary run Main
+// with the command it gives, in place of the tests, so that a test can run
+// Main in a process of its own.
+const commandForMain = "SWW_TEST_COMMAND"
+
+func TestMain(m *testing.M) {
+	command, set := os.LookupEnv(commandForMain)
+	if !set {
+		os.Exit(m.Run())
+	}
+
+	os.Args = strings.Fields("sww " + command)
+
+	os.Exit(sww.Main(testVersion))
+}
+
+func TestMainRunsDiscriminator(t *testing.T) {
+	t.Parallel()
+
+	binary, err := os.Executable()
+	if err != nil {
+		t.Fatalf("find the test binary: %v", err)
+	}
+
+	ctx, cancel := context.WithTimeout(t.Context(), 3*waitLimit)
+	defer cancel()
+
+	out := &output{}
+	cmd := exec.CommandContext(ctx, binary) //nolint:gosec // this test binary
+	cmd.Stdout = out
+
+	cmd.Env = append(os.Environ(),
+		commandForMain+"="+discriminatorCommand,
+		listenAddr+"="+localhost+":0",
+		stateDir+"="+t.TempDir(),
+		rulesDir+"="+t.TempDir(),
+		lookupSource+"=off")
+
+	err = cmd.Start()
+	if err != nil {
+		t.Fatalf("start: %v", err)
+	}
+
+	t.Cleanup(func() {
+		_ = cmd.Process.Kill()
+		_ = cmd.Wait()
+	})
+
+	addr, _ := out.line(t, "msg", "starting")["address"].(string)
+	ask(t, addr, "/", http.StatusOK, discriminator.DispositionAllow)
+
+	err = cmd.Process.Signal(syscall.SIGTERM)
+	if err != nil {
+		t.Fatalf("signal: %v", err)
+	}
+
+	err = cmd.Wait()
+	if err != nil {
+		t.Errorf("exit: %v; output:\n%s", err, out.text())
+	}
+}
+
+func TestMainWithoutACommandItKnowsPrintsTheCommands(t *testing.T) {
+	t.Parallel()
+
+	binary, err := os.Executable()
+	if err != nil {
+		t.Fatalf("find the test binary: %v", err)
+	}
+
+	for _, command := range []string{"", "serve"} {
+		t.Run("command "+command, func(t *testing.T) {
+			t.Parallel()
+
+			ctx, cancel := context.WithTimeout(t.Context(), waitLimit)
+			defer cancel()
+
+			var stdout, stderr bytes.Buffer
+
+			cmd := exec.CommandContext(ctx, binary)
+			cmd.Stdout, cmd.Stderr = &stdout, &stderr
+
+			cmd.Env = append(os.Environ(), commandForMain+"="+command)
+
+			err := cmd.Run()
+
+			if code := cmd.ProcessState.ExitCode(); code != 2 {
+				t.Errorf("exit status %d (%v), want 2", code, err)
+			}
+
+			for _, name := range []string{"rproxy", discriminatorCommand, "healthcheck"} {
+				if !strings.Contains(stderr.String(), name) {
+					t.Errorf("stderr %q does not name %s", stderr.String(), name)
+				}
+			}
+
+			if stdout.Len() != 0 {
+				t.Errorf("stdout %q, want nothing", stdout.String())
+			}
+		})
+	}
+}
+
+// ask sends discriminator at addr rproxy's question about a request for path from
+// placed, checks the answer's status and disposition, and closes it.
+func ask(t *testing.T, addr, path string, status int, disposition string) {
+	t.Helper()
+
+	req, err := http.NewRequestWithContext(t.Context(), http.MethodGet,
+		"http://"+addr+path, http.NoBody)
+	if err != nil {
+		t.Fatalf("new request: %v", err)
+	}
+
+	req.Header.Set(discriminator.ClientHeader, placed)
+	req.Header.Set("X-Request-ID", rand.Text())
+
+	transport := &http.Transport{}
+	defer transport.CloseIdleConnections()
+
+	res, err := (&http.Client{Transport: transport}).Do(req)
+	if err != nil {
+		t.Fatalf("question: %v", err)
+	}
+
+	_, _ = io.Copy(io.Discard, res.Body)
+	_ = res.Body.Close()
+
+	if res.StatusCode != status ||
+		res.Header.Get(discriminator.DispositionHeader) != disposition {
+		t.Errorf("answered %d with disposition %q, want %d with %q", res.StatusCode,
+			res.Header.Get(discriminator.DispositionHeader), status, disposition)
+	}
+}
+
+// runDiscriminator runs `sww discriminator` with the settings in env, has use
+// send it requests at the address it listens on, stops it as SIGTERM does, and
+// checks that it stopped in order.
+func runDiscriminator(t *testing.T, env map[string]string, use func(addr string)) {
+	t.Helper()
+
+	ctx, stop := context.WithCancel(t.Context())
+	out := &output{}
+	exited := make(chan int, 1)
+
+	go func() {
+		exited <- run(ctx, env, out)
+	}()
+
+	addr, _ := out.line(t, "msg", "starting")["address"].(string)
+	use(addr)
+	stop()
+
+	select {
+	case status := <-exited:
+		if status != 0 {
+			t.Fatalf("exit status %d, want 0; output:\n%s", status, out.text())
+		}
+	case <-time.After(waitLimit):
+		t.Fatal("still running after being told to stop")
+	}
+
+	out.line(t, "msg", "stopped")
+}
+
+func TestDiscriminatorKeepsItsBansAcrossARestartAndNeedsNoApp(t *testing.T) {
+	t.Parallel()
+
+	env := map[string]string{
+		listenAddr: localhost + ":0",
+		stateDir:   t.TempDir(),
+		rulesDir:   filepath.Join("..", "..", "share", "rules.d"),
+	}
+
+	runDiscriminator(t, env, func(addr string) {
+		ask(t, addr, "/", http.StatusOK, discriminator.DispositionAllow)
+		// The default rule file bans a probe for this.
+		ask(t, addr, "/.env", http.StatusForbidden, discriminator.DispositionDeny)
+	})
+
+	runDiscriminator(t, env, func(addr string) {
+		ask(t, addr, "/", http.StatusForbidden, discriminator.DispositionDeny)
+	})
+}

internal/sww/healthcheck.go

@@ -0,0 +1,145 @@
+package sww
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/url"
+	"time"
+
+	"sneak.berlin/go/smallwebwaf/internal/config"
+	"sneak.berlin/go/smallwebwaf/internal/discriminator"
+)
+
+// healthCheckTimeout bounds the whole health check.
+const healthCheckTimeout = 5 * time.Second
+
+var errHealthEndpoint = errors.New("the health endpoint answered")
+
+// HealthCheck is the container's health check. In discriminator's container, where
+// SWW_COMMAND is discriminator, it returns 0 while discriminator answers its health
+// endpoint on 127.0.0.1, at the port in SWW_LISTEN_ADDR. In rproxy's container,
+// where SWW_COMMAND is rproxy, unset or empty, it returns 0 while rproxy
+// answers its health endpoint on 127.0.0.1, at the port in SWRP_LISTEN_ADDR, and
+// the app accepts connections at the address in SWRP_UPSTREAM_URL. Otherwise,
+// any other value of SWW_COMMAND included, it writes why to stderr and
+// returns 1. It reads no other setting, nor a file that another names, so neither
+// can fail it.
+// args are the arguments after `healthcheck`; it takes none, and given
+// one it names it on stderr and returns 1 without checking anything.
+func HealthCheck(
+	ctx context.Context, args []string, lookupEnv func(string) (string, bool),
+	stderr io.Writer,
+) int {
+	if len(args) > 0 {
+		_, _ = fmt.Fprintf(stderr,
+			"sww healthcheck: unexpected argument %q\n", args[0])
+
+		return 1
+	}
+
+	ctx, cancel := context.WithTimeout(ctx, healthCheckTimeout)
+	defer cancel()
+
+	command, err := config.Command(lookupEnv)
+
+	switch {
+	case err != nil:
+		err = fmt.Errorf("invalid setting: %w", err)
+	case command == "discriminator":
+		err = checkDiscriminator(ctx, lookupEnv)
+	default:
+		err = checkRProxyAndApp(ctx, lookupEnv)
+	}
+
+	if err != nil {
+		_, _ = fmt.Fprintln(stderr, "unhealthy:", err)
+
+		return 1
+	}
+
+	return 0
+}
+
+// checkDiscriminator asks discriminator's health endpoint.
+func checkDiscriminator(
+	ctx context.Context, lookupEnv func(string) (string, bool),
+) error {
+	listenAddr, err := config.DiscriminatorListenAddr(lookupEnv)
+	if err != nil {
+		return fmt.Errorf("invalid setting: %w", err)
+	}
+
+	err = askHealthEndpoint(ctx, listenAddr)
+	if err != nil {
+		return fmt.Errorf("ask discriminator: %w", err)
+	}
+
+	return nil
+}
+
+// checkRProxyAndApp asks rproxy's health endpoint, and connects to the app.
+func checkRProxyAndApp(
+	ctx context.Context, lookupEnv func(string) (string, bool),
+) error {
+	listenAddr, upstreamURL, err := config.RProxyListenAddrAndUpstreamURL(lookupEnv)
+	if err != nil {
+		return fmt.Errorf("invalid setting: %w", err)
+	}
+
+	err = askHealthEndpoint(ctx, listenAddr)
+	if err != nil {
+		return fmt.Errorf("ask rproxy: %w", err)
+	}
+
+	conn, err := (&net.Dialer{}).DialContext(ctx, "tcp", appAddress(upstreamURL))
+	if err != nil {
+		return fmt.Errorf("connect to the app: %w", err)
+	}
+
+	_ = conn.Close()
+
+	return nil
+}
+
+// askHealthEndpoint asks the health endpoint on 127.0.0.1 at the port of
+// listenAddr, which the settings have checked has one.
+func askHealthEndpoint(ctx context.Context, listenAddr string) error {
+	_, port, _ := net.SplitHostPort(listenAddr)
+	health := "http://" + net.JoinHostPort("127.0.0.1", port) + discriminator.HealthPath
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, health, http.NoBody)
+	if err != nil {
+		return fmt.Errorf("make the request: %w", err)
+	}
+
+	res, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return fmt.Errorf("send the request: %w", err)
+	}
+
+	_ = res.Body.Close()
+
+	if res.StatusCode != http.StatusOK {
+		return fmt.Errorf("%w %s", errHealthEndpoint, res.Status)
+	}
+
+	return nil
+}
+
+// appAddress is the host and port of the app's URL, the port being the
+// scheme's own when the URL names none.
+func appAddress(app *url.URL) string {
+	port := app.Port()
+	if port == "" {
+		port = "80"
+		if app.Scheme == "https" {
+			port = "443"
+		}
+	}
+
+	return net.JoinHostPort(app.Hostname(), port)
+}

internal/sww/healthcheck_internal_test.go

@@ -0,0 +1,27 @@
+package sww
+
+import (
+	"net/url"
+	"testing"
+)
+
+func TestAppAddress(t *testing.T) {
+	t.Parallel()
+
+	for app, want := range map[string]string{
+		"http://127.0.0.1:8081": "127.0.0.1:8081",
+		"http://app":            "app:80",
+		"https://app/":          "app:443",
+		"https://[::1]":         "[::1]:443",
+	} {
+		parsed, err := url.Parse(app)
+		if err != nil {
+			t.Fatalf("parse %q: %v", app, err)
+		}
+
+		got := appAddress(parsed)
+		if got != want {
+			t.Errorf("appAddress(%q) is %q, want %q", app, got, want)
+		}
+	}
+}

internal/sww/healthcheck_test.go

@@ -0,0 +1,196 @@
+package sww_test
+
+import (
+	"bytes"
+	"context"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"sneak.berlin/go/smallwebwaf/internal/sww"
+)
+
+const (
+	// commandSetting says which command a container runs, and which the health
+	// check checks.
+	commandSetting = "SWW_COMMAND"
+	// discriminatorCommand is the command that runs discriminator.
+	discriminatorCommand = "discriminator"
+)
+
+func TestHealthCheckInDiscriminatorsContainer(t *testing.T) {
+	t.Parallel()
+
+	ctx, stop := context.WithCancel(t.Context())
+	defer stop()
+
+	out := &output{}
+	exited := make(chan int, 1)
+	settings := map[string]string{
+		listenAddr: localhost + ":0",
+		stateDir:   t.TempDir(),
+		rulesDir:   t.TempDir(),
+	}
+
+	go func() {
+		exited <- run(ctx, settings, out)
+	}()
+
+	addr, _ := out.line(t, "msg", "starting")["address"].(string)
+	_, port, _ := net.SplitHostPort(addr)
+
+	// The container's settings: an address to listen on with an empty
+	// host part, which the health check asks at 127.0.0.1.
+	wantHealthCheck(t, map[string]string{
+		commandSetting: discriminatorCommand, listenAddr: ":" + port,
+	}, 0, "")
+
+	// The health check reads that setting alone, here given as a file: a removed
+	// or invalid token file, or an invalid value of another setting, does not
+	// fail it.
+	for _, other := range []struct{ name, value string }{
+		{"SWW_METRICS_TOKEN_FILE", filepath.Join(t.TempDir(), "removed")},
+		{"SWW_METRICS_TOKEN_FILE", writeFile(t, "too short\n")},
+		{"SWW_MODE", "neither"},
+	} {
+		wantHealthCheck(t, map[string]string{
+			commandSetting:       discriminatorCommand,
+			listenAddr + "_FILE": writeFile(t, ":"+port+"\n"),
+			other.name:           other.value,
+		}, 0, "")
+	}
+
+	// Listening on port 0 takes any free port, so nothing listens on port 0
+	// itself.
+	wantHealthCheck(t, map[string]string{
+		commandSetting: discriminatorCommand, listenAddr: ":0",
+	}, 1, "unhealthy: ask discriminator: ")
+	wantHealthCheck(t, map[string]string{
+		commandSetting: discriminatorCommand, listenAddr: "8080",
+	}, 1, "unhealthy: invalid setting: SWW_LISTEN_ADDR: ")
+
+	stop()
+
+	select {
+	case <-exited:
+	case <-time.After(waitLimit):
+		t.Fatal("still running after being told to stop")
+	}
+}
+
+func TestHealthCheckInRProxysContainer(t *testing.T) {
+	t.Parallel()
+
+	app := httptest.NewServer(http.NotFoundHandler())
+	defer app.Close()
+
+	// Nothing listens at the backend. The health check asks rproxy and the app,
+	// whether or not discriminator answers.
+	site, _ := startRProxy(t, "http://127.0.0.1:9", app.URL)
+	_, port, _ := net.SplitHostPort(strings.TrimPrefix(site, "http://"))
+	settings := func(more ...string) map[string]string {
+		env := map[string]string{rproxyListenAddr: ":" + port, rproxyUpstreamURL: app.URL}
+		for i := 0; i < len(more); i += 2 {
+			env[more[i]] = more[i+1]
+		}
+
+		return env
+	}
+
+	wantHealthCheck(t, settings(), 0, "")
+	wantHealthCheck(t, settings(commandSetting, ""), 0, "")
+	wantHealthCheck(t, settings(commandSetting, "rproxy"), 0, "")
+
+	// SWW_COMMAND cannot be given as a file.
+	wantHealthCheck(t, settings(commandSetting+"_FILE", writeFile(t, "discriminator\n")),
+		0, "")
+
+	// Another setting does not fail it, even an invalid one, and neither does
+	// SWRP_WAF_BACKEND being unset.
+	wantHealthCheck(t, settings("SWRP_TIMEOUT_MS", "soon"), 0, "")
+
+	// Another test can take a closed server's port at once, so the app's
+	// address here is a multicast address, to which TCP cannot connect.
+	wantHealthCheck(t, settings(rproxyUpstreamURL, "http://224.0.0.1"), 1,
+		"unhealthy: connect to the app: ")
+
+	// Listening on port 0 takes any free port, so nothing listens on port 0
+	// itself.
+	wantHealthCheck(t, settings(rproxyListenAddr, ":0"), 1, "unhealthy: ask rproxy: ")
+	wantHealthCheck(t, settings(rproxyListenAddr, "8080"), 1,
+		"unhealthy: invalid setting: SWRP_LISTEN_ADDR: ")
+
+	// No setting of rproxy's has a form given as a file.
+	wantHealthCheck(t, map[string]string{
+		rproxyListenAddr + "_FILE": writeFile(t, ":"+port+"\n"),
+		rproxyUpstreamURL:          app.URL,
+	}, 1, "unhealthy: ask rproxy: ")
+}
+
+func TestHealthCheckFailsWithAnyOtherCommand(t *testing.T) {
+	t.Parallel()
+
+	wantHealthCheck(t, map[string]string{commandSetting: "healthcheck"}, 1,
+		"unhealthy: invalid setting: SWW_COMMAND: "+
+			"\"healthcheck\" is not rproxy or discriminator\n")
+}
+
+func TestHealthCheckRefusesAnArgument(t *testing.T) {
+	t.Parallel()
+
+	var stderr bytes.Buffer
+
+	noSettings := func(string) (string, bool) {
+		return "", false
+	}
+
+	got := sww.HealthCheck(t.Context(), []string{"now"}, noSettings, &stderr)
+
+	want := "sww healthcheck: unexpected argument \"now\"\n"
+	if got != 1 || stderr.String() != want {
+		t.Errorf("health check returned %d and wrote %q, want 1 and %q",
+			got, stderr.String(), want)
+	}
+}
+
+// wantHealthCheck runs the health check with the settings in env, and
+// checks its exit status and the start of what it writes to stderr,
+// which is nothing when message is empty.
+func wantHealthCheck(t *testing.T, env map[string]string, status int, message string) {
+	t.Helper()
+
+	var stderr bytes.Buffer
+
+	got := sww.HealthCheck(t.Context(), nil, func(name string) (string, bool) {
+		value, ok := env[name]
+
+		return value, ok
+	}, &stderr)
+
+	wrote := stderr.String()
+	if got != status || !strings.HasPrefix(wrote, message) ||
+		(message == "" && wrote != "") {
+		t.Errorf("health check returned %d and wrote %q, want %d and %q",
+			got, wrote, status, message)
+	}
+}
+
+// writeFile writes contents to a file in a directory of its own, removed
+// when the test ends, and returns the file's path.
+func writeFile(t *testing.T, contents string) string {
+	t.Helper()
+
+	path := filepath.Join(t.TempDir(), "setting")
+
+	err := os.WriteFile(path, []byte(contents), 0o600)
+	if err != nil {
+		t.Fatalf("write %s: %v", path, err)
+	}
+
+	return path
+}

internal/sww/rproxy.go

@@ -0,0 +1,99 @@
+package sww
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"net"
+	"net/http"
+	"os"
+
+	"sneak.berlin/go/smallwebwaf/internal/config"
+	"sneak.berlin/go/smallwebwaf/internal/requestlog"
+	"sneak.berlin/go/smallwebwaf/internal/rproxy"
+)
+
+// RunRProxy is `sww rproxy`: it reads rproxy's settings, then serves
+// requests until ctx is done. Then it gives the requests in progress
+// shutdownTimeout to finish, and the reports being sent SWRP_TIMEOUT_MS. It
+// returns the process's exit status, 1 when rproxy cannot start. Its process
+// messages name the host's name as instance, and it writes those at info or
+// more severe: rproxy has no setting for either.
+func RunRProxy(ctx context.Context, params Params) int {
+	hostname, _ := os.Hostname() // "" when the host has no name to give
+	processLog := requestlog.NewProcessLogger(params.Stdout, hostname, slog.LevelInfo)
+
+	cfg, err := config.RProxyFromEnvironment(params.LookupEnv)
+	if err != nil {
+		processLog.Error("invalid setting", "error", err.Error())
+
+		return 1
+	}
+
+	server := rproxy.New(rproxy.Params{
+		Config:     cfg,
+		RequestLog: params.Stdout,
+		ProcessLog: processLog,
+	})
+
+	listener, err := (&net.ListenConfig{}).Listen(ctx, "tcp", cfg.ListenAddr)
+	if err != nil {
+		processLog.Error("cannot listen on SWRP_LISTEN_ADDR", "error", err.Error())
+
+		return 1
+	}
+
+	processLog.Info("starting",
+		"version", params.Version,
+		"address", listener.Addr().String(),
+		"settings", cfg)
+
+	return serveRProxy(ctx, server, listener, processLog)
+}
+
+// serveRProxy serves requests on listener until ctx is done, and then stops as
+// RunRProxy says.
+func serveRProxy(
+	ctx context.Context, server *rproxy.Server, listener net.Listener,
+	processLog *slog.Logger,
+) int {
+	served := make(chan error, 1)
+
+	go func() {
+		served <- server.Serve(listener)
+	}()
+
+	select {
+	case err := <-served:
+		processLog.Error("serving failed", "error", err.Error())
+
+		return 1
+	case <-ctx.Done():
+	}
+
+	processLog.Info("stopping")
+
+	shutdownCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx),
+		shutdownTimeout)
+	defer cancel()
+
+	err := server.Shutdown(shutdownCtx)
+	if err != nil {
+		processLog.Warn("requests still in progress were cut off",
+			"error", err.Error())
+
+		_ = server.Close()
+	}
+
+	err = <-served
+	if !errors.Is(err, http.ErrServerClosed) {
+		processLog.Error("serving failed", "error", err.Error())
+
+		return 1
+	}
+
+	server.WaitForReports()
+	processLog.Info("stopped")
+
+	return 0
+}

internal/sww/rproxy_test.go

@@ -0,0 +1,791 @@
+package sww_test
+
+import (
+	"bufio"
+	"context"
+	"encoding/pem"
+	"io"
+	"maps"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync/atomic"
+	"syscall"
+	"testing"
+	"time"
+
+	"sneak.berlin/go/smallwebwaf/internal/discriminator"
+	"sneak.berlin/go/smallwebwaf/internal/sww"
+)
+
+const (
+	// wafBackend is the setting that names discriminator to rproxy.
+	wafBackend = "SWRP_WAF_BACKEND"
+	// rproxyListenAddr is the setting that says where rproxy listens.
+	rproxyListenAddr = "SWRP_LISTEN_ADDR"
+	// rproxyUpstreamURL is the setting that says where the app is.
+	rproxyUpstreamURL = "SWRP_UPSTREAM_URL"
+	// rproxyTimeout is the setting that says how long rproxy waits for discriminator.
+	rproxyTimeout = "SWRP_TIMEOUT_MS"
+	// bannedProbe is a path the default rule file bans a client for.
+	bannedProbe = "/.env"
+	// otherClient is a client the tests do not ban.
+	otherClient = "198.51.100.7"
+)
+
+// runRProxy runs `sww rproxy` with the settings in env until ctx is
+// done, and returns its exit status.
+func runRProxy(ctx context.Context, env map[string]string, out *output) int {
+	return sww.RunRProxy(ctx, sww.Params{
+		Version: testVersion,
+		LookupEnv: func(name string) (string, bool) {
+			value, ok := env[name]
+
+			return value, ok
+		},
+		Stdout: out,
+	})
+}
+
+// startRProxy runs rproxy with discriminator at backend and the app at appURL, and
+// returns its URL and what it writes. It believes X-Forwarded-For from the
+// tests. It stops as the test ends, and that it stopped in order is checked.
+func startRProxy(t *testing.T, backend, appURL string) (string, *output) {
+	t.Helper()
+
+	ctx, stop := context.WithCancel(t.Context())
+	out := &output{}
+	exited := make(chan int, 1)
+
+	go func() {
+		exited <- runRProxy(ctx, map[string]string{
+			rproxyListenAddr:       localhost + ":0",
+			"SWRP_UPSTREAM_URL":    appURL,
+			"SWRP_TRUSTED_PROXIES": localhost + "/32",
+			wafBackend:             backend,
+		}, out)
+	}()
+
+	t.Cleanup(func() {
+		stop()
+
+		select {
+		case status := <-exited:
+			if status != 0 {
+				t.Errorf("rproxy's exit status %d, want 0; output:\n%s", status, out.text())
+			}
+		case <-time.After(waitLimit):
+			t.Error("rproxy still running after being told to stop")
+		}
+
+		out.line(t, "msg", "stopped")
+	})
+
+	addr, _ := out.line(t, "msg", "starting")["address"].(string)
+
+	return "http://" + addr, out
+}
+
+// startDiscriminator runs discriminator with the settings in env, and returns
+// its address and URL, what it writes, and what stops it, which checks that it
+// stopped in order.
+func startDiscriminator(t *testing.T, env map[string]string) (string, *output, func()) {
+	t.Helper()
+
+	ctx, stop := context.WithCancel(t.Context())
+	out := &output{}
+	exited := make(chan int, 1)
+
+	go func() {
+		exited <- run(ctx, env, out)
+	}()
+
+	addr, _ := out.line(t, "msg", "starting")["address"].(string)
+
+	return addr, out, func() {
+		stop()
+
+		select {
+		case status := <-exited:
+			if status != 0 {
+				t.Fatalf("discriminator's exit status %d, want 0; output:\n%s", status, out.text())
+			}
+		case <-time.After(waitLimit):
+			t.Fatal("discriminator still running after being told to stop")
+		}
+	}
+}
+
+// discriminatorSettings are discriminator's settings with the default rule
+// file, to listen at any free port.
+func discriminatorSettings(t *testing.T) map[string]string {
+	t.Helper()
+
+	return map[string]string{
+		listenAddr: localhost + ":0",
+		stateDir:   t.TempDir(),
+		rulesDir:   filepath.Join("..", "..", "share", "rules.d"),
+	}
+}
+
+func TestABanThroughOneSiteRefusesTheClientOnTheOther(t *testing.T) {
+	t.Parallel()
+
+	addr, discriminatorOut, stop := startDiscriminator(t, discriminatorSettings(t))
+	defer stop()
+
+	backend := "http://" + addr
+	first, firstOut := startRProxy(t, backend, startApp(t))
+	second, _ := startRProxy(t, backend, startApp(t))
+
+	wantStatus(t, first+"/", placed, http.StatusOK)
+	wantStatus(t, second+"/", placed, http.StatusOK)
+
+	// The default rule file bans a probe for this.
+	wantStatus(t, first+bannedProbe, placed, http.StatusForbidden)
+	wantStatus(t, second+"/", placed, http.StatusForbidden)
+	wantStatus(t, second+"/", otherClient, http.StatusOK)
+
+	// One request's three lines carry one id, rproxy's with discriminator's answer.
+	line := firstOut.line(t, "path", bannedProbe)
+	if line["disposition"] != discriminator.DispositionDeny ||
+		line["client_ip"] != placed {
+		t.Errorf("rproxy's line is %v", line)
+	}
+
+	id, _ := line["request_id"].(string)
+	decision := discriminatorLine(t, discriminatorOut, id, "decision")
+	report := discriminatorLine(t, discriminatorOut, id, "report")
+
+	if decision["host"] != line["host"] || decision["client_ip"] != placed ||
+		report["status"] != float64(http.StatusForbidden) {
+		t.Errorf("discriminator's lines are %v and %v", decision, report)
+	}
+}
+
+// discriminatorLine waits for discriminator's line with request id and
+// discriminator_line kind.
+func discriminatorLine(t *testing.T, out *output, id, kind string) map[string]any {
+	t.Helper()
+
+	deadline := time.Now().Add(waitLimit)
+	for time.Now().Before(deadline) {
+		line := out.findLine(t, func(fields map[string]any) bool {
+			return fields["request_id"] == id && fields["discriminator_line"] == kind
+		})
+		if line != nil {
+			return line
+		}
+
+		time.Sleep(pollInterval)
+	}
+
+	t.Fatalf("no %s line for %s in:\n%s", kind, id, out.text())
+
+	return nil
+}
+
+func TestTheSiteAnswersWhileDiscriminatorIsStoppedAndStartedAgain(t *testing.T) {
+	t.Parallel()
+
+	env := discriminatorSettings(t)
+	addr, _, stop := startDiscriminator(t, env)
+	site, out := startRProxy(t, "http://"+addr, startApp(t))
+
+	// A visitor who keeps asking throughout, and is never refused.
+	visitorDone := make(chan struct{})
+	visitorStop := make(chan struct{})
+
+	var refused atomic.Int64
+
+	go func() {
+		defer close(visitorDone)
+
+		for {
+			select {
+			case <-visitorStop:
+				return
+			case <-time.After(pollInterval):
+			}
+
+			if !answersNormally(t.Context(), site+"/", otherClient) {
+				refused.Add(1)
+			}
+		}
+	}()
+
+	wantStatus(t, site+bannedProbe, placed, http.StatusForbidden)
+	wantStatus(t, site+"/", placed, http.StatusForbidden)
+	stop()
+
+	// With discriminator down, the site answers, and rproxy says the default did it.
+	wantStatus(t, site+"/after-stop", placed, http.StatusOK)
+
+	line := out.line(t, "path", "/after-stop")
+	if line["disposition"] != discriminator.DispositionAllow ||
+		line["default_disposition"] != true {
+		t.Errorf("rproxy's line is %v", line)
+	}
+
+	// Back at its address, with its state files, it applies the ban again.
+	restarted := maps.Clone(env)
+	restarted[listenAddr] = addr
+	_, _, stop = startDiscriminator(t, restarted)
+
+	defer stop()
+
+	wantStatus(t, site+"/", placed, http.StatusForbidden)
+
+	close(visitorStop)
+	<-visitorDone
+
+	if refused.Load() != 0 {
+		t.Errorf("the site refused its visitor %d times", refused.Load())
+	}
+}
+
+// answersNormally reports whether a request to url from the client at from, as
+// X-Forwarded-For names it, gets the app's answer.
+func answersNormally(ctx context.Context, url, from string) bool {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, http.NoBody)
+	if err != nil {
+		return false
+	}
+
+	req.Header.Set("X-Forwarded-For", from)
+
+	transport := &http.Transport{}
+	defer transport.CloseIdleConnections()
+
+	res, err := transport.RoundTrip(req)
+	if err != nil {
+		return false
+	}
+
+	body, err := io.ReadAll(res.Body)
+	_ = res.Body.Close()
+
+	return err == nil && res.StatusCode == http.StatusOK && string(body) == greeting
+}
+
+// sendRaw sends request, as it is, to rproxy at site, and returns the status of
+// the answer.
+func sendRaw(t *testing.T, site, request string) int {
+	t.Helper()
+
+	conn, err := (&net.Dialer{}).DialContext(t.Context(), "tcp",
+		strings.TrimPrefix(site, "http://"))
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+
+	defer func() {
+		_ = conn.Close()
+	}()
+
+	err = conn.SetDeadline(time.Now().Add(waitLimit))
+	if err != nil {
+		t.Fatalf("set the deadline: %v", err)
+	}
+
+	_, err = io.WriteString(conn, request)
+	if err != nil {
+		t.Fatalf("send: %v", err)
+	}
+
+	res, err := http.ReadResponse(bufio.NewReader(conn), nil)
+	if err != nil {
+		t.Fatalf("read the answer: %v", err)
+	}
+
+	_ = res.Body.Close()
+
+	return res.StatusCode
+}
+
+func TestAChunkedBodyStaysABodyAtDiscriminator(t *testing.T) {
+	t.Parallel()
+
+	addr, discriminatorOut, stop := startDiscriminator(t, discriminatorSettings(t))
+	defer stop()
+
+	site, _ := startRProxy(t, "http://"+addr, startApp(t))
+
+	// Read as a further request, this body would ban otherClient on every site.
+	const smuggledID = "smuggled-request"
+
+	smuggled := "GET " + bannedProbe + " HTTP/1.1\r\nHost: site.example\r\n" +
+		discriminator.ClientHeader + ": " + otherClient + "\r\n" +
+		"X-Request-ID: " + smuggledID + "\r\n\r\n"
+
+	status := sendRaw(t, site, "POST /upload HTTP/1.1\r\nHost: site.example\r\n"+
+		"X-Forwarded-For: "+placed+"\r\nTransfer-Encoding: chunked\r\n\r\n"+
+		strconv.FormatInt(int64(len(smuggled)), 16)+"\r\n"+smuggled+"\r\n0\r\n\r\n")
+	if status != http.StatusOK {
+		t.Errorf("status %d, want the app's 200", status)
+	}
+
+	// Read as a further request, it would act straight after the answer.
+	for range 20 {
+		wantStatus(t, site+"/", otherClient, http.StatusOK)
+		time.Sleep(pollInterval)
+	}
+
+	line := discriminatorOut.findLine(t, func(fields map[string]any) bool {
+		return fields["request_id"] == smuggledID
+	})
+	if line != nil {
+		t.Errorf("discriminator read the body as a request: %v", line)
+	}
+}
+
+func TestATargetThatCannotGoToDiscriminatorIsRefusedForABannedClient(t *testing.T) {
+	t.Parallel()
+
+	addr, _, stop := startDiscriminator(t, discriminatorSettings(t))
+	defer stop()
+
+	site, out := startRProxy(t, "http://"+addr, startApp(t))
+
+	wantStatus(t, site+bannedProbe, placed, http.StatusForbidden)
+
+	status := sendRaw(t, site, "GET x:admin HTTP/1.1\r\nHost: site.example\r\n"+
+		"X-Forwarded-For: "+placed+"\r\n\r\n")
+	if status != http.StatusBadRequest {
+		t.Errorf("status %d, want 400", status)
+	}
+
+	line := out.line(t, "path", "admin")
+	if line["disposition"] != nil || line["default_disposition"] != nil {
+		t.Errorf("rproxy's line is %v, want no disposition", line)
+	}
+}
+
+func TestAClientWithAZoneIsTheClientWithout(t *testing.T) {
+	t.Parallel()
+
+	const zoned = "2001:db8::1%eth0"
+
+	addr, _, stop := startDiscriminator(t, discriminatorSettings(t))
+	defer stop()
+
+	site, out := startRProxy(t, "http://"+addr, startApp(t))
+
+	wantStatus(t, site+bannedProbe, zoned, http.StatusForbidden)
+	wantStatus(t, site+"/", zoned, http.StatusForbidden)
+	wantStatus(t, site+"/", "2001:db8::1", http.StatusForbidden)
+
+	line := out.line(t, "path", bannedProbe)
+	if line["client_ip"] != "2001:db8::1" || line["default_disposition"] != nil {
+		t.Errorf("rproxy's line is %v", line)
+	}
+}
+
+func TestAStatusDiscriminatorDoesNotTakeIsReportedAs599(t *testing.T) {
+	t.Parallel()
+
+	const status = 799
+
+	app := httptest.NewServer(http.HandlerFunc(
+		func(w http.ResponseWriter, _ *http.Request) {
+			w.WriteHeader(status)
+		}))
+	t.Cleanup(app.Close)
+
+	addr, discriminatorOut, stop := startDiscriminator(t, discriminatorSettings(t))
+	defer stop()
+
+	site, out := startRProxy(t, "http://"+addr, app.URL)
+
+	wantStatus(t, site+"/", placed, status)
+
+	line := out.line(t, "path", "/")
+	if line["status"] != float64(status) {
+		t.Errorf("rproxy's line is %v, want the status the visitor got", line)
+	}
+
+	id, _ := line["request_id"].(string)
+
+	report := discriminatorLine(t, discriminatorOut, id, "report")
+	if report["status"] != 599.0 {
+		t.Errorf("discriminator's line is %v, want status 599", report)
+	}
+}
+
+// startRProxyProcess runs `sww rproxy` in a process of its own, with the
+// settings in env, each as NAME=value. It returns where rproxy listens, the
+// process, and what rproxy writes.
+func startRProxyProcess(t *testing.T, env ...string) (string, *exec.Cmd, *output) {
+	t.Helper()
+
+	binary, err := os.Executable()
+	if err != nil {
+		t.Fatalf("find the test binary: %v", err)
+	}
+
+	ctx, cancel := context.WithTimeout(t.Context(), 3*waitLimit)
+	t.Cleanup(cancel)
+
+	out := &output{}
+	cmd := exec.CommandContext(ctx, binary) //nolint:gosec // this test binary
+	cmd.Stdout = out
+
+	cmd.Env = append(os.Environ(),
+		commandForMain+"=rproxy",
+		"SWRP_LISTEN_ADDR="+localhost+":0")
+	cmd.Env = append(cmd.Env, env...)
+
+	err = cmd.Start()
+	if err != nil {
+		t.Fatalf("start: %v", err)
+	}
+
+	t.Cleanup(func() {
+		_ = cmd.Process.Kill()
+		_ = cmd.Wait()
+	})
+
+	addr, _ := out.line(t, "msg", "starting")["address"].(string)
+
+	return addr, cmd, out
+}
+
+func TestRProxyAsksADiscriminatorThatSpeaksHTTP2OverHTTP1(t *testing.T) {
+	t.Parallel()
+
+	protocols := make(chan string, 10)
+	backend := httptest.NewUnstartedServer(http.HandlerFunc(
+		func(w http.ResponseWriter, r *http.Request) {
+			if r.Header.Get(discriminator.ClientHeader) != "" {
+				protocols <- r.Proto
+
+				w.Header().Set(discriminator.DispositionHeader, discriminator.DispositionDeny)
+				w.WriteHeader(http.StatusForbidden)
+			}
+		}))
+	backend.EnableHTTP2 = true
+	backend.StartTLS()
+	t.Cleanup(backend.Close)
+
+	// rproxy has no setting for the certificates it trusts, so it is told of this
+	// one as Go's client is told of any, in a process of its own.
+	certificates := filepath.Join(t.TempDir(), "discriminator.pem")
+
+	err := os.WriteFile(certificates, pem.EncodeToMemory(&pem.Block{
+		Type: "CERTIFICATE", Bytes: backend.Certificate().Raw,
+	}), 0o600)
+	if err != nil {
+		t.Fatalf("write the certificate: %v", err)
+	}
+
+	addr, _, _ := startRProxyProcess(t,
+		wafBackend+"="+backend.URL,
+		"SWRP_UPSTREAM_URL="+startApp(t),
+		"SSL_CERT_FILE="+certificates)
+
+	for name, headers := range map[string]string{
+		"no such header":      "",
+		"Upgrade":             "Upgrade: anything\r\n",
+		"Connection":          "Connection: anything\r\n",
+		"a WebSocket upgrade": "Connection: Upgrade\r\nUpgrade: websocket\r\n",
+	} {
+		status := sendRaw(t, addr, "GET / HTTP/1.1\r\nHost: site.example\r\n"+headers+"\r\n")
+		if status != http.StatusForbidden {
+			t.Errorf("with %s: status %d, want discriminator's 403", name, status)
+		}
+
+		select {
+		case protocol := <-protocols:
+			if protocol != "HTTP/1.1" {
+				t.Errorf("with %s: asked over %s, want HTTP/1.1", name, protocol)
+			}
+		case <-time.After(waitLimit):
+			t.Errorf("with %s: discriminator was not asked", name)
+		}
+	}
+}
+
+func TestRProxyProcessLinesNameTheHostAndComeAtInfo(t *testing.T) {
+	t.Parallel()
+
+	hostname, err := os.Hostname()
+	if err != nil {
+		t.Fatalf("the host's name: %v", err)
+	}
+
+	// Nothing listens at the backend's address; no request asks it.
+	_, out := startRProxy(t, "http://127.0.0.1:9", startApp(t))
+
+	line := out.line(t, "msg", "starting")
+	if line["instance"] != hostname || line["level"] != "INFO" {
+		t.Errorf("rproxy's line is %v, want instance %s at INFO", line, hostname)
+	}
+}
+
+func TestRProxyStoppingWaitsForAReportBeingSent(t *testing.T) {
+	t.Parallel()
+
+	reportGotThere := make(chan struct{})
+	release := make(chan struct{})
+	backend := httptest.NewServer(http.HandlerFunc(
+		func(w http.ResponseWriter, r *http.Request) {
+			if r.Header.Get(discriminator.ClientHeader) != "" {
+				w.Header().Set(discriminator.DispositionHeader, discriminator.DispositionAllow)
+
+				return
+			}
+
+			close(reportGotThere)
+			<-release
+			w.WriteHeader(http.StatusNoContent)
+		}))
+	t.Cleanup(backend.Close)
+
+	ctx, stop := context.WithCancel(t.Context())
+	defer stop()
+
+	env := map[string]string{
+		rproxyListenAddr:    localhost + ":0",
+		"SWRP_UPSTREAM_URL": startApp(t),
+		rproxyTimeout:       "60000",
+		wafBackend:          backend.URL,
+	}
+	out := &output{}
+	exited := make(chan int, 1)
+
+	go func() {
+		exited <- runRProxy(ctx, env, out)
+	}()
+
+	addr, _ := out.line(t, "msg", "starting")["address"].(string)
+
+	wantStatus(t, "http://"+addr+"/", placed, http.StatusOK)
+
+	select {
+	case <-reportGotThere:
+	case <-time.After(waitLimit):
+		t.Fatal("no report")
+	}
+
+	stop()
+
+	select {
+	case <-exited:
+		t.Error("rproxy stopped with a report on its way")
+		close(release)
+
+		return
+	case <-time.After(10 * pollInterval):
+	}
+
+	close(release)
+
+	select {
+	case status := <-exited:
+		if status != 0 {
+			t.Errorf("rproxy's exit status %d, want 0; output:\n%s", status, out.text())
+		}
+	case <-time.After(waitLimit):
+		t.Error("rproxy still running, the report having been answered")
+	}
+}
+
+// sendAndKeepOpen sends request to rproxy at addr, and returns the connection. A
+// read from it ends with an error once waitLimit is up.
+func sendAndKeepOpen(t *testing.T, addr, request string) net.Conn {
+	t.Helper()
+
+	conn, err := (&net.Dialer{}).DialContext(t.Context(), "tcp", addr)
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+
+	t.Cleanup(func() {
+		_ = conn.Close()
+	})
+
+	err = conn.SetDeadline(time.Now().Add(waitLimit))
+	if err != nil {
+		t.Fatalf("set the deadline: %v", err)
+	}
+
+	_, err = io.WriteString(conn, request)
+	if err != nil {
+		t.Fatalf("send: %v", err)
+	}
+
+	return conn
+}
+
+func TestRProxyStoppingCutsOffARequestAfterFiveSeconds(t *testing.T) {
+	t.Parallel()
+
+	const (
+		cutOffAfter     = 5 * time.Second
+		upstreamSetting = "SWRP_UPSTREAM_URL"
+	)
+
+	asked := make(chan struct{})
+	release := make(chan struct{})
+	app := httptest.NewServer(http.HandlerFunc(
+		func(http.ResponseWriter, *http.Request) {
+			close(asked)
+			<-release
+		}))
+	t.Cleanup(app.Close)
+	t.Cleanup(func() { close(release) })
+
+	backend := httptest.NewServer(http.HandlerFunc(
+		func(w http.ResponseWriter, _ *http.Request) {
+			w.Header().Set(discriminator.DispositionHeader, discriminator.DispositionAllow)
+		}))
+	t.Cleanup(backend.Close)
+
+	ctx, stop := context.WithCancel(t.Context())
+	defer stop()
+
+	env := map[string]string{
+		rproxyListenAddr: localhost + ":0",
+		upstreamSetting:  app.URL,
+		wafBackend:       backend.URL,
+	}
+	out := &output{}
+	exited := make(chan int, 1)
+
+	go func() {
+		exited <- runRProxy(ctx, env, out)
+	}()
+
+	addr, _ := out.line(t, "msg", "starting")["address"].(string)
+	conn := sendAndKeepOpen(t, addr, "GET / HTTP/1.1\r\nHost: site.example\r\n\r\n")
+
+	select {
+	case <-asked:
+	case <-time.After(waitLimit):
+		t.Fatal("the app was not asked")
+	}
+
+	stopped := time.Now()
+
+	stop()
+
+	text, err := io.ReadAll(conn)
+	if err != nil || len(text) != 0 {
+		t.Errorf("read %q (%v), want the connection cut off", text, err)
+	}
+
+	if elapsed := time.Since(stopped); elapsed < cutOffAfter {
+		t.Errorf("cut off after %s, want %s", elapsed, cutOffAfter)
+	}
+
+	select {
+	case status := <-exited:
+		if status != 0 {
+			t.Errorf("rproxy's exit status %d, want 0; output:\n%s", status, out.text())
+		}
+	case <-time.After(waitLimit):
+		t.Error("rproxy still running after the cut-off")
+	}
+
+	out.line(t, "msg", "requests still in progress were cut off")
+}
+
+func TestRProxyInvalidSettingStopsTheStart(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range []struct {
+		name, setting string
+		env           map[string]string
+	}{
+		{"no backend", wafBackend, nil},
+		{
+			"another invalid", rproxyTimeout,
+			map[string]string{wafBackend: "http://discriminator:8080", rproxyTimeout: "soon"},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			out := &output{}
+
+			if status := runRProxy(t.Context(), tc.env, out); status != 1 {
+				t.Errorf("exit status %d, want 1", status)
+			}
+
+			line := out.line(t, "msg", "invalid setting")
+			message, _ := line["error"].(string)
+
+			if !strings.HasPrefix(message, tc.setting+": ") {
+				t.Errorf("start refused with %v, want it to name %s", line, tc.setting)
+			}
+		})
+	}
+}
+
+func TestRProxyAddressInUseStopsTheStart(t *testing.T) {
+	t.Parallel()
+
+	taken, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", localhost+":0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+
+	defer func() {
+		_ = taken.Close()
+	}()
+
+	out := &output{}
+
+	status := runRProxy(t.Context(), map[string]string{
+		wafBackend:       "http://discriminator:8080",
+		rproxyListenAddr: taken.Addr().String(),
+	}, out)
+	if status != 1 {
+		t.Errorf("exit status %d, want 1", status)
+	}
+
+	out.line(t, "msg", "cannot listen on SWRP_LISTEN_ADDR")
+}
+
+func TestMainRunsRProxy(t *testing.T) {
+	t.Parallel()
+
+	// Nothing listens there; the health check does not ask.
+	addr, cmd, out := startRProxyProcess(t, wafBackend+"=http://127.0.0.1:9")
+
+	req, err := http.NewRequestWithContext(t.Context(), http.MethodGet,
+		"http://"+addr+discriminator.HealthPath, http.NoBody)
+	if err != nil {
+		t.Fatalf("new request: %v", err)
+	}
+
+	transport := &http.Transport{}
+	defer transport.CloseIdleConnections()
+
+	res, err := transport.RoundTrip(req)
+	if err != nil {
+		t.Fatalf("health check: %v", err)
+	}
+
+	body, err := io.ReadAll(res.Body)
+	_ = res.Body.Close()
+
+	if err != nil || string(body) != "ok\n" {
+		t.Errorf("health check answered %q (%v)", body, err)
+	}
+
+	err = cmd.Process.Signal(syscall.SIGTERM)
+	if err != nil {
+		t.Fatalf("signal: %v", err)
+	}
+
+	err = cmd.Wait()
+	if err != nil {
+		t.Errorf("exit: %v; output:\n%s", err, out.text())
+	}
+}

internal/sww/sww.go

@@ -0,0 +1,417 @@
+// Package sww runs the sww process, as the command it is given.
+// `sww discriminator` reads the settings, the rule files, the lookup database
+// and the state files, answers rproxy's questions until it is told to stop, and
+// then stops in an orderly way, writing the state files. `sww rproxy`
+// reads rproxy's settings alone, and runs the reverse proxy in front of one app
+// that asks discriminator about each request. `sww healthcheck` is the
+// container's health check.
+package sww
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"net"
+	"net/http"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"sneak.berlin/go/smallwebwaf/internal/alerts"
+	"sneak.berlin/go/smallwebwaf/internal/config"
+	"sneak.berlin/go/smallwebwaf/internal/discriminator"
+	"sneak.berlin/go/smallwebwaf/internal/lookup"
+	"sneak.berlin/go/smallwebwaf/internal/remotelog"
+	"sneak.berlin/go/smallwebwaf/internal/reputation"
+	"sneak.berlin/go/smallwebwaf/internal/requestlog"
+	"sneak.berlin/go/smallwebwaf/internal/rules"
+	"sneak.berlin/go/smallwebwaf/internal/state"
+	"sneak.berlin/go/smallwebwaf/internal/waf"
+)
+
+// shutdownTimeout is how long requests in progress may take to finish
+// once sww is told to stop, before their connections are closed.
+// runit and docker wait a little longer before they kill the process.
+const shutdownTimeout = 5 * time.Second
+
+// remoteLogStopTimeout is how long, as discriminator stops, the log lines still
+// waiting are sent to SWW_LOG_REMOTE_URL before they are given up. stdout has
+// carried them.
+const remoteLogStopTimeout = 2 * time.Second
+
+// usageStatus is the exit status with no command, or one sww does not
+// know.
+const usageStatus = 2
+
+// usage is what sww prints then.
+const usage = `usage: sww <command>
+
+commands:
+  rproxy          the reverse proxy in front of one app
+  discriminator   decides for rproxy, and keeps bans, counters and history
+  healthcheck     the container's health check
+`
+
+// Params are what RunDiscriminator and RunRProxy need from the process.
+type Params struct {
+	// Version is the version of the binary, set when it is built.
+	Version string
+	// LookupEnv reads an environment variable, normally os.LookupEnv.
+	LookupEnv func(string) (string, bool)
+	// Stdout receives the request log and the process's own messages.
+	Stdout io.Writer
+}
+
+// Main runs the command os.Args names, and returns the process's exit status.
+// `sww rproxy` and `sww discriminator` run until SIGTERM or SIGINT.
+// With no command, or one it does not know, it prints the commands and returns
+// usageStatus.
+func Main(version string) int {
+	args := os.Args[1:]
+	if len(args) == 0 {
+		return printUsage(os.Stderr)
+	}
+
+	ctx, stop := signal.NotifyContext(context.Background(),
+		syscall.SIGTERM, os.Interrupt)
+	defer stop()
+
+	params := Params{Version: version, LookupEnv: os.LookupEnv, Stdout: os.Stdout}
+
+	switch args[0] {
+	case "rproxy":
+		return RunRProxy(ctx, params)
+	case "discriminator":
+		return RunDiscriminator(ctx, params)
+	case "healthcheck":
+		return HealthCheck(ctx, args[1:], os.LookupEnv, os.Stderr)
+	default:
+		return printUsage(os.Stderr)
+	}
+}
+
+// printUsage writes the commands to w, and returns usageStatus.
+func printUsage(w io.Writer) int {
+	_, _ = fmt.Fprint(w, usage)
+
+	return usageStatus
+}
+
+// RunDiscriminator is `sww discriminator`: it reads the settings, the rule files,
+// the lookup database and the state files, then serves requests until ctx is
+// done. It returns the process's exit status, 1 when discriminator cannot start.
+func RunDiscriminator(ctx context.Context, params Params) int {
+	// Until the settings are read, the one message is an invalid setting's
+	// error, which every SWW_LOG_LEVEL lets through.
+	processLog := requestlog.NewProcessLogger(params.Stdout,
+		config.InstanceName(params.LookupEnv), slog.LevelError)
+
+	cfg, err := config.FromEnvironment(params.LookupEnv)
+	if err != nil {
+		processLog.Error("invalid setting", "error", err.Error())
+
+		return 1
+	}
+
+	// While SWW_LOG_REMOTE_URL is set, every line on stdout from here on
+	// is sent there too.
+	stdout := params.Stdout
+
+	var remote *remotelog.Sender
+
+	if cfg.LogRemoteURL != nil {
+		remote = newRemoteLogSender(cfg)
+		stdout = io.MultiWriter(params.Stdout, remote)
+	}
+
+	processLog = requestlog.NewProcessLogger(stdout, cfg.InstanceName, cfg.LogLevel)
+
+	if remote != nil {
+		stopSending := startSending(ctx, remote, processLog)
+		defer stopSending()
+	}
+
+	// The state files and the alerts give times in UTC.
+	now := func() time.Time { return time.Now().UTC() }
+
+	alertQueue := newAlertQueue(cfg, now, processLog)
+
+	ruleFiles, err := rules.Load(rules.Params{
+		Dir:        cfg.RulesDir,
+		Enabled:    cfg.RulesEnabled,
+		ProcessLog: processLog,
+		Alerts:     alertQueue,
+	})
+	if err != nil {
+		processLog.Error("cannot use the rule files", "error", err.Error())
+
+		return 1
+	}
+
+	crs, err := discriminator.NewCoreRuleSet(cfg)
+	if err != nil {
+		processLog.Error("cannot use the Core Rule Set", "error", err.Error())
+
+		return 1
+	}
+
+	server, err := newServer(cfg, stdout, processLog, now, ruleFiles, alertQueue, crs)
+	if err != nil {
+		processLog.Error("cannot use the lookup database", "error", err.Error())
+
+		return 1
+	}
+
+	if remote != nil {
+		server.Metrics.AddRemoteLog(remote)
+	}
+
+	files, err := loadStateFiles(cfg, server, alertQueue, now, processLog)
+	if err != nil {
+		processLog.Error("cannot use the state files", "error", err.Error())
+
+		return 1
+	}
+
+	listener, err := (&net.ListenConfig{}).Listen(ctx, "tcp", cfg.ListenAddr)
+	if err != nil {
+		processLog.Error("cannot listen on SWW_LISTEN_ADDR", "error", err.Error())
+
+		return 1
+	}
+
+	processLog.Info("starting",
+		"version", params.Version,
+		"address", listener.Addr().String(),
+		"settings", cfg)
+
+	return serve(ctx, server, listener, files, ruleFiles, alertQueue, processLog)
+}
+
+// newServer returns the server discriminator runs, with the metrics of the alerts,
+// after reading the lookup database while SWW_LOOKUP_SOURCE is file. A lookup
+// database that cannot be read is an error.
+func newServer(
+	cfg *config.Config, stdout io.Writer, processLog *slog.Logger,
+	now func() time.Time, ruleFiles *rules.Files, alertQueue *alerts.Queue,
+	crs *waf.CoreRuleSet,
+) (*discriminator.Server, error) {
+	var lookupFile *lookup.File
+
+	if cfg.LookupSource == "file" {
+		var err error
+
+		lookupFile, err = lookup.OpenFile(lookup.FileParams{
+			Path:       cfg.LookupDBPath,
+			Now:        now,
+			ProcessLog: processLog,
+			Alerts:     alertQueue,
+		})
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	server := discriminator.New(discriminator.Params{
+		Config:       cfg,
+		RequestLog:   stdout,
+		ProcessLog:   processLog,
+		GeoJSURL:     lookup.URL,
+		AbuseIPDBURL: reputation.AbuseIPDBURL,
+		LookupFile:   lookupFile,
+		CoreRuleSet:  crs,
+		Now:          now,
+		Rules:        ruleFiles,
+		Alerts:       alertQueue,
+	})
+	server.Metrics.AddAlerts(alertQueue)
+
+	if lookupFile != nil {
+		server.Metrics.AddLookupFile(lookupFile.LastRead, lookupFile.ReadFailures)
+	}
+
+	return server, nil
+}
+
+// loadStateFiles reads the state files into the parts of server and into
+// alertQueue, as state.Load does.
+func loadStateFiles(
+	cfg *config.Config, server *discriminator.Server, alertQueue *alerts.Queue,
+	now func() time.Time, processLog *slog.Logger,
+) (*state.Files, error) {
+	return state.Load(state.Params{
+		Dir:             cfg.StateDir,
+		WriteDelay:      cfg.StateWriteDelay,
+		CounterInterval: cfg.StateCounterInterval,
+		Ledger:          server.Ledger,
+		Limiter:         server.Limiter,
+		GeoJS:           server.GeoJS,
+		Lists:           server.Lists,
+		DNSBL:           server.DNSBL,
+		AbuseIPDB:       server.AbuseIPDB,
+		Alerts:          alertQueue,
+		Anomalies:       server.Anomalies,
+		Now:             now,
+		ProcessLog:      processLog,
+		Metrics:         server.Metrics,
+	})
+}
+
+// newAlertQueue returns the queue of the alerts to the webhook, Slack and
+// ntfy, with the settings for them.
+func newAlertQueue(
+	cfg *config.Config, now func() time.Time, processLog *slog.Logger,
+) *alerts.Queue {
+	return alerts.New(alerts.Params{
+		WebhookURL:     cfg.AlertWebhookURL,
+		WebhookHeaders: cfg.AlertWebhookHeaders,
+		SlackURL:       cfg.AlertSlackWebhookURL,
+		NtfyURL:        cfg.AlertNtfyURL,
+		NtfyToken:      cfg.AlertNtfyToken,
+		Events:         cfg.AlertEvents,
+		Cooldown:       cfg.AlertCooldown,
+		MaxPerHour:     cfg.AlertMaxPerHour,
+		Instance:       cfg.InstanceName,
+		Now:            now,
+		ProcessLog:     processLog,
+	})
+}
+
+// newRemoteLogSender returns a sender of the log lines to
+// SWW_LOG_REMOTE_URL, with the settings for it.
+func newRemoteLogSender(cfg *config.Config) *remotelog.Sender {
+	return remotelog.New(remotelog.Params{
+		URL:      cfg.LogRemoteURL,
+		RootCAs:  cfg.LogRemoteTLSCAs,
+		Buffer:   cfg.LogRemoteBuffer,
+		Facility: cfg.LogRemoteFacility,
+		AppName:  cfg.LogRemoteAppName,
+	})
+}
+
+// startSending runs remote until the function it returns is called, which
+// then waits at most remoteLogStopTimeout for the lines still waiting to
+// be sent. Sending goes on after ctx is done, so that the lines written
+// while sww stops are sent too.
+func startSending(
+	ctx context.Context, remote *remotelog.Sender, processLog *slog.Logger,
+) func() {
+	sending, stop := context.WithCancel(context.WithoutCancel(ctx))
+	sent := make(chan struct{})
+
+	go func() {
+		remote.Run(sending, processLog)
+		close(sent)
+	}()
+
+	return func() {
+		stop()
+
+		select {
+		case <-sent:
+		case <-time.After(remoteLogStopTimeout):
+		}
+	}
+}
+
+// serve serves requests on listener, writes the state files as they are
+// due, takes in an admin's edits of them, reads the rule files again as
+// they change, and the lookup database when it is replaced, fetches the
+// lists the settings name by URL as they are due, and sends the alerts,
+// until ctx is done. Then it gives the requests in progress
+// shutdownTimeout to finish, and writes every state file, alerts.json with
+// the alerts still waiting.
+func serve(
+	ctx context.Context, server *discriminator.Server, listener net.Listener,
+	files *state.Files, ruleFiles *rules.Files, alertQueue *alerts.Queue,
+	processLog *slog.Logger,
+) int {
+	served := make(chan error, 1)
+
+	go func() {
+		served <- server.Serve(listener)
+	}()
+
+	writing, stopWriting := context.WithCancel(ctx)
+	defer stopWriting()
+
+	written := inBackground(func() { files.Run(writing) })
+	watched := inBackground(func() { files.Watch(writing) })
+	rulesWatched := inBackground(func() { ruleFiles.Watch(writing) })
+	lookupFileWatched := inBackground(func() {
+		if server.LookupFile != nil {
+			server.LookupFile.Watch(writing)
+		}
+	})
+	listsFetched := inBackground(func() { server.Lists.Run(writing) })
+	alertsSent := inBackground(func() { alertQueue.Run(writing) })
+
+	select {
+	case err := <-served:
+		processLog.Error("serving failed", "error", err.Error())
+
+		return 1
+	case <-ctx.Done():
+	}
+
+	processLog.Info("stopping")
+
+	shutdownCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx),
+		shutdownTimeout)
+	defer cancel()
+
+	err := server.Shutdown(shutdownCtx)
+	if err != nil {
+		processLog.Warn("requests still in progress were cut off",
+			"error", err.Error())
+
+		_ = server.Close()
+	}
+
+	err = <-served
+	if !errors.Is(err, http.ErrServerClosed) {
+		processLog.Error("serving failed", "error", err.Error())
+
+		return 1
+	}
+
+	// Run and Watch have ended, so nothing else reads or writes the
+	// files, and no alert is being sent, so that alerts.json keeps every
+	// alert not yet sent. Every request has ended too, but for one cut off
+	// because Shutdown timed out, which adds to its client's history only as
+	// it ends. That can be after this write, and then that request is missing
+	// from clients.json.
+	<-written
+	<-watched
+	<-rulesWatched
+	<-lookupFileWatched
+	<-listsFetched
+	<-alertsSent
+
+	err = files.WriteAll()
+	if err != nil {
+		processLog.Error("writing the state files failed", "error", err.Error())
+
+		return 1
+	}
+
+	processLog.Info("stopped")
+
+	return 0
+}
+
+// inBackground runs task on a goroutine of its own, and returns a channel
+// that is closed once task has returned.
+func inBackground(task func()) <-chan struct{} {
+	done := make(chan struct{})
+
+	go func() {
+		task()
+		close(done)
+	}()
+
+	return done
+}

internal/sww/sww_internal_test.go

@@ -0,0 +1,82 @@
+package sww
+
+import (
+	"log/slog"
+	"net/url"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"sneak.berlin/go/smallwebwaf/internal/remotelog"
+)
+
+// The stop's tests run in a synctest bubble, where the time package runs
+// on a clock of the test's own, so that how long the stop takes can be
+// told exactly. The sender is held up by its process log, not by the
+// network: a goroutine of the bubble that waits on the network keeps that
+// clock from moving on.
+
+func TestStopWaitsForTheSenderToFinish(t *testing.T) {
+	t.Parallel()
+
+	synctest.Test(t, func(t *testing.T) {
+		took := stopHeldSender(t, time.Second)
+		if took != time.Second {
+			t.Errorf("the stop took %s, want the second the sender took", took)
+		}
+	})
+}
+
+func TestStopWaitsForTheSenderAtMostTwoSeconds(t *testing.T) {
+	t.Parallel()
+
+	synctest.Test(t, func(t *testing.T) {
+		took := stopHeldSender(t, time.Minute)
+		if took != 2*time.Second {
+			t.Errorf("the stop took %s, want 2s", took)
+		}
+	})
+}
+
+// heldLog holds each line written to it until it is closed.
+type heldLog chan struct{}
+
+// Write waits until the log is closed.
+func (l heldLog) Write(p []byte) (int, error) {
+	<-l
+
+	return len(p), nil
+}
+
+// stopHeldSender starts sending to an endpoint the sender cannot connect
+// to, holds the sender as it logs that failure until release has passed,
+// stops the sending, and returns how long the stop took. It returns once
+// the sender has ended, as a bubble must.
+func stopHeldSender(t *testing.T, release time.Duration) time.Duration {
+	t.Helper()
+
+	log := make(heldLog)
+	sender := remotelog.New(remotelog.Params{
+		// No port is 65536, so each attempt to connect fails at once,
+		// before it reaches the network.
+		URL:    &url.URL{Scheme: remotelog.SchemeTCP, Host: "127.0.0.1:65536"},
+		Buffer: 1,
+	})
+
+	stopSending := startSending(t.Context(), sender,
+		slog.New(slog.NewJSONHandler(log, nil)))
+
+	synctest.Wait()
+	time.AfterFunc(release, func() { close(log) })
+
+	stopped := time.Now()
+
+	stopSending()
+
+	took := time.Since(stopped)
+
+	time.Sleep(release)
+	synctest.Wait()
+
+	return took
+}